03 · 从零手写一个 Agent

🎯 难度：中级⏱ 预计：2 小时# 循环# 工具注册# 预算

这一课写一个可复用运行器，后面编程助手、研究员都复用它。先理解每一步，再复制完整代码。

五步循环

1. 把 messages + tools 发给模型
2. 保存模型返回的 assistant 消息
3. 没有 tool_calls → 返回答案
4. 有 tool_calls → 逐个验证并执行，追加 tool 消息
5. 回到第 1 步，直到完成或触达预算

不是递归魔法，就是 for 循环。参数错误当作工具反馈，让模型有机会修正；API 连接错误则向上抛，由应用决定如何提示。

准备

沿用 config.py 与 openai / pydantic / python-dotenv 依赖。模型必须支持工具调用。

创建以下完整文件：

import json
from collections.abc import Callable
from pydantic import BaseModel
from config import get_client, model_name


def dispatch(name: str, raw: str, handlers: dict) -> str:
    """只执行注册工具；给模型返回有限、可解析的结果。"""
    if name not in handlers:
        return json.dumps({"error": "未知工具"}, ensure_ascii=False)
    if len(raw) > 10_000:
        return json.dumps({"error": "参数过长"}, ensure_ascii=False)
    schema, function = handlers[name]
    try:
        args = schema.model_validate_json(raw)
        result = function(**args.model_dump())
        return json.dumps(result, ensure_ascii=False, allow_nan=False)
    except (ValueError, TypeError, OSError):
        # 不回传异常中的路径、密钥或其他潜在敏感细节
        return json.dumps({"error": "参数无效或工具执行失败，请检查参数"}, ensure_ascii=False)


def tool_spec(name: str, description: str, schema: type[BaseModel]) -> dict:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": schema.model_json_schema(),
        },
    }


def run_agent(
    question: str,
    tools: list[dict],
    handlers: dict,
    system: str = "你是中文助手。按需要使用工具，不要编造工具结果。",
    max_steps: int = 8,
    max_tool_calls: int = 16,
    on_event: Callable | None = None,
) -> str:
    client = get_client()
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": question},
    ]
    calls = 0
    try:
        for step in range(max_steps):
            response = client.chat.completions.create(
                model=model_name(), messages=messages, tools=tools,
                tool_choice="auto", max_tokens=1_000,
            )
            choice = response.choices[0]
            message = choice.message
            if choice.finish_reason not in ("stop", "tool_calls"):
                raise RuntimeError(f"模型输出未正常结束：{choice.finish_reason}")
            messages.append(message.model_dump(exclude_none=True))
            if on_event:
                on_event({"type": "model", "step": step + 1})
            if not message.tool_calls:
                return message.content or "模型未提供文本回答。"
            for call in message.tool_calls:
                calls += 1
                if calls > max_tool_calls:
                    raise RuntimeError("工具调用预算已用完")
                name = call.function.name
                if on_event:
                    on_event({"type": "tool", "name": name, "step": step + 1})
                result = dispatch(name, call.function.arguments, handlers)
                messages.append({
                    "role": "tool", "tool_call_id": call.id,
                    "content": result,
                })
        raise RuntimeError("模型调用预算已用完，请缩小任务")
    finally:
        client.close()

装上两只「手」

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
from runner import run_agent, tool_spec

class WeatherArgs(BaseModel):
    model_config = ConfigDict(extra="forbid")
    city: Literal["北京", "上海"]

class ConvertArgs(BaseModel):
    model_config = ConfigDict(extra="forbid")
    celsius: float = Field(ge=-100, le=100, allow_inf_nan=False)


def get_weather(city: str) -> dict:
    data = {"北京": 22, "上海": 26}
    return {"city": city, "temp_c": data[city], "demo": True}


def convert(celsius: float) -> dict:
    return {"fahrenheit": round(celsius * 9 / 5 + 32, 2)}

handlers = {
    "get_weather": (WeatherArgs, get_weather),
    "convert": (ConvertArgs, convert),
}
tools = [
    tool_spec("get_weather", "查询北京/上海的模拟天气，不是实时天气", WeatherArgs),
    tool_spec("convert", "把摄氏度换算为华氏度", ConvertArgs),
]

if __name__ == "__main__":
    answer = run_agent(
        "查北京的模拟温度并换算成华氏度。必须明确这是演示数据。",
        tools, handlers, on_event=lambda event: print("事件：", event),
    )
    print(answer)

uv run demo_agent.py

模型应调用天气工具，再调用换算工具，最终报告模拟数据 22°C / 71.6°F。具体措辞和调用次数可能不同；日志与最终数值才是验收证据。

为什么有两个预算？

max_steps 限制模型请求轮数；同一轮可能请求多个工具，因此还需要 max_tool_calls。本示例的 timeout 是网络超时，工具都很快；连接外部服务时，工具也要设置超时和输出长度限制。

max_tokens 只控制单次输出，不是总费用上限。生产中还需要总 Token、总耗时和用户额度限制。

练习与验收

• 问「你好」：允许不调工具直接回答。
• 请求不存在的城市：模型应说明限制，不能说查询到真实天气。
• 直接测试 dispatch("unknown", "{}", handlers)：拒绝未知工具。
• 直接测试非法 JSON、celsius=10000：返回错误，不执行函数。
• 把 max_steps 改成 1：需要多轮的任务应显式失败，不假装完成。

核心里程碑

到这里你已经理解 Agent 的最小运行机制。框架主要是帮你管理状态、路由、持久化和观测，而不是另一个神秘原理。