agent.py

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