Spaces:
Sleeping
Sleeping
| """The tool loop. Blueprint 8.3: at most 5 iterations per turn, then force a text answer. | |
| Each model turn that calls tools is appended to the contents **verbatim** and followed by the | |
| function responses. Verbatim matters: newer models attach a thought signature to their content, | |
| and a re-serialised copy that dropped it would make the next call fail (blueprint 7.1). | |
| """ | |
| import logging | |
| from collections.abc import Sequence | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| from ..llm.base import Content, ToolCall, function_response_content | |
| from ..llm.errors import EmptyReply | |
| from ..persona.types import Mode | |
| from .registry import ToolContext, ToolRegistry | |
| log = logging.getLogger(__name__) | |
| MAX_ITERATIONS = 5 | |
| #: toolConfig that makes the forced final call text-only. | |
| NO_TOOLS = {"functionCallingConfig": {"mode": "NONE"}} | |
| class LoopResult: | |
| text: str | |
| #: Everything the model has seen up to (not including) its final text answer, tool results | |
| #: included. A guardrail retry replays exactly this, so tools are never executed twice. | |
| contents: list[Content] | |
| client_actions: list[dict[str, Any]] = field(default_factory=list) | |
| sources: list[dict[str, str]] = field(default_factory=list) | |
| artifacts: list[dict[str, Any]] = field(default_factory=list) | |
| #: One line per tool call, for the HUD activity log. | |
| tool_log: list[str] = field(default_factory=list) | |
| model: str = "" | |
| key_slot: int = 0 | |
| ms: int = 0 | |
| async def run_tool_loop( | |
| provider: Any, | |
| system_prompt: str, | |
| contents: Sequence[Content], | |
| registry: ToolRegistry | None, | |
| ctx: ToolContext | None, | |
| *, | |
| temperature: float, | |
| max_output_tokens: int, | |
| force_tool: str | None = None, | |
| ) -> LoopResult: | |
| history = list(contents) | |
| result = LoopResult(text="", contents=history) | |
| declarations = registry.declarations(ctx.mode if ctx else Mode.NORMAL) if registry else None | |
| tool_config = _force(force_tool, registry, ctx) | |
| total_ms = 0 | |
| for _ in range(MAX_ITERATIONS): | |
| turn = await provider.step( | |
| system_prompt, | |
| history, | |
| tools=declarations or None, | |
| tool_config=tool_config, | |
| temperature=temperature, | |
| max_output_tokens=max_output_tokens, | |
| ) | |
| total_ms += turn.ms | |
| result.model, result.key_slot = turn.model, turn.key_slot | |
| # Forcing a tool is for the first step only; after that the model decides. | |
| tool_config = None | |
| if not turn.calls: | |
| result.text = turn.text | |
| result.ms = total_ms | |
| if not turn.text.strip(): | |
| raise EmptyReply("The model ended the turn with no text") | |
| return result | |
| history.append(turn.content) | |
| responses: list[tuple[ToolCall, dict[str, Any]]] = [] | |
| for call in turn.calls: | |
| outcome = await registry.execute(call, ctx) if registry and ctx else None | |
| if outcome is None: | |
| responses.append((call, {"ok": False, "error": "tools are not available"})) | |
| continue | |
| responses.append((call, outcome.response)) | |
| result.client_actions += outcome.client_actions | |
| result.sources += outcome.sources | |
| result.artifacts += outcome.artifacts | |
| status = "ok" if outcome.response.get("ok") else outcome.response.get("error", "failed") | |
| result.tool_log.append(f"{call.name}: {status}") | |
| history.append(function_response_content(responses)) | |
| # Five rounds of tools and still no answer: make it answer with what it has. | |
| log.warning("Tool loop hit %d iterations; forcing a text answer", MAX_ITERATIONS) | |
| turn = await provider.step( | |
| system_prompt, | |
| history, | |
| tools=declarations or None, | |
| tool_config=NO_TOOLS if declarations else None, | |
| temperature=temperature, | |
| max_output_tokens=max_output_tokens, | |
| ) | |
| result.text = turn.text | |
| result.model, result.key_slot = turn.model, turn.key_slot | |
| result.ms = total_ms + turn.ms | |
| if not turn.text.strip(): | |
| raise EmptyReply("The model returned no text after the tool loop") | |
| return result | |
| def _force(tool: str | None, registry: ToolRegistry | None, ctx: ToolContext | None) -> dict | None: | |
| """Master prompt 6: the UI's Explain / Review / Debug buttons force one specific tool.""" | |
| if not tool or registry is None or ctx is None or tool not in registry.names(ctx.mode): | |
| return None | |
| return {"functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [tool]}} | |