File size: 3,746 Bytes
9113b4c | 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 | from __future__ import annotations
from dataclasses import dataclass
from model import ModelRunner
from parsing import parse_response
from tools import ToolRegistry
import time
@dataclass
class Agent:
runner: ModelRunner
tools: ToolRegistry
system_prompt: str
def run(
self,
task: str,
max_steps: int = 5,
temperature: float = 0.3,
verbose: bool = False,
) -> tuple[str | None, list[dict]]:
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": task},
]
trace: list[dict] = []
for step in range(1, max_steps + 1):
time.sleep(1.5)
response = self.runner.generate(messages, temperature=temperature)
if verbose:
print(f"[step {step}] {response}")
messages.append({"role": "assistant", "content": response})
kind, payload = parse_response(response)
if kind == "final":
tool_succeeded = any(
entry.get("type") == "action" for entry in trace
)
if not tool_succeeded:
trace.append({
"step": step,
"type": "blocked_final",
"content": payload,
})
messages.append({
"role": "user",
"content": (
"You cannot give a FINAL answer without first calling a tool. "
"Call the appropriate tool first before answering."
),
})
continue
trace.append({"step": step, "type": "final", "content": payload})
return payload, trace
if kind == "action":
name, args = payload
try:
result = self.tools.execute(name, *args)
result_text = (
f"${float(result):.2f}"
if isinstance(result, (int, float))
else str(result)
)
observation = f"Tool result: {result_text}"
trace.append({
"step": step,
"type": "action",
"tool": name,
"args": args,
"result": result_text,
})
except KeyError as error:
observation = f"Error: {error}"
trace.append({
"step": step,
"type": "error",
"tool": name,
"args": args,
"error": str(error),
})
except Exception as error:
observation = f"Error: {error}"
trace.append({
"step": step,
"type": "error",
"tool": name,
"args": args,
"error": str(error),
})
messages.append({"role": "user", "content": observation})
else:
trace.append({"step": step, "type": "unknown", "raw": payload})
messages.append({
"role": "user",
"content": (
"Invalid format. Respond with exactly one line starting with "
"ACTION: tool_name(ARGS) or FINAL: <answer>."
),
})
return None, trace
|