Stage 01:把 Tool Loop 变成 Agent Runtime
Language: English | 简体中文
上一章我们故意停在了一个有点别扭的位置:程序已经能让模型请求一次 Tool,Python 也能执行,再把结果送回模型,但整个流程还是写死的。
这就像你请了一个助理,第一天的工作流程是:先去档案室查资料,回来以后做一次计算,然后向你汇报。你当然可以把三步写成三行固定流程。问题是第二天任务变了:有时根本不用查资料,有时要连续查两次,有时查完以后才知道下一步该做什么。你总不能每天早上先猜助理今天会走几步,再准备 first_response、second_response、third_response。
所以这章真正要解决的,不是“再加一个工具”,而是:
当模型每一轮都可能决定下一步时,应用怎样把这种不确定性装进一个可控的循环里?
这个循环就是 Agent Runtime 的核心。
1. 先看看固定脚本哪里开始变笨
Stage 00 的工具示例,本质上是这样:
first = call_model(user_request)
call = read_tool_call(first)
result = execute(call)
final = call_model(result)
return final.output_text
对于“一次工具调用就结束”的任务,这段代码完全没问题。甚至我会说,它比一上来就套一个庞大框架更好懂。
麻烦出现在任务路径不固定的时候。
比如用户说:
读取东京的教学天气,并把摄氏度换成华氏度。
模型可能先请求天气 Tool,拿到 18.0°C 后再请求温度换算 Tool,最后才给答案:
user
↓
model: get_teaching_weather("Tokyo")
↓
application: 18.0°C, cloudy
↓
model: celsius_to_fahrenheit(18.0)
↓
application: 64.4°F
↓
model: final answer
也可能遇到另一个任务,模型第一轮就直接回答,根本不需要 Tool。
你会发现,程序真正知道的不是“总共有三轮”,而是一个重复规则:每一轮先让模型决定下一步;如果它要调用工具,执行完再继续;如果它给出最终答案,就结束。
把固定脚本和 Runtime Loop 并排看,这个变化会更直观:
再把右侧的循环压缩成伪代码:
for step in range(max_steps):
turn = model.generate(messages, tools)
if turn.final_text is not None:
return turn.final_text
for call in turn.tool_calls:
observation = execute(call)
messages.append(observation)
raise MaxStepsExceeded
这几行就是本章的主角。后面的所有类——ToolCall、ModelTurn、ToolRegistry、Adapter——都是为了让这段循环的职责更清楚,而不是为了把简单事情写复杂。
2. Agent 和普通 Workflow 到底差在哪
在继续写代码之前,我们先解决一个很容易越学越糊的问题:是不是只要程序里用了 LLM,就叫 Agent?
不是。
看一个普通的确定性流程:
weather = get_weather("Tokyo")
fahrenheit = celsius_to_fahrenheit(weather["temperature_c"])
return format_answer(weather, fahrenheit)
这里下一步做什么,是程序员早就写好的。即使 format_answer() 内部调用了模型,整体路线仍然由代码决定。这样的系统更适合叫 Workflow。
Agent loop 的区别,不是代码里多了一个 while,而是模型获得了有限的下一步决策权:
turn = model.generate(messages, available_tools)
模型可以决定“现在回答”“先查天气”“先做换算”。但别把“决策权”理解成“系统控制权”。它仍然只能从 Runtime 暴露给它的出口里选。
我习惯用一句话区分三者:
Model 提议下一步,Runtime 管理下一步,Tool 实现下一步。
Runtime 像舞台监督,不负责演戏,却决定什么时候开场、哪个道具能上台、什么时候必须收工。模型像演员,可以根据现场情况做选择,但演员不能因为台词里写了“现在炸掉舞台”就真的获得炸药权限。
2.1 什么时候反而不该用 Agent
这个问题值得现在就说,因为很多教程会让人产生一种错觉:Agent 比 Workflow 更“高级”,所以能用 Agent 就尽量用 Agent。
实际工程里恰恰相反。如果下一步可以靠清楚的 if/else、状态机或固定流程决定,就优先使用确定性代码。它更容易测试、更容易估算成本,也更容易解释“为什么系统做了这一步”。
只有当任务确实需要模型理解开放语言、观察结果,并据此选择下一步时,Agent loop 才开始有价值。
换句话说,Agent 不是默认升级包,而是一种用复杂度换灵活性的工具。
3. ReAct:不要把它学成“打印 Thought”
你会经常看到 ReAct 这个词。它来自 Reasoning and Acting。历史上很多示例会写成:
Thought: I need the weather first.
Action: get_weather
Observation: ...
Thought: Now I should convert the temperature.
这个形式帮助人理解,但工程实现里千万别误会成“Runtime 必须读取模型的隐藏思维链”。我们真正需要的是可观察、可校验的事件:模型请求了哪个 Tool、参数是什么、Tool 返回了什么、模型什么时候结束。
所以本章把 ReAct 理解成:
Decision
↓
Action / Tool Call
↓
Application executes
↓
Observation
↓
Next Decision
如果 Runtime 依赖:
if "Action:" in model_text:
...
那你的控制协议其实建立在标点符号上。模型少写一个冒号,系统就像门禁因为员工忘记说“芝麻开门”而彻底失灵。
结构化 Tool Call 的意义就在这里:我们让“动作请求”成为明确的数据,而不是一段需要猜格式的散文。
4. Runtime 先需要一套自己的“内部语言”
真实模型服务的响应对象往往很丰富。以某个 Provider 为例,你可能看到 response.output、function_call、arguments、response.id 等字段。
如果 Runtime 直接写:
for item in response.output:
if item.type == "function_call":
...
它当然能跑,但核心循环已经和某家 Provider 的 wire format 绑在一起了。以后字段变化,或者你接另一个 Provider,Runtime 也要跟着改。
我们先问一个更朴素的问题:Runtime 真正需要知道多少?
其实很少。
一个工具请求只需要:
@dataclass(frozen=True)
class ToolCall:
call_id: str
name: str
arguments: dict[str, Any]
一次模型决策则只有两种结果:
@dataclass(frozen=True)
class ModelTurn:
final_text: str | None = None
tool_calls: tuple[ToolCall, ...] = ()
要么模型给最终文本,要么它给一个或多个 Tool Call。
4.1 为什么 ModelTurn 要强制二选一
本章的实现会检查:
has_final = self.final_text is not None
has_calls = bool(self.tool_calls)
if has_final == has_calls:
raise InvalidModelTurnError(
"A model turn must contain exactly one of final_text or tool_calls"
)
也就是说,不能两者都没有,也不能两者同时有。
真实 Provider 的输出形式可能比这复杂,但内部协议没必要照单全收。我们是在为 Runtime 设计一个容易推理的状态转移:
ModelTurn(final_text=...)
→ 结束
ModelTurn(tool_calls=...)
→ 执行 → Observation → 下一轮
内部协议越清楚,Runtime 越容易测试。Adapter 的工作,就是把外部世界那些花花绿绿的字段翻译成这两种明确结果。
4.2 call_id 为什么要求唯一
ToolCall 还会检查 call_id、工具名和参数类型。Runtime 甚至会拒绝同一次 run 中重复出现的 call_id。
这是因为 Tool Call 和 Tool Output 的对应关系依赖这个 ID。重复使用同一个 ID,相当于快递公司给两件不同包裹贴同一个单号。简单例子里可能暂时看不出问题,等你开始有多轮调用,结果关联就会变得含糊。
5. Tool 不是“一个 Python 函数”这么简单
如果只写 Demo,我们可以准备一个字典:
handlers = {
"get_weather": get_weather,
"convert": convert,
}
但只靠函数名还不够。模型需要知道这个能力是做什么的、参数长什么样;应用执行之前还需要验证参数。
所以本章的 Tool 把这几件事放在一起:
@dataclass(frozen=True)
class Tool:
name: str
description: str
arguments_model: type[BaseModel]
handler: Callable[[Any], Any]
这四个字段刚好横跨两个世界。
给模型看的,是 name、description 和根据 arguments_model 生成的 JSON Schema。给应用执行的,是 handler。
5.1 为什么参数模型要在 Runtime 里再验证
我们用 Pydantic 定义天气工具参数:
class WeatherArguments(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
city: Literal["Tokyo", "Paris"]
然后 Tool 在真正调用 handler 前执行:
arguments = self.arguments_model.model_validate(raw_arguments)
这里和 Stage 00 的思想一致:Provider 侧的 strict schema 能帮助模型生成正确结构,但真正承担执行后果的是 Runtime,所以 Runtime 仍然验证自己即将接收的参数。
这并不多余。想想 Web 开发:前端已经写了表单校验,后端还会不会校验?当然会。因为前端只是一个输入来源,真正写数据库的是后端。
5.2 ToolRegistry 到底解决什么
Runtime 不应该根据模型返回的任意名字去找 Python 函数,所以我们有一个 Registry:
class ToolRegistry:
def __init__(self, tools: list[Tool]) -> None:
self._tools = {tool.name: tool for tool in tools}
执行时只从已注册 Tool 中查找:
tool = self._tools.get(call.name)
if tool is None:
raise UnknownToolError(f"Unknown tool: {call.name}")
Registry 不是完整的权限系统,但它至少建立了一个很重要的事实:模型只能请求应用明确提供的能力。 模型生成 delete_everything 这个名字,并不会凭空让系统多出一个函数。
6. 现在来看 Runtime 的核心循环
完整实现位于 code/runtime.py。你可以先运行:
python stages/01-react-runtime/code/runtime.py
它使用一个确定性的 ScriptedWeatherModel,所以不需要 API Key。预期轨迹大概是:
[1] ACTION get_teaching_weather({'city': 'Tokyo'})
[1] OBSERVE {"city": "Tokyo", "temperature_c": 18.0, "condition": "cloudy"}
[2] ACTION celsius_to_fahrenheit({'temperature_c': 18.0})
[2] OBSERVE {"temperature_f": 64.4}
[3] FINAL Tokyo's deterministic teaching record is 18.0°C (64.4°F), cloudy.
先别急着看所有类。把刚才的执行轨迹展开,一次 Runtime run 的控制流如下:
接下来我们再沿着 AgentRuntime.run(),把图里的每一步和代码对应起来。
6.1 Runtime 先创建自己的运行记录
messages: list[dict[str, Any]] = [
{"role": "user", "content": user_input}
]
这个 messages 很重要,因为它明确说明“当前 run 发生过什么”。
模型第一次返回 Tool Call 后,Runtime 会把模型的动作请求记录下来:
messages.append(
{
"role": "assistant",
"content": "",
"tool_calls": [asdict(call) for call in turn.tool_calls],
}
)
工具执行后,又追加 Observation:
messages.append(
{
"role": "tool",
"tool_call_id": call.call_id,
"name": call.name,
"content": observation,
}
)
于是第二轮模型看到的不是“请继续猜”,而是一条明确轨迹:用户提了什么、模型请求过什么、程序实际返回了什么。
这里顺便澄清一个常见说法:“模型记住了上一轮。”从这个代码看,更准确的描述应该是:应用把上一轮发生的事情放进下一轮输入,所以模型看到了它。
6.2 每一轮只让模型做一次决定
循环中心是:
for step in range(1, self.max_steps + 1):
turn = self.model.generate(messages, self.registry.schemas())
如果模型给最终文本:
if turn.final_text is not None:
return RunResult(...)
如果模型给 Tool Call,就执行:
for call in turn.tool_calls:
result = self.registry.execute(call)
然后把结果放回 messages,进入下一轮。
注意 Runtime 并没有“替模型规划”。它只是维持规则:你可以回答,也可以请求已注册工具;请求工具以后,我来执行并把 Observation 给你;直到你回答或达到停止条件。
这就是 Runtime 控制权与模型决策权的区别。
6.3 一轮多个 Tool Call,不等于并发执行
ModelTurn 可以表示多个 tool_calls,但当前实现使用:
for call in turn.tool_calls:
result = self.registry.execute(call)
所以它们仍然按顺序执行。
这点非常容易被“parallel tool calls”几个字带偏。模型一次提出多个请求,只表示请求被一起产生;Runtime 是否并发执行,是另一件事。并发会牵涉共享状态、执行顺序、取消、部分失败等新问题,本章故意先保持同步顺序。
6.4 max_steps 到底数什么
如果模型永远不结束,Runtime 最后会:
raise MaxStepsExceeded(
f"The run did not finish within max_steps={self.max_steps} model turns"
)
这里的 max_steps 数的是模型决策轮数,不是 Tool 调用总数。因为一轮里可能有多个 Tool Call。
这个限制不是装饰参数。没有它,模型如果不停请求工具,Runtime 就会一直继续。无限 while True 在白板上很有自由精神,在真实账单里通常不太浪漫。
当然,max_steps 也不是万能预算。它不能等价于“最多花多少钱”或“最多运行几秒”。它只是本章最基本的停止边界。
7. 错误不要统统叫“Agent 出错了”
Agent 系统有一个特别容易养成的坏习惯:任何问题都描述成“模型没做好”。
但沿着刚才的调用链看,错误其实发生在不同层。
模型如果返回一个不符合内部协议的结果,是 InvalidModelTurnError。模型请求了 Registry 里没有的工具,是 UnknownToolError。工具参数没通过 Pydantic,是 ToolArgumentsError。参数合法,但 handler 自己执行失败,是 ToolExecutionError。
这些错误看起来都可能导致任务没完成,但责任完全不同。比如:
UnknownToolError
先检查模型是否请求了不暴露的能力,或 Tool 列表是否配置错误
ToolArgumentsError
先检查参数 Schema、模型参数和应用输入边界
ToolExecutionError
先检查真实 Python handler 或外部服务
把它们全吞进:
except Exception:
pass
不会让系统更健壮,只会让错误从“报出来”升级成“消失了”。
7.1 为什么本章 Tool 失败就停止
你可能想到另一种设计:Tool 失败后,把错误作为 Observation 再交给模型,让它改参数后重试。
这当然可以,而且很多系统会这么做。但一旦加入自动重试,就必须回答:这个动作能安全重复吗?已经产生了一半副作用怎么办?最多重试几次?
这些问题会迅速把一章基础 Runtime 拖进另一整套可靠性策略里。
所以本章选择一个容易推理的规则:Tool 执行失败,本次 run 直接失败。 这样你能准确知道 handler 执行了几次。先把简单语义弄明白,再谈更复杂的恢复策略。
8. 为什么先用 ScriptedWeatherModel,不直接上真实模型
如果你第一次测试 Runtime 就接真实模型,一旦轨迹不对,会遇到一个很烦的问题:到底是 Runtime 写错了,还是模型这次选择变了?
所以 runtime.py 里准备了一个确定性的模型替身:
class ScriptedWeatherModel:
...
它没有语言智能,只按观察结果数量决定下一步:没有 Observation 时请求天气;有一次 Observation 时请求换算;有两次时返回最终答案。
这听起来“不智能”,但对于测试控制器来说恰恰是优点。我们希望输入一样,轨迹就一样。
这里 Model 定义成 Protocol:
class Model(Protocol):
def generate(
self,
messages: list[dict[str, Any]],
tools: list[dict[str, Any]],
) -> ModelTurn:
...
只要对象满足这个接口,它可以是真实 Provider Adapter,也可以是 ScriptedWeatherModel。Runtime 不需要知道区别。
这是一种很重要的工程习惯:测试控制逻辑时,尽量把模型随机性从测试里拿掉。
9. 测试的重点不是“最后一句话对不对”
运行本章检查:
python stages/01-react-runtime/code/runtime_checks.py
完整测试代码在 code/runtime_checks.py。
先看成功路径。测试不仅检查最终答案里有 64.4°F:
self.assertIn("64.4°F", result.answer)
还检查了 Tool Observation 的调用编号:
self.assertEqual(
[message["tool_call_id"] for message in tool_messages],
["call-weather", "call-convert"],
)
为什么要多此一举?因为 Agent 系统里“最终答案碰巧对了”和“执行轨迹正确”不是一回事。
想象一个系统本来应该先查数据库再回答,但模型凭空猜中了一次。如果测试只看最后字符串,它会通过;如果测试检查是否真的发生了 Tool Call,就会发现问题。
测试还故意制造了未知 Tool、非法参数、handler 报错、重复 call_id 和永不结束的模型。这些坏例子不是为了难为代码,而是为了把 Runtime 的边界变成可执行规则。
一条很实用的判断是:如果你无法写出一个确定性的反例测试,那你可能还没说清楚这个 Runtime 到底承诺什么。
10. 最后再接真实 Provider:Adapter 只做翻译
现在核心 Runtime 已经能离线工作,我们再把 DeepSeek Responses API 接进来。
运行前配置:
export DEEPSEEK_API_KEY="your-deepseek-api-key"
export DEEPSEEK_MODEL="deepseek-v4-flash"
PowerShell:
$env:DEEPSEEK_API_KEY="your-deepseek-api-key"
$env:DEEPSEEK_MODEL="deepseek-v4-flash"
Windows CMD:
set "DEEPSEEK_API_KEY=your-deepseek-api-key"
set "DEEPSEEK_MODEL=deepseek-v4-flash"
然后执行:
python stages/01-react-runtime/code/deepseek_runtime.py
完整代码在 code/deepseek_runtime.py。虽然代码仍从 openai 包导入兼容客户端,但 base_url="https://api.deepseek.com" 和 DEEPSEEK_API_KEY 决定了请求实际发往 DeepSeek。
关键类是:
class DeepSeekResponsesModel:
...
它满足 Model.generate(...) -> ModelTurn,所以对 Runtime 来说,它和 ScriptedWeatherModel 没区别。
Adapter 做的事情大致是三种翻译:
Runtime Tool schema
↓
DeepSeek function tool
DeepSeek function_call
↓
ToolCall
Runtime tool observation
↓
function_call_output
核心 Runtime 不需要 import Provider SDK,也不需要知道 response.output 里是什么对象。这就是 Adapter 的价值:不是“为了面向对象多写一个类”,而是把容易变化的外部协议挡在核心控制逻辑外面。
10.1 为什么 Adapter 每轮都发送完整轨迹
DeepSeek 的 Responses API 是无状态的,不支持用 previous_response_id 继续上一份响应。因此,对话状态必须由应用程序保存,并在下一次请求时重新发送。
Runtime 已经在 messages 中保存了完整轨迹,所以 Adapter 每轮都会调用:
self._to_deepseek_input(messages)
把内部消息转换成 DeepSeek 接受的输入项:
Runtime user message → DeepSeek user message
Runtime assistant ToolCall → DeepSeek function_call
Runtime tool observation → DeepSeek function_call_output
这样第二轮请求同时包含最初的用户问题、模型提出的 Function Call 和 Python 执行得到的 Tool Output。DeepSeek 可以从这些输入恢复完整上下文。
这正好印证前文的说法:不是模型自己记住了轨迹,而是应用记录轨迹,并在下一轮重新提供给模型。
10.2 为什么不能只发送最新的 Tool Output
如果第二轮只发送最新的 function_call_output,无状态的 Provider 并不知道它对应哪个函数,也看不到用户原来的问题。因此 Adapter 必须连同对应的 function_call 和更早的相关输入一起发送。
这种设计也让状态归属非常明确:Runtime 保存 Provider 无关的 messages;Adapter 只负责把它们转换成外部 API 的 wire format。
10.3 多个 Tool Call 怎样处理
DeepSeek 可能在一轮中返回多个 Tool Call。内部 ModelTurn 能保存多个调用,当前 Runtime 会按返回顺序逐个执行。这里的“模型可以一次提出多个调用”和“Python 是否并发执行它们”仍然是两件不同的事。
11. 到这里,一个最小 Runtime 已经具备什么
现在回头看,Runtime 已经不只是“一个 while 循环”。它把几类职责放到了明确的位置:模型通过统一协议返回决策;Tool 有清晰的描述、Schema 和 handler;Registry 限制可调用能力;Runtime 维护运行记录并控制继续或结束;参数在执行前被验证;错误有明确类型;max_steps 防止无限决策;确定性模型替身让控制逻辑可以离线测试;Provider Adapter 把外部格式挡在核心循环之外。
这些东西组合起来以后,才算真正有了一个“小而完整”的 Agent Runtime。
但它仍然很小。当前实现同步、顺序执行 Tool;状态只存在当前进程内;Tool 失败直接结束;没有自动重试,没有并发调度,也没有额外的持久化机制。
这不是“还没来得及补的 TODO 列表”,而是本章的边界。先让读者清楚系统现在确切会做什么,比用一堆“生产级”“企业级”形容词包住 Demo 更有价值。
12. 用几个实验把 Runtime 真正拆开
读完代码后,最有效的练习不是再抄一遍,而是故意改变一个假设。
你可以让 ScriptedWeatherModel 在同一轮返回两个不同的 Tool Call,观察 Runtime 会按什么顺序执行。然后试着给两个调用使用相同的 call_id,看看内部协议在哪一步拒绝它。
也可以加一个第三个工具,比如 describe_temperature,把华氏温度分类成 cold、mild、hot。只修改 Tool 和模型替身,不要改 AgentRuntime.run()。如果为了多一个工具就必须改核心循环,那说明抽象还不够稳定。
再试一个更有意思的:让 handler 主动抛异常,然后思考“失败就终止”和“把错误作为 Observation 交给模型”两种设计有什么差别。别急着选一个“更高级”的答案,先问:如果 Tool 有副作用,自动重试会不会重复执行?
最后,尝试把 max_steps 改成 1。你会看到 Runtime 不是“模型想走几步就走几步”,而是模型决策始终被应用的执行预算包在外面。
13. 本章结束时,你应该能讲清楚一条完整轨迹
假设我现在问你:“东京教学天气是多少,并换算成华氏度?”
你应该能从程序角度讲出:用户输入进入 Runtime;Runtime 把消息和 Tool Schema 交给 Model;Model 返回 ToolCall;Runtime 用 Registry 找到 Tool;Pydantic 验证参数;handler 执行;结果被序列化成 Tool Observation;下一轮 Model 看见 Observation,再请求换算 Tool;第二个 Observation 返回;最终 Model 返回 final_text;Runtime 停止并返回 RunResult。
如果你能顺着这条轨迹解释每一步“谁拥有控制权”,那你已经理解了本章最重要的内容。
你不需要背“Agent = LLM + Tools + Memory + Planning”这种公式。真正有用的理解,是打开代码时知道:模型在哪儿做决策,应用在哪儿执行,状态在哪儿保存,错误在哪儿被挡住,循环在哪儿停止。
14. 本章代码
stages/01-react-runtime/
├── README.md
├── README.zh-CN.md
└── code/
├── runtime.py
├── deepseek_runtime.py
├── runtime_checks.py
└── requirements.txt
完整实现只维护在 code/ 中;正文中的代码片段用于解释具体机制。