本文承接系列（一）。上一篇的目标是做出一个安全、可理解的最小 Agent：模型决定是否调用工具，工具结果回到模型，直到它给出回答。这一篇只把这条循环补成更可靠的版本，不会假装已经实现了完整的生产系统。

这一篇解决什么问题

最小循环能跑起来，但它仍有几个会在真实使用中暴露的问题：

• 模型调用工具后，如果漏存 assistant 消息或绑错工具结果，下一轮上下文就不完整。
• 直接把内部调试字段塞进模型请求，可能导致接口报错，也不利于排查问题。
• 模型陷入反复读文件、反复改同一段代码时，没有上限就会持续消耗时间和调用费用。
• 循环结束时只返回一段文本，调用方无法区分“任务完成”和“预算耗尽”。

所以本篇只加入三条工程化规则：消息必须配对、历史与请求要分开、每次任务都有明确预算和退出状态。

这里的代码是我在系列（一）基础上继续实现的教学版本。它参考了 Hermes Agent 的循环设计 与 learn-hermes-agent 教程项目，但不把两者等同：前者是持续演进的真实项目，后者是选择性讲解其机制的学习项目。

先看循环的新边界

  flowchart TD
  A["用户输入"] --> B["写入 messages 历史"]
  B --> C["构造干净的 API 请求"]
  C --> D["调用模型"]
  D --> E["先保存 assistant 消息"]
  E --> F{"是否有 tool_calls?"}
  F -->|"否"| G["返回 completed"]
  F -->|"是"| H["受限工具执行"]
  H --> I["每个结果绑定 tool_call_id"]
  I --> J{"达到回合预算?"}
  J -->|"否"| C
  J -->|"是"| K["返回 budget_exhausted"]
  

和上一篇相比，工具权限没有变宽。仍然只使用上一篇实现的受限 workspace/ 文件工具和固定测试命令。消息边界、预算和状态管理解决的是可靠性，不是权限控制。

规则一：assistant 消息和工具结果必须成对写回

一次模型回复若包含工具调用，历史至少要保留两类信息：

1. assistant 消息，记录模型发起了哪些调用。
2. tool 消息，逐条用 tool_call_id 指回对应调用。

尤其是模型一次发起多个工具调用时，顺序相同并不足以保证匹配正确。tool_call_id 才是稳定的关联键。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 仅示意配对顺序：先存模型的调用意图，再存每个调用的结果
# （完整版见下方 agent.py，那里会把 tool_calls 转成纯 dict，原因稍后解释）
messages.append({
    "role": "assistant",
    "content": assistant.content,
    "tool_calls": assistant.tool_calls,
})

for tool_call in assistant.tool_calls:
    output = await execute_tool(
        tool_call.function.name,
        tool_call.function.arguments,
    )
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "name": tool_call.function.name,
        "content": output,
    })

先看看不配对会怎样坏。假设你偷懒，跳过 assistant 消息、直接把工具结果 append 进历史，下一轮请求时 OpenAI 接口会直接拒绝，报出类似这样的错误：

1
2

BadRequestError: messages with role 'tool' must be a response to a
preceding message with 'tool_calls'.

也就是说，在本文采用的 OpenAI-compatible 消息协议中，一条 role=tool 的消息必须紧跟在一条带 tool_calls 的 assistant 消息之后，这不是「最好这样」，而是「不这样就报错」。反过来，如果你存了 assistant 却漏掉某个 tool_call 的结果，接口会抱怨「有 tool_call 没有对应的 tool 响应」。Anthropic 原生 Messages API 的 tool_use / tool_result 是另一种消息形状；它同样要求调用意图与结果完整对应，但不能把本节的字段名原样搬过去。

所以这条规则的两半各自兜住一种崩溃：漏掉 assistant 消息，模型下一轮既看不到「是谁、为什么发起调用」，接口也会报错；漏掉 tool_call_id，多工具调用时结果归属错乱，模型可能把 A 工具的结果当成 B 的。顺序相同不等于配对正确，tool_call_id 才是稳定的关联键。

规则二：历史记录与 API 请求分开

messages 是我们保存、恢复和调试一段任务时用的历史。向模型发送的请求则应该是从历史中构造出的兼容副本。

为什么要多此一举?先看看历史「膨胀」之后的样子。上一篇的历史里，每条消息只有模型协议认识的字段。但真实应用几乎一定会想在历史里多记点东西：这条回复来自哪次请求、花了多少毫秒、当时用的哪个模型。于是一条 assistant 历史记录很可能长成这样：

1
2
3
4
5
6
7
8
9
10

# 应用历史里的一条消息：协议字段 + 你自己加的调试字段
{
    "role": "assistant",
    "content": "我先读一下 README。",
    "tool_calls": [...],
    "trace_id": "t-0007",      # ← 模型接口不认识
    "elapsed_ms": 820,         # ← 模型接口不认识
    "model": "gpt-4o-mini",    # ← 模型接口不认识
    "timestamp": "2026-07-16T20:31:05Z",
}

这些多出来的字段对你很有用(排查慢在哪、复现某次对话)，但如果原样发给模型接口，轻则被忽略，重则某些提供商直接报「未知字段」。build_api_messages() 做的就是每次请求前把历史「过滤」一遍，只留协议认识的字段：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

SYSTEM_PROMPT = (
    "你是编程助手。只能使用提供的工具处理 workspace 内的文件。"
    "修改前先阅读相关文件；测试失败时如实说明结果。"
)

API_FIELDS = ("role", "content", "tool_calls", "tool_call_id", "name")


def build_api_messages(messages: list[dict]) -> list[dict]:
    """从应用历史中挑出模型接口允许的字段。"""
    api_messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    for message in messages:
        clean_message = {
            key: message[key]
            for key in API_FIELDS
            if key in message and message[key] is not None
        }
        api_messages.append(clean_message)

    return api_messages

上面那条带 trace_id、elapsed_ms 的历史，经过 build_api_messages() 后就只剩 role/content/tool_calls，调试信息留在了你的历史里，发给模型的是一份干净副本。历史服务于应用，请求服务于模型协议，这条边界现在就具体了：它不是所有 SDK 都强制的「两份数据结构」，而是你主动划出的一道界线，让「你想记什么」和「协议允许发什么」互不干扰。

再注意这里和上一篇的一处改动。上一篇把系统提示直接放进了 messages[0]，跟着历史一起走；这一篇把它移出历史，你注意到 build_api_messages() 第一行就是把 SYSTEM_PROMPT 硬拼到最前面，而不是从 messages 里读。为什么改?因为系统提示是「应用配置」，不是「对话发生过的事」，把它和用户、工具的真实对话混在一起，之后做提示词版本管理、上下文压缩、会话恢复时，三类数据就会缠成一团。所以从这一篇起，你的 messages 里不再包含 system 那一条，它由 build_api_messages() 负责在每次请求时补上。Hermes 的提示词组装文档也是这样区分「稳定的系统提示词」与「请求时才附加的内容」。

规则三：用回合预算和结构化状态结束循环

先说清「一轮」是什么。循环里的一个 turn，指的是一次完整的模型往返：发一次请求、拿一次回复。如果这次回复要求调工具，那这一轮里可能执行了好几个工具，但它们只算一轮，预算数的是「问了模型几次」，因为模型请求才是最花时间和钱的部分。所以 MAX_TURNS = 10 的意思是：一次用户任务里，最多允许模型来回思考 10 次。

预算首先是单次任务的保险丝。到达上限后，不让 Agent 默默继续尝试，而是把控制权交回调用方。模型可能因为工具反复报错、目标不清而陷进「读文件—改代码—再读—再改」的循环里；没有预算，它会一直烧你的时间和调用费，而你甚至不知道它是不是卡住了。

光有上限还不够，还要让调用方知道循环是「正常说完了」还是「被预算掐断的」。上一篇的循环结束时只返回一段文本，调用方无从区分这两种情况。所以这一篇引入一个结构化的返回值 AgentResult：

1
2
3
4
5
6

@dataclass
class AgentResult:
    status: Literal["completed", "budget_exhausted"]
    reply: str
    turns: int
    messages: list[dict]

两个可能陌生的语法：@dataclass 是 Python 的装饰器，省掉手写 __init__，你直接写字段，它自动生成构造函数;Literal["completed", "budget_exhausted"] 是类型提示，声明 status 只应是这两个字符串之一(它只帮 IDE 和类型检查器提示，运行时并不会拦住你传别的值，所以真正的约束还得靠代码本身只返回这两种)。

把「模型最终说的话」(reply)与「循环为什么结束」(status)分开返回，CLI、网页或未来的网关才能采取不同动作：completed 就正常展示回复;budget_exhausted 则提示用户任务未确认完成、并借 messages 保存现场供后续诊断。下面完整实现里会看到这两个分支分别在哪里返回。

完整的 agent.py

下面的实现直接接续上一篇。假设你已经有：

• llm.py 中的 call_llm(messages, tools=...)；
• tools.py 中的 tool_schemas() 和受限的 execute_tool(name, raw_arguments)；
• 上一篇 export 好的模型 API Key 等环境变量。

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

from dataclasses import dataclass
from typing import Literal

from llm import call_llm
from tools import execute_tool, tool_schemas


MAX_TURNS = 10
SYSTEM_PROMPT = (
    "你是编程助手。只能使用提供的工具处理 workspace 内的文件。"
    "修改前先阅读相关文件；测试失败时如实说明结果。"
)
API_FIELDS = ("role", "content", "tool_calls", "tool_call_id", "name")


@dataclass
class AgentResult:
    status: Literal["completed", "budget_exhausted"]
    reply: str
    turns: int
    messages: list[dict]


def assistant_message_for_history(message) -> dict:
    """保留下一轮调用需要的标准 tool_calls 结构。"""
    item = {"role": "assistant", "content": message.content}
    tool_calls = []

    for tool_call in message.tool_calls or []:
        tool_calls.append({
            "id": tool_call.id,
            "type": "function",
            "function": {
                "name": tool_call.function.name,
                "arguments": tool_call.function.arguments,
            },
        })

    if tool_calls:
        item["tool_calls"] = tool_calls
    return item


def build_api_messages(messages: list[dict]) -> list[dict]:
    api_messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    for message in messages:
        clean_message = {
            key: message[key]
            for key in API_FIELDS
            if key in message and message[key] is not None
        }
        api_messages.append(clean_message)

    return api_messages


async def run_agent(messages: list[dict], user_input: str) -> AgentResult:
    messages.append({"role": "user", "content": user_input})

    for turn in range(1, MAX_TURNS + 1):
        response = await call_llm(
            build_api_messages(messages),
            tools=tool_schemas(),
        )
        messages.append(assistant_message_for_history(response))

        tool_calls = response.tool_calls or []
        if not tool_calls:
            return AgentResult(
                status="completed",
                reply=response.content or "模型没有返回文本回复。",
                turns=turn,
                messages=messages,
            )

        for tool_call in tool_calls:
            output = await execute_tool(
                tool_call.function.name,
                tool_call.function.arguments,
            )
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": tool_call.function.name,
                "content": output,
            })

    return AgentResult(
        status="budget_exhausted",
        reply=(
            f"本次任务已达到 {MAX_TURNS} 轮预算，已停止继续调用工具。"
            "可以根据保存的 messages 继续诊断，或让用户补充更明确的目标。"
        ),
        turns=MAX_TURNS,
        messages=messages,
    )

这里有个细节值得停一下:assistant_message_for_history() 为什么要把 message.tool_calls 一个个拆开、手动拼成普通 dict,而不是像规则一的示意片段那样直接 "tool_calls": message.tool_calls?因为 call_llm 返回的 message.tool_calls 是 LiteLLM 的 pydantic 对象,不是普通字典。留在内存里直接用没问题,但它不能干净地序列化,而下一篇要把 messages 存进 SQLite,那时能不能 json.dumps 就成了硬要求。现在就转成纯 dict,历史才真正是「一份你能保存、能重新加载的数据」,这正是规则二说的「历史服务于应用」落到实处的地方。这也是规则一那段特意标注「完整版会转成 dict」的原因。

再补一条：给消息明确“谁可以改”的所有权

上面的教学代码故意把 messages 作为可变列表传入并原地追加，目的是让数据流直观；这不是可以悄悄带进运行时的默认约定。只要 UI 订阅者、重试逻辑、会话存储和子 Agent 都握着同一个列表或同一层嵌套对象，任何一方改动 tool_calls、参数或结果，都可能污染其他观察者看到的历史。

《动手学 Pi》的 Loop 对此采用了一个值得复用的契约：调用时先深拷贝输入上下文，运行期间只修改自己的局部 transcript；对外发布状态、返回结果或恢复会话时再交付独立副本。这里的关键不是 deepcopy 这个函数名，而是下面三条边界：

• 调用者交给 run 的历史不应在它不知情时被改写；若选择原地更新，必须把它声明为 API 契约。
• 运行中的列表不能直接借给订阅回调或缓存层；它们拿到的是 snapshot，而不是可回写的 live reference。
• 只复制最外层列表不够。tool_calls、函数参数和工具结果同样是嵌套对象，需要随快照一起复制。

本系列下一篇做 JSONL 会话存储时，应把这条约束落实成测试：修改 run_agent() 的返回值，不能反向改掉保存中的会话；修改 Store 读出的对象，也不能影响下一轮模型请求。否则“历史能序列化”依然不等于“历史可靠”。

再补一条：结束原因不等于最终文本

本篇完整代码只实现了 completed 和 budget_exhausted，这正好够把“模型答完了”和“系统主动停下”分开。不要为了看起来更“生产级”而在这一章假装已经处理所有故障；但从现在起，调用方应把下面这些结束原因当成不同的产品状态，而不是都显示成一段普通回复：

后四种会在接入超时、审批和交互运行时后才有对应实现。先写下这张表的价值在于：持久化与 UI 层从一开始就知道，reply 是给人看的文字，status 才是系统为什么停止的事实。

在入口中，调用方可以显式处理这两种状态：

1
2
3
4
5

result = await run_agent(messages, user_input)
print(result.reply)

if result.status == "budget_exhausted":
    print("提示：本轮任务未确认完成，历史已保留，可继续排查。")

想亲眼看到 budget_exhausted 触发,最简单的办法是把 MAX_TURNS 临时改成 2,再给一个需要好几步的任务(比如「读完所有文件、逐个总结、再汇总成一段」)。终端会看到类似:

1
2
3
4
5

你: 读取 workspace 下每个文件并逐一总结，最后合并成一段说明。
  调用工具: list_files
  调用工具: read_file
本次任务已达到 2 轮预算，已停止继续调用工具。可以根据保存的 messages 继续诊断，或让用户补充更明确的目标。
提示：本轮任务未确认完成，历史已保留，可继续排查。

注意最后模型并没有给出总结,它被预算掐断了,而 status 如实标成 budget_exhausted,调用方于是走了「提示未完成」的分支,而不是把半截结果当成功。这个示例没有把「超预算」伪装成「已完成」。这是比无限重试更诚实、也更容易被上层产品正确处理的接口。

关于 Hermes 的“90 轮”与子 Agent

Hermes 当前文档中，主 Agent 的 agent.max_turns 默认值是 90。这个数字可以理解为一次循环的默认上限，但不应据此推导出“所有子 Agent 共用父 Agent 的同一个钱包”。其文档说明子 Agent 有自己的预算，并通过 delegation.max_iterations 受到单独限制；父子 Agent 的总轮数可能超过父 Agent 的单独上限。

这也提醒我：不要把一个框架的某个实现细节，写成所有 Agent 都遵循的普遍规律。本文自己的 MAX_TURNS = 10 只是教学项目的一次请求预算，选小一点是为了便于观察、控制成本和调试。

跑不通时，先查这几处

这一篇引入了配对、过滤、预算三条规则，出问题也大多集中在这三处：

报 messages with role 'tool' must be a response to ... 之类的 400 错误。配对断了。回到规则一：每次模型要求调工具，必须先 append 那条 assistant 消息，再为每一个 tool_call 各 append 一条带对应 tool_call_id 的 tool 结果，一个都不能少。改动循环后最容易在这里破。

报 Object of type ... is not JSON serializable（多在下一篇存库时冒出来）。你多半是把 message.tool_calls（pydantic 对象）直接塞进了历史，而没走 assistant_message_for_history() 转成纯 dict。规则一的示意片段是简化版，真正入历史要用转换后的 dict。

模型接口报「未知字段」或行为诡异。你给历史加了自定义字段（trace_id、耗时等），但请求没经过 build_api_messages() 过滤，把这些字段原样发出去了。所有发往模型的请求都应该从 build_api_messages() 出，而不是直接把 messages 丢给 call_llm。

正常任务却总是 budget_exhausted。MAX_TURNS 设得太小，或系统提示没引导模型「拿到足够信息就收尾」，导致它反复读文件不肯总结。先把每轮的 messages 打印出来，看它是卡在重复调用，还是真的需要更多轮。

排查这类问题的通用手段还是那句话:打印 AgentResult。它把 status、turns、完整 messages 都带出来了，status 告诉你为什么停、turns 告诉你花了几轮、messages 让你逐条还原模型每一步看到了什么。

动手改改看

1. 给历史消息加一个 timestamp 字段（在 append user/assistant/tool 时都带上），然后打印 build_api_messages() 的输出，确认这个字段没有出现在发往模型的请求里。这是把规则二「亲手验证一遍」。
2. 把 MAX_TURNS 改成 run_agent 的一个参数，默认 10。想一想：什么样的任务值得给更大的预算，什么场景反而要调小到 2、3 来快速失败?
3. 给 AgentResult 再加一个 status 取值，比如 tool_error（当某个工具连续多次返回失败时提前退出）。这会逼你思考：循环除了「说完了」和「超预算」，还有哪些该被显式表达的结束方式?

这还不是完整的“生产级”

有了三条规则，循环更可靠了，但离生产环境仍有不少距离。下面这些能力不应只写成一句“以后加上”，而应在后续章节分别实现和验证：

特别是“异步桥接”和“Gateway 实例可重建”属于运行时与服务层设计。没有对应的事件循环、会话存储和并发测试，只在 Agent 循环里讲概念，会让初学者误以为复制一段代码就已经获得这些能力。因此我把它们留到后面，届时给出可验证的实现。

安全边界仍要保留

本篇没有新增任意命令执行能力，仍应沿用上一篇的限制：

• 读写路径必须解析后仍位于 workspace/ 内。
• 测试工具只能运行预先固定的 pytest -q，不能接受模型拼出的 shell 命令。
• 工具输出应截断，避免将大量内容无节制塞回上下文。

即便是固定测试，也可能执行项目中的测试代码。处理不可信仓库或不可信输入时，仍应在隔离环境中运行，并在接入删除、联网、部署等高影响工具前增加人工审批。

小结

• 可靠循环先保证消息配对：存 assistant，再按 tool_call_id 存 tool 结果。
• 让 messages 保存应用历史，让 build_api_messages() 构造兼容模型接口的请求。
• 用 MAX_TURNS 限制单次任务，并以 completed 或 budget_exhausted 返回明确状态。
• 预算、消息边界不能替代沙箱和审批。安全边界仍由受限工具实现。
• 子 Agent、异步桥接和 Gateway 是后续独立问题，不能被一句“生产级循环”覆盖。

下一篇先实现 fresh-context 子 Agent：把高输出但可独立完成的查找任务隔离到另一份历史中，并只把可核验的摘要回传给父 Agent。会话存储紧接在下一篇：先采用容易观察、逐条追加的 JSONL 记录，再讨论何时需要换成 SQLite。到那时，“历史”和“API 请求”分开的价值会变得更直观。

参考资料

• Hermes Agent Loop 开发文档
• Hermes Prompt Assembly 开发文档
• learn-hermes-agent 教学项目
• 消息输入所有权与 snapshot 边界的可运行示例：动手学 Pi · Checkpoint 07
• 本系列（一）：对话循环与工具调用