Spaces:
Running
Running
File size: 4,592 Bytes
210ef30 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """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"}}
@dataclass
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]}}
|