Agent 调度循环：工具调用、结果回灌与三个出口

上一篇把 Agent 拆成了模型、工具、记忆、循环四个组件。本文实现其中的循环，不依赖任何 Agent 框架，只用 openai SDK 和几十行 Python。

示例任务是查询「双11-美妆拉新 10 月 31 日的 ROI」。完成它需要两个工具：先按计划名查出 campaign_id，再按 ID 查询报表。报表服务使用本地模拟数据；真实项目中它对应广告平台的 Marketing API，这类接口都需要鉴权，没有免费的公开版本。

本文用到的术语如下，其中回灌、终答、终止、出口与上一篇的定义一致：

Table of contents

一、普通问答与 Agent 在代码上的差别

普通问答只请求一次，取出 content 即结束：

response = client.chat.completions.create(model=model, messages=messages)
print(response.choices[0].message.content)

变成 Agent，接口地址、模型和鉴权都不用改，只是在请求里加上 tools：

response = client.chat.completions.create(
    model=model,
    messages=messages,
    tools=registry.schemas,  # 工具 schema 列表，见第三节
    tool_choice="auto",
)

响应的结构没有变，但多了一种情况：模型可以不直接回答，而是在 message.tool_calls 里请求调用工具，此时 finish_reason 为 "tool_calls"。一条请求调用工具的 message 大致如下：

{
    "role": "assistant",
    "content": None,
    "tool_calls": [
        {
            "id": "call_abc123",
            "type": "function",
            "function": {
                "name": "search_campaign",
                "arguments": "{\"name\": \"双11-美妆拉新\"}",
            },
        }
    ],
}

这里有三点需要注意：

• tool_calls 是一个列表，一次响应可以包含多个调用，每个调用有自己的 id。
• arguments 是 JSON 字符串而不是对象，需要代码自己解析，而且不保证一定是合法 JSON。
• 请求工具时 content 通常为 None，但有些模型会同时返回一段说明文字，例如「我先查一下这个计划」。

因此，判断本轮是否结束要看 tool_calls 是否存在，而不是 content 是否有值。如果代码只读取 content，要么拿到 None，要么把这段说明文字当成最终回答，模型的工具请求都会被丢掉。

表格最后两行是本文的重点：错误处理和退出条件都由循环代码负责，模型不会主动停下来。如果代码不设上限，模型反复请求同一个工具，循环就会一直请求下去，每一轮都产生费用。

二、循环主体与三个关键细节

下面是循环的主体 run_agent，取自源码 agent.py，省略了日志。工具注册表 registry 和它的 dispatch 方法留到第三节，出口三的 token 预算留到第六节。

def run_agent(
    client: OpenAI,
    registry: ToolRegistry,
    *,
    model: str,
    system_prompt: str,
    goal: str,
    max_turns: int = 8,
    token_budget: int = 20000,
    history: list[dict] | None = None,
) -> RunResult:
    # 传入 history 时在原有 messages 上继续（例如用户确认之后），此时 system_prompt 不再重复写入
    messages: list[dict] = list(history) if history else [{"role": "system", "content": system_prompt}]
    messages.append({"role": "user", "content": goal})
    total_tokens = 0

    for turn in range(1, max_turns + 1):
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            tools=registry.schemas,
            tool_choice="auto",
        )
        if response.usage is not None:
            total_tokens += response.usage.total_tokens
        message = response.choices[0].message

        # 出口一：本轮没有工具调用，content 就是终答
        if not message.tool_calls:
            messages.append(message.model_dump(exclude_none=True))
            return RunResult(message.content or "", "final_answer", turn, total_tokens, messages)

        # 出口三：模型还要调用工具，但累计 token 已超出预算
        if total_tokens > token_budget:
            return RunResult("累计 token 超出预算，已终止。", "token_budget", turn, total_tokens, messages)

        # 带 tool_calls 的 assistant 消息必须先写入 messages
        messages.append(message.model_dump(exclude_none=True))

        for tool_call in message.tool_calls:
            result = registry.dispatch(tool_call.function.name, tool_call.function.arguments)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(result, ensure_ascii=False),
                }
            )

    # 出口二：达到轮数上限
    return RunResult("达到最大轮数仍未得到终答，已终止。", "max_turns", max_turns, total_tokens, messages)

* 之后的参数只能按名字传入。model、system_prompt、goal 都是字符串，max_turns 和 token_budget 都是整数，按位置传很容易写反，而且不会报错。

返回值 RunResult 记录终答、从哪个出口退出（final_answer、max_turns 或 token_budget）、轮数、累计 token 和完整的 messages。出口一返回前会把终答也写入 messages，这份记录因此是完整的：调用方据此区分正常结束和被迫终止，「评测」一篇的轨迹检查直接使用它，下一篇的追投确认则把它作为 history 传回，在同一份上下文上继续。

整个过程没有隐藏状态。模型每一轮看到什么、做过什么，全部记录在 messages 这个列表里。

这段代码里有三处细节，漏掉任何一处都会出问题：

第一，带 tool_calls 的 assistant 消息必须原样写入 messages。 tool 消息必须跟在发起调用的那条 assistant 消息之后。如果漏写这一条，下一轮请求里的 tool 消息就找不到对应的调用，OpenAI 兼容接口通常直接返回 400。

第二，每个 tool_call_id 都要有一条对应的 tool 消息。 模型可能在一轮里同时请求多个工具。这些工具可以串行执行，也可以并发执行，但每个调用都必须回灌一条 tool 消息，且 tool_call_id 一一对应。少回灌一条，接口同样返回 400，而不是忽略缺失的部分。

第三，循环用 for turn in range(1, max_turns + 1)，而不是 while True。 模型可能反复以相同参数调用同一个工具（即打转）。没有轮数上限，这种情况下循环不会结束，费用会一直累积。

三、工具：schema 与 dispatch

每个工具分为两部分：提供给模型的 schema，和真正执行的 Python 函数。模型只能看到 schema，决定何时调用、参数怎么填；函数实现对模型完全不可见。源码用一个 Tool 把两者放在一起，避免 schema 和函数各改各的：

@dataclass(frozen=True)
class Tool:
    name: str
    description: str
    parameters: dict[str, Any]  # JSON Schema
    fn: Callable[..., dict]

    @property
    def schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

本例的两个工具有先后依赖：报表接口需要 campaign_id，而用户只会说计划名称。模型必须先调用 search_campaign 拿到 ID，再调用 get_campaign_report。

SEARCH_CAMPAIGN_TOOL = Tool(
    name="search_campaign",
    description=(
        "按计划名搜索广告计划，返回 campaign_id。"
        "只有计划名、没有 campaign_id 时，先调用此工具。"
    ),
    parameters={...},  # 省略
    fn=search_campaign,
)

GET_CAMPAIGN_REPORT_TOOL = Tool(
    name="get_campaign_report",
    description=(
        "按 campaign_id 和日期查询计划日报表（ROI、消耗、成交额等）。"
        "没有 campaign_id 时，先调用 search_campaign。"
        "报表 T+1 产出，只能查询昨天及更早的日期。"
    ),
    parameters={...},  # 省略
    fn=get_campaign_report,
)

get_campaign_report 描述里的第二句「没有 campaign_id 时，先调用 search_campaign」不是给开发者看的注释，而是模型安排调用顺序的依据。删掉这句，模型可能跳过搜索，直接编一个形如 CP-xxxx 的 ID 去查报表。如果这个 ID 恰好对应账户里的另一个计划，接口不会报错，模型会拿别的计划的 ROI 给出一个看似正常的回答。schema 的写法在下一篇专门讨论。

工具注册进 ToolRegistry，循环通过它拿到全部 schema，并用 dispatch 把模型的调用请求转成函数调用。模型给出的 arguments 是一段字符串，不能保证是合法 JSON，也不能保证参数名与函数签名一致，必须当作不可信输入处理：

class ToolRegistry:
    def __init__(self, tools: Iterable[Tool]):
        self._tools_by_name = {tool.name: tool for tool in tools}

    @property
    def schemas(self) -> list[dict[str, Any]]:
        return [tool.schema for tool in self._tools_by_name.values()]

    def dispatch(self, name: str, raw_arguments: str) -> dict:
        tool = self._tools_by_name.get(name)
        if tool is None:
            return {"error": f"未知工具 {name}，可用工具：{sorted(self._tools_by_name)}"}

        try:
            arguments = json.loads(raw_arguments or "{}")
        except json.JSONDecodeError:
            return {"error": f"arguments 不是合法 JSON：{raw_arguments[:200]}"}
        if not isinstance(arguments, dict):
            return {"error": "arguments 必须是 JSON 对象"}

        try:
            inspect.signature(tool.fn).bind(**arguments)
        except TypeError as e:
            return {"error": f"参数与工具定义不符：{e}"}

        type_error = find_type_error(tool.parameters, arguments)
        if type_error is not None:
            return {"error": type_error}

        return tool.fn(**arguments)

dispatch 处理四类参数问题：工具名不存在、arguments 不是合法的 JSON 对象、参数名或个数与函数签名不符、参数类型与 schema 中的 type 不符。它们都以返回值的形式交给循环回灌，让模型下一轮修正。模型只能调用注册过的工具，这也是一道白名单。

参数检查用 inspect.signature(tool.fn).bind(**arguments) 单独完成，而不是把 tool.fn(**arguments) 整个包进 try ... except TypeError。原因是工具函数内部的 bug 也可能抛出 TypeError，一并捕获就会把代码错误当成参数错误回灌给模型。tool.fn(**arguments) 内部抛出的异常不在这里处理，交给第五节的终止逻辑。

类型检查由 find_type_error 完成，它按 schema 里每个参数的 type 检查对应的值，例如 "name": ["双11-美妆拉新"] 会被拒绝。schema 不能保证模型遵守，工具函数却通常默认参数是字符串，没有这一步，一个数组就会让 dict.get 抛出 TypeError，整个 run 因此中断。

四、messages 如何逐轮增长

Chat Completions 接口本身不保存会话状态，每次请求都要把完整的 messages 发过去。messages 因此是 Agent 唯一的状态来源：只追加，不修改。四种角色各有分工：system 给出全局规则，user 给出目标，assistant 给出回答或工具调用请求，tool 返回工具执行结果。

以「双11-美妆拉新 10 月 31 日的 ROI 怎么样」为例，一次执行分三轮：

1. 第 1 轮。 上下文里只有计划名，模型请求 search_campaign(name="双11-美妆拉新")。代码执行后回灌 {"campaign_id": "CP-1024", "name": "双11-美妆拉新", "status": "active"}。
2. 第 2 轮。 模型从上一条回灌中取出 campaign_id，请求 get_campaign_report(campaign_id="CP-1024", date="2026-10-31")。代码回灌报表数据，包括 ROI、消耗、成交额、展示和点击。
3. 第 3 轮。 信息已经齐全，模型不再请求工具，直接给出回答。tool_calls 为空，循环从出口一退出。

第 2 轮能把「10 月 31 日」补全为 2026-10-31，是因为 system prompt 开头写了「今天是 2026-11-01」。模型不知道当前日期，只能按训练数据推断年份；去掉这句后实测，模型请求的是 2024-10-31，工具返回「没有投放数据」。凡是涉及「昨天」「10 月 31 日」这类相对日期的 Agent，都要在 prompt 里给出当前日期。

这个过程有两个代价，后续文章会分别处理：

• 累计 token 随轮数平方增长。 每轮都要重发完整的 messages。如果每轮新增的内容大致相同，N 轮的累计输入 token 约与 N² 成正比，而不是 N。
• 上下文最终会超出窗口。 messages 只增不减，迟早会超出模型的上下文长度。工具返回的 JSON 字段越多，这一天来得越早。字段裁剪和上下文压缩在「上下文与记忆」一篇讨论。

五、错误处理：哪些回灌，哪些终止

工具执行失败时，常见的做法是直接 raise，但这样整个循环会随之中断。而很多失败模型自己就能修正，例如参数格式错误、计划名拼错、参数名写错。这类情况下，工具应当返回一条说明问题的错误信息，由循环照常回灌：

def search_campaign(name: str) -> dict:
    campaigns_by_name = load_mock_data().campaigns_by_name
    campaign = campaigns_by_name.get(name)
    if campaign is None:
        similar_names = difflib.get_close_matches(name, campaigns_by_name.keys(), n=3)
        return {"error": f"未找到计划「{name}」", "similar": similar_names}
    return {"campaign_id": campaign["campaign_id"], "name": name, "status": campaign["status"]}

用户把计划名写成「双11-美装拉新」时，模型会收到 {"error": "未找到计划「双11-美装拉新」", "similar": ["双11-美妆拉新"]}，下一轮要么改用建议的名称重新调用，要么停下来向用户确认是否指「双11-美妆拉新」。两种做法都基于错误里的 similar 字段。错误信息写得越具体，模型越容易修正；只返回 {"error": "invalid"}，模型无从判断该改什么。

但不是所有错误都应该回灌。判断标准只有一个：模型能否通过修改调用参数解决这个错误？

最后一行尤其要注意。如果用 except Exception 把所有异常都回灌给模型，循环看起来永远不会出错，模型总能给出一个通顺的回答，但代码里的 KeyError 之类的 bug 会被掩盖，日志里也看不到。原则是：参数问题交给模型修正，代码问题通过告警交给工程师修复。

终止的实现方式是让这类错误以异常的形式离开循环。鉴权失败、配额耗尽这类可预期的系统错误，由工具抛出专门的异常类型；代码 bug 则不做任何捕获，让它自然抛出。接入真实报表接口时，写法如下：

class FatalToolError(Exception):
    """改参数也无法恢复的错误，例如鉴权失败、配额耗尽。"""


def get_campaign_report(campaign_id: str, date: str) -> dict:
    response = report_api.get(campaign_id, date)
    if response.status == 401:
        raise FatalToolError("报表服务鉴权失败")
    ...

源码使用模拟数据，没有真实的 401。在 .env 中设置 SIMULATE_REPORT_AUTH_EXPIRED=1，get_campaign_report 会直接抛出 FatalToolError，可以观察这条终止路径。

FatalToolError 在 run_agent() 外层捕获，源码里是命令行入口 __main__.py：

try:
    result = run_agent(client, registry, model=model, system_prompt=system_prompt, goal=goal)
except FatalToolError as e:
    logger.error("run 终止：%s", e)
    print(f"\n无法完成：{e}")
    return 1

记录日志后给用户一个明确的失败说明；其他未预期的异常不在这里捕获，直接进入告警。

表中的指数退避是工具内部对 429 和 5xx 的重试，与模型在循环中根据错误修改参数的重试是两回事。工具超时的处理在「部署与并发」一篇讨论。

六、三个出口

第二节的 run_agent 有三个出口，分别对应 RunResult.exit_reason 的三个取值：

轮数上限针对的是打转：模型以相同参数反复调用同一个工具，每次得到相同的结果，却一直不给出终答。在工具执行前识别并拦截打转，是「可靠性与成本」一篇的内容；在最基础的实现里，先保证循环一定能停下来。

累计 token 预算要按整个 run 的所有请求求和，不能只看单轮。每轮都重发完整的 messages，即使每一轮都不大，累计值也会随轮数平方增长。实现上只需在每轮请求后累加 usage，即第二节代码中的这几行：

total_tokens = 0
for turn in range(1, max_turns + 1):
    response = client.chat.completions.create(...)
    if response.usage is not None:
        total_tokens += response.usage.total_tokens
    message = response.choices[0].message

    if not message.tool_calls:
        ...
        return RunResult(message.content or "", "final_answer", ...)

    if total_tokens > token_budget:
        return RunResult("累计 token 超出预算，已终止。", "token_budget", ...)
    ...

预算检查放在出口一之后、执行工具之前：如果这一轮已经拿到终答，费用已经产生，直接返回终答；如果模型还要调用工具，超出预算就不再继续。usage.total_tokens 包含本轮的输入和输出，超出预算的那一轮已经计费，所以预算应留出一轮的余量。部分兼容 OpenAI 协议的服务不返回 usage，这时需要用 tokenizer 自行估算。

七、源码与调试

完整代码见 agent-in-action/ad-boost：

cp .env.example .env              # 填写 OPENAI_API_KEY
uv run python -m ad_boost          # 默认问题：双11-美妆拉新 10 月 31 日的 ROI 怎么样？

调试时，先看每一轮输出的 [turn ...] 日志：本轮调用了哪个工具、传了什么参数、回灌了什么结果。结果不对时，先从调用记录里找出是哪一轮出了问题，再去看终答。后续「评测」一篇的轨迹检查就建立在这份调用记录之上。

八、常见问题排查

小结

Agent 的核心是一个带出口的循环：请求模型，执行模型要求的工具，把结果回灌，再请求模型。模型只声明要调用哪个工具；执行、校验、回灌和退出判断都由代码完成。

实现时有三条规则必须遵守：带 tool_calls 的 assistant 消息要原样写入 messages，且每个调用都要回灌一条配对的 tool 消息；循环必须同时具备三个出口，即没有 tool_calls、轮数上限、累计 token 预算；模型能通过修改参数解决的错误回灌，其余错误终止。

循环跑通之后，效果好坏主要取决于 schema：description 决定模型能否在正确的时机调用工具，校验和回灌策略决定出错后能否恢复。下一篇以广告组追投为例，讨论 schema 的写法。