工具循环引擎:ToolRunner 如何驱动一次 agent turn

30 秒导读: 一个 agent 之所以能「自己动手」,靠的是一段循环——模型先说话,如果它要求调用工具就去执行,把工具结果喂回去让模型接着说,直到模型某一次不再要工具为止。这一章讲 fast-agent 里驱动这段循环的引擎:ToolRunner(一个 async 迭代器)和它背后 ToolAgent 的工具执行。不讲工具从哪来(MCP 聚合见 05)、模型底层怎么调(见 04)。

本章在整套货架里的位置:上一章 02-agent-class-stack.md 讲了 ToolAgent 这个类栈层级本身;这一章讲的是它运行时最核心的那台机器。

---

1. 这是什么(零基础也能懂)

一句话定义: ToolRunner 是「一次对话回合(turn)」的循环控制器——它反复地「问模型 → 执行模型要的工具 → 把结果还给模型」,直到模型给出不带工具请求的最终回答。

为什么需要它。 大语言模型本身只会「输入文字、输出文字」。当你希望它能查数据库、读文件、调 API,模型能做的其实只是在输出里说「我想调用工具 X,参数是 Y」。真正去执行那个工具、把结果拿回来、再让模型基于结果继续——这些「跑腿」的活儿,得由框架侧的一段循环来做。ToolRunner 就是这段循环。

用起来什么样。 使用者几乎感觉不到它的存在。你调用 agent 的 generate(...),框架内部建一个 ToolRunner 并把循环跑到底:

# 示意,非源码:使用者视角只有一句话
response = await agent.generate("帮我查一下今天北京的天气并总结")
# 这一句背后:模型说"调 get_weather" → 框架执行 → 结果喂回 → 模型给出总结
# 中间可能来回好几轮,ToolRunner 全程替你转

一句话直觉/类比。 把一次 turn 想成打乒乓球:模型发球(说话),如果球上写着「请帮我做 X」(工具请求),裁判(ToolRunner)就去做 X 再把球打回给模型;模型再打一板……直到某一板模型直接把球扣死(给出最终答案、不再要工具),这一回合才结束。

---

2. 顶层全景(一次 turn 大概怎么转)

2.1 两个角色

一次 turn 的循环由两个协作对象完成,职责清晰分开:

角色	干什么	在哪个文件
ToolRunner	循环的骨架:决定「该再问一次模型,还是该收尾」;管中断回滚、持久化、auto-compaction 挂钩	src/fast_agent/agents/tool_runner.py:131
ToolAgent	循环的两只手:一手调模型(_tool_runner_llm_step),一手规划并执行工具(run_tools)	src/fast_agent/agents/tool_agent.py:50

ToolRunner 只关心「循环的形状」,不关心工具怎么执行、模型怎么调——那两件事它都回调给 agent 去做。这种「循环控制」与「具体动作」的解耦,是这套引擎最干净的地方。

2.2 一次 turn 的循环流向

先说怎么读这张图:从上往下是时间;ToolRunner 是一个迭代器,外层 until_done() 反复向它取下一条消息;每取到一条模型消息就看它的 stop_reason——TOOL_USE 就回到顶上再转一圈,其它就收尾。

                   ┌───────────────────────────────────────────┐
   until_done()    │            外层:反复取下一条消息            │
   反复 async for  │              (tool_runner.py:287)          │
                   └───────────────────┬───────────────────────┘
                                       │ __anext__()
                                       ▼
        ┌──────────────────────────────────────────────────────┐
        │ ① 先处理"上一板留下的工具请求"                          │
        │    _prepare_next_llm_step → _ensure_tool_response_staged│
        │    有 pending 工具请求? ── 有 ─► 执行工具(run_tools) │
        │                                  把结果暂存为下一次输入 │
        │                              ── 无 ─► 直接往下           │
        └───────────────────┬──────────────────────────────────┘
                            │(没有已暂存的"终局消息"就继续)
                            ▼
        ┌──────────────────────────────────────────────────────┐
        │ ② 问模型                                               │
        │    _maybe_auto_compact_before_followup_llm  (可选压缩) │
        │    before_llm_call 钩子                                 │
        │    _call_llm ──► agent._tool_runner_llm_step(消息, 工具)│
        │    after_llm_call 钩子                                  │
        └───────────────────┬──────────────────────────────────┘
                            │ 得到一条 assistant_message
                            ▼
        ┌──────────────────────────────────────────────────────┐
        │ ③ 看 stop_reason 决定去留                               │
        │    _apply_assistant_message_state                      │
        │      == TOOL_USE ─► 记下 pending 工具请求,回到 ①      │
        │      结构化收尾  ─► 追加"出最终 JSON"指令,回到 ②       │
        │      其它        ─► _done = True,循环结束              │
        └───────────────────┬──────────────────────────────────┘
                            │ yield assistant_message 给外层
                            ▼
        外层看到 TOOL_USE → 持久化 checkpoint,继续循环
        外层看到 非 TOOL_USE → 触发 after_turn_complete,返回最终消息

主线走一遍(高层): 输入消息进 ToolRunner → 第一圈没有待办工具,直接问模型 → 模型回 TOOL_USE → 下一圈先把工具跑掉、结果暂存 → 再问模型 → 模型这次回 END_TURN → 收尾返回。中间来回几圈由模型自己决定,上限 199 圈(见 §3.4)。

---

3. 核心原理(逐个机制,由浅入深)

3.1 为什么用「async 迭代器」来写循环

它要解决的小问题: 循环每转一圈都会产出一条模型消息,而调用方(CLI、Web、上层 workflow)往往想一条一条地看到这些中间消息(比如流式显示「模型正在调用工具…」),而不是等整个 turn 跑完才拿到一坨结果。

思路/直觉: 把「一次 turn」建模成一个可迭代对象——每 __anext__ 一次就吐出「循环的下一条 assistant 消息」;迭代自然结束(StopAsyncIteration)就代表 turn 结束。这样,想要中间消息的调用方就 async for 逐条拿,不想要的就用便捷方法 until_done() 一次跑到底。

ToolRunner 同时提供这两种用法:

# 示意,非源码:两种消费方式,同一台引擎

# 逐条消费(想看中间过程)
async for message in runner:
    render(message)          # 每条模型回复都能实时显示

# 一把跑完(只要最终结果)
final = await runner.until_done()

真实实现: __anext__(tool_runner.py:188)是整个引擎的心跳。它的结构非常克制,一眼能看完:

# tool_runner.py:188  ToolRunner.__anext__(节选,真实源码)
async def __anext__(self) -> PromptMessageExtended:
    staged = await self._prepare_next_llm_step()   # ① 先消化上一板的工具 / 判断是否该停
    if staged is not None:
        return staged                              #    有"已备好的终局消息"就直接吐

    await self._maybe_auto_compact_before_followup_llm()  # ② 问模型前的准备
    await self._ensure_tools_ready()
    await self._run_before_llm_hook()
    assistant_message = await self._call_llm()     #    真正问一次模型
    await self._run_after_llm_hook(assistant_message)
    self._apply_assistant_message_state(assistant_message)  # ③ 按 stop_reason 决定去留
    return assistant_message

until_done(tool_runner.py:287)就是把这个迭代器 async for 到底,顺手做每圈的持久化和收尾钩子(见 §3.5)。

3.2 停止的唯一判据:stop_reason == TOOL_USE

它要解决的小问题: 循环凭什么知道「该继续」还是「该收尾」?

答案:看模型这一条回复的 stop_reason。 LlmStopReason 是一个字符串枚举(llm_stop_reason.py:6),把 MCP 标准停止原因和 fast-agent 自定义的几种合在一起:

stop_reason	含义	对循环的影响
TOOL_USE ("toolUse")	模型停下来要调工具	唯一让循环继续的信号
END_TURN / STOP_SEQUENCE / MAX_TOKENS / PAUSE	模型正常说完	收尾:_done = True
ERROR / CANCELLED / TIMEOUT / SAFETY	异常/中断/超时/安全	收尾(CANCELLED 走特殊回滚,见 §3.6)

真实实现: 决策集中在 _apply_assistant_message_state(tool_runner.py:277)——整个引擎「继续 vs 结束」的开关就在这里:

# tool_runner.py:277  _apply_assistant_message_state(真实源码)
def _apply_assistant_message_state(self, assistant_message):
    if assistant_message.stop_reason == LlmStopReason.TOOL_USE:
        self._pending_tool_request = assistant_message   # 记下"待执行的工具请求"
        self._pending_tool_response = None
        return                                           # 不置 _done,循环继续
    if self._should_start_deferred_structured_finalization(assistant_message):
        self._start_deferred_structured_finalization(assistant_message)  # 结构化收尾,见 §3.7
        return
    self._done = True                                    # 其它一律收尾

重点看: TOOL_USE 分支只记下待办、不结束;它把「执行工具」推迟到下一次 __anext__ 开头的 _prepare_next_llm_step 里做。也就是说,一条 TOOL_USE 消息先被 yield 给调用方(让它显示「要调工具了」),工具的实际执行发生在下一圈——这正是 async 迭代器能实时汇报中间状态的关键。

3.3 工具请求怎么被消化并喂回模型

它要解决的小问题: 模型说「调 get_weather」之后,谁去真正跑它,跑完的结果又怎么变成模型下一次的输入?

流程: 下一圈 __anext__ 一进来,_prepare_next_llm_step → _ensure_tool_response_staged(tool_runner.py:780)发现有 pending 工具请求,于是:

1. 调 generate_tool_call_response()(tool_runner.py:534)执行工具——内部走 before_tool_call 钩子 → agent.run_tools(...) → after_tool_call 钩子;
2. 迭代计数 +1 并检查是否超上限(§3.4);
3. 把工具结果通过 _stage_tool_response(tool_runner.py:626)塞进 _delta_messages,作为下一次问模型的输入。

一个巧妙的取舍: 结果怎么塞,取决于是否开启历史(use_history):

# tool_runner.py:626  _stage_tool_response(节选,真实源码)
if self._use_history_enabled():
    self._delta_messages = staged_messages          # 开历史:只把"增量"设为工具结果
                                                    # (完整历史由 agent 侧维护)
else:
    if self._last_message is not None:
        self._delta_messages.append(self._last_message)  # 不开历史:自己把上一条也带上
    self._delta_messages.extend(staged_messages)

_delta_messages 是「下一次 LLM 调用要发的增量消息」,不是完整历史(见属性注释 tool_runner.py:610)。开启历史时,完整对话由 agent 自己维护,ToolRunner 只需把最新的工具结果作为增量;关闭历史时,ToolRunner 得手动把上一条 assistant 消息也拼进去,模型才有上下文。

工具结果为什么长成 user 消息? run_tools 最终返回的是一条 role="user" 且 tool_results=... 的消息(见 tool_agent.py:749、_finalize_tool_results at tool_agent.py:835)。在多数模型的对话协议里,工具执行结果是以「用户侧」身份回给模型的——ToolRunner 顺应了这个约定。

3.4 迭代上限:防止无限循环

模型理论上可以「一直要工具」,循环就永不停。引擎用一个简单的硬闸拦住:每消化一次工具结果,_iteration += 1,超过 max_iterations 就强制 _done。

# tool_runner.py:803(节选,真实源码)
self._iteration += 1
max_iterations = (
    self._request_params.max_iterations
    if self._request_params is not None
    else DEFAULT_MAX_ITERATIONS
)
if self._iteration > max_iterations:
    self._done = True
    return

DEFAULT_MAX_ITERATIONS = 199(constants.py:55),注释把它定义为「最大 User/Assistant 回合数」。到顶就停,不会再问模型。

3.5 生命周期钩子:在循环的关键点插手

它要解决的小问题: 上层想在「问模型前 / 得到回复后 / 执行工具前后 / 整个 turn 结束时」插入自定义逻辑(改历史、调参数、发进度、做压缩),但又不想改动循环本身。

思路: ToolRunner 暴露一组钩子点,用一个 frozen dataclass ToolRunnerHooks(tool_runner.py:103)承载。钩子是「low-level 且允许改动」的——它们能读写 agent 历史、改 request 参数、追加消息:

钩子	触发时机
before_llm_call	每次问模型之前(tool_runner.py:220)
after_llm_call	每次拿到模型回复之后(tool_runner.py:247)
before_tool_call	执行工具之前(tool_runner.py:542)
after_tool_call	拿到工具结果之后(tool_runner.py:573)
after_turn_complete	整个 turn 结束(仅当 stop_reason != TOOL_USE)后一次性触发(tool_runner.py:307)

注意最后一个的语义:after_turn_complete 在 until_done 里、循环彻底结束后才触发一次(tool_runner.py:307),而不是每圈都触发——它对应「这一整回合真的说完了」。fast-agent 自己就用这套钩子做循环进度上报:ToolAgent._build_loop_progress_hooks(tool_agent.py:446)把 before_llm_call / before_tool_call 接到一个 emitter,前端就能显示「正在思考 / 正在调用工具 X」。多套钩子还能用 _merge_tool_runner_hooks(tool_agent.py:489)串起来依次调用。

3.6 中断与取消:历史回滚 + 持久化

它要解决的小问题: 用户在工具跑到一半时按了 Ctrl-C(或外部取消了任务)。此时对话历史里很可能只有模型「我要调工具」那半句,却没有工具结果——这是一段「悬空」的历史,下次接着聊会让模型协议报错。必须把它「补齐」再存下来。

思路: 取消时做两件事——回滚/修补历史 和 尽力持久化。

① 回滚:补一条「被中断」的工具结果。 核心是静态方法 reconcile_interrupted_history(tool_runner.py:447)。它检查历史末尾是不是一条「悬空的工具请求」(assistant + 有 tool_calls + stop_reason == TOOL_USE,判定见 _pending_tool_request_at_history_end at tool_runner.py:710);若是,就伪造一条工具结果补上:

# tool_runner.py:498  _build_interrupted_tool_result(节选,真实源码)
interrupted_text = "**The user interrupted this tool call**"
tool_results = {}
for tool_id in pending_request.tool_calls or {}:
    tool_results[tool_id] = CallToolResult(
        content=[text_content(interrupted_text)],
        isError=True,          # 标成错误结果,如实告诉模型"这次被打断了"
    )

这样每个悬空的 tool_call 都配上一条「被用户打断」的错误结果,历史重新自洽。reconcile_interrupted_history 返回一个 HistoryRollbackState(tool_runner.py:93)记录发生了什么(补了 / 空历史 / 未开历史 / 无需改动),便于审计。_reset_history_after_cancelled_turn(tool_runner.py:491)是它在取消路径上的封装。

② 持久化:即使自己正被取消也要把状态存下。 until_done 的异常处理(tool_runner.py:317 起)分三类:asyncio.CancelledError、KeyboardInterrupt、其它 Exception——每类都先清理钩子缓冲、回滚历史,再持久化。其中 asyncio.CancelledError 有个精巧处理:任务已被取消时,协程里再 await 存盘会立刻又被取消,所以 _persist_cancelled_turn_state_after_task_cancel(tool_runner.py:426)先临时 task.uncancel()、存完再把取消「还回去」:

# tool_runner.py:426(节选,真实源码)
for _ in range(cancellation_requests):
    task.uncancel()                 # 暂时"解除取消",好让存盘 await 能跑完
try:
    await self._persist_cancelled_turn_state()
finally:
    for _ in range(cancellation_requests):
        task.cancel()               # 存完再把取消原样还回去

③ 每圈 checkpoint。 正常路径下,until_done 每见到一条 TOOL_USE 消息就 _persist_tool_loop_checkpoint(tool_runner.py:376)存一次进度,底层是 best-effort 的 _persist_session_history_best_effort(tool_runner.py:393)——存盘失败只 warning、绝不打断循环。这几条合起来保证:turn 中途无论正常暂停、崩溃还是被杀,历史都能落到一个可恢复的一致点。

同一套修补逻辑在入口也复用了:ToolAgent.generate_impl(tool_agent.py:368)开跑前,若历史里有上次遗留的悬空工具请求,会先 reconcile_interrupted_history 自愈一遍再建 ToolRunner——「自愈」发生在存盘时,也发生在下次读取时。

3.7 auto-compaction 挂钩:长对话中途「瘦身」

它要解决的小问题: 工具来回多轮后,上下文可能撑爆模型窗口。理想是在「下一次问模型之前」悄悄把旧历史压缩成摘要,又不能破坏当前正在进行的这一板。

挂钩点: _maybe_auto_compact_before_followup_llm(tool_runner.py:340),在 __anext__ 里紧挨着「再问一次模型」之前调用。它只在上一条是 TOOL_USE(即确实处在 followup 场景)时才动作,把活儿转交给 auto_compact_history_mid_turn(hooks/compaction.py:52),后者用 min_keep_turns=1 保住当前 turn。

关键取舍——绝不因压缩而弄坏循环:

# tool_runner.py:340(节选,真实源码)
try:
    await auto_compact_history_mid_turn(HookContext(runner=self, agent=..., message=message,
                                                    hook_type="before_followup_llm_call"))
except Exception:
    # 中途压缩是"机会主义"的;绝不让它中断工具循环
    _logger.exception("Auto-compaction failed during tool loop; history unchanged")

压缩失败被整段吞掉、只记日志——引擎的态度是「压缩能成最好,不成就带着原历史继续」,循环的健壮性优先于省 token。

---

4. 深入实现:ToolAgent 侧的工具规划与执行

ToolRunner 把「执行工具」整件事回调给 agent.run_tools(...)。这一节看 ToolAgent 真正怎么把「一批工具请求」跑出来。

4.1 规划(plan)→ 执行(run)→ 收尾(finalize)

run_tools(tool_agent.py:741)是三段式:

1. 规划 _plan_tool_calls(tool_agent.py:597)→ 底层 plan_tool_calls(tool_call_planning.py:46):把每个 tool_call 解析成一个 PlannedToolCall(correlation_id, name, arguments)。一旦发现某个工具名不在可用工具里,立即短路,把它标成「不可用」错误结果,不再往下执行。
2. 执行:根据 should_parallelize_tool_calls(constants.py:49,规则:>1 个工具且未强制串行就并行)选择并行或串行。
3. 收尾 _finalize_tool_results(tool_agent.py:835)→ build_tool_result_message:把所有结果、计时、元数据打包成一条 role="user" 的消息回给 ToolRunner。

4.2 并行 vs 串行

并行(_run_parallel_tool_calls,tool_agent.py:671):用 gather_with_cancel 并发跑所有 planned call,每个结果用 correlation_id 对号入座;单个工具抛异常会被兜成一条 isError 结果,不拖垮其它并行分支。

串行(_run_sequential_tool_calls,tool_agent.py:714):一个个 await,顺序执行。

两条路都为每个工具记 ToolTimingInfo(耗时)并调 display.show_tool_result 实时显示。选并行还是串行,只由「工具个数」和全局开关 FORCE_SEQUENTIAL_TOOL_CALLS(constants.py:45)决定。

4.3 单个工具的执行:call_tool 与结果类型转换

单个工具最终落到 call_tool(tool_agent.py:855)。它从本地 _execution_tools 取出 FastMCP 的 FunctionTool,await fast_tool.run(arguments) 得到一个原生 ToolResult,再转成协议层通用的 CallToolResult(MCP 类型):

# tool_agent.py:926  _native_tool_result_to_mcp_result(真实源码)
@staticmethod
def _native_tool_result_to_mcp_result(result: ToolResult) -> CallToolResult:
    return CallToolResult(
        content=result.content,
        structuredContent=result.structured_content,
        _meta=result.meta,
        isError=False,
    )

为什么要转: FastMCP 执行工具产出的是它自己的 ToolResult(fastmcp.tools.ToolResult),而 fast-agent 循环内部统一以 MCP SDK 的 CallToolResult 流通(错误也统一包成 isError=True 的 CallToolResult)。这一层转换,让「本地函数工具」和「远程 MCP 工具」在循环眼里长得一模一样——循环无需关心工具到底来自哪里。工具执行途中还会通过 ToolExecutionHandler(_get_tool_handler,tool_agent.py:913)上报 on_tool_start/on_tool_complete 进度。

方向对照表:

阶段	类型	出处
FastMCP 工具执行产出	ToolResult(原生)	fastmcp.tools
转换后循环内流通	CallToolResult(MCP)	_native_tool_result_to_mcp_result (tool_agent.py:926)
打包回给模型	role="user" + tool_results	_finalize_tool_results (tool_agent.py:835)

---

5. 巧妙之处(可借鉴的技术)

• 循环控制与动作彻底解耦。 ToolRunner 不知道工具怎么跑、模型怎么调,全靠回调 agent 的 _tool_runner_llm_step / run_tools。同一台循环引擎因此能驱动任意 agent(纯 LLM、工具 agent、编码 agent),见 _ToolLoopAgent 协议(tool_runner.py:42)。
• 用 async 迭代器同时满足「要中间态」和「只要结果」两类调用方(§3.1),一套代码两种消费姿势。
• 停止判据收敛到单一信号 stop_reason == TOOL_USE(§3.2),循环的「继续/结束」开关集中在一个方法里,读代码时一眼可辨。
• 中断即自愈的历史模型(§3.6):悬空工具请求会被补一条 isError 的「被打断」结果,存盘时补、读取时也补,双保险让长会话在崩溃/取消后仍能续聊。
• 取消中还能存盘的 uncancel/recancel 小技巧(tool_runner.py:426):临时解除取消跑完存盘、再把取消还回去,细节到位。
• 压缩失败绝不弄坏循环(§3.7):auto-compaction 全程 try/except 吞掉,健壮性优先。

---

6. 边界与局限

• 循环上限 199 圈(constants.py:55)。模型若一直要工具,到顶被硬停,最终消息可能不是模型「说完」的自然结束。
• auto-compaction 是机会主义的:失败静默、只记日志(§3.7);它保证不弄坏循环,但不保证一定压缩成功。
• 持久化是 best-effort:_persist_session_history_best_effort(tool_runner.py:393)存盘失败只 warning。可靠落盘不由本引擎兜底。
• 工具「不可用」即短路:plan_tool_calls(tool_call_planning.py:46)一遇未知工具名就停止规划后续工具,把它标错——同一批里排在它后面的工具本轮不会被执行。
• 本章不覆盖:工具从哪聚合而来(MCP 服务器连接/工具过滤,见 05);模型字符串解析与 provider 适配(见 04);把多个 agent 编排成工作流(见 06)。

---

7. 代码地图(导航索引)

主题	文件路径	符号名
循环骨架 / async 迭代器	src/fast_agent/agents/tool_runner.py	ToolRunner
单圈心跳(问模型 or 收尾)	src/fast_agent/agents/tool_runner.py	ToolRunner.__anext__
消化上一板工具 / 判停	src/fast_agent/agents/tool_runner.py	_prepare_next_llm_step / _ensure_tool_response_staged
跑到底 + 收尾 + 异常处理	src/fast_agent/agents/tool_runner.py	ToolRunner.until_done
停/继续 决策	src/fast_agent/agents/tool_runner.py	_apply_assistant_message_state
停止原因枚举	src/fast_agent/types/llm_stop_reason.py	LlmStopReason
生命周期钩子	src/fast_agent/agents/tool_runner.py	ToolRunnerHooks
中断历史自愈	src/fast_agent/agents/tool_runner.py	reconcile_interrupted_history / _build_interrupted_tool_result
取消回滚	src/fast_agent/agents/tool_runner.py	_reset_history_after_cancelled_turn
checkpoint / 持久化	src/fast_agent/agents/tool_runner.py	_persist_tool_loop_checkpoint / _persist_session_history_best_effort
取消中存盘技巧	src/fast_agent/agents/tool_runner.py	_persist_cancelled_turn_state_after_task_cancel
mid-turn 压缩挂钩	src/fast_agent/agents/tool_runner.py	_maybe_auto_compact_before_followup_llm
建 ToolRunner 的入口	src/fast_agent/agents/tool_agent.py	ToolAgent.generate_impl
问模型的回调	src/fast_agent/agents/tool_agent.py	_tool_runner_llm_step
工具规划→执行→收尾	src/fast_agent/agents/tool_agent.py	run_tools / _plan_tool_calls / _finalize_tool_results
并行 / 串行执行	src/fast_agent/agents/tool_agent.py	_run_parallel_tool_calls / _run_sequential_tool_calls
单工具执行 + 结果转换	src/fast_agent/agents/tool_agent.py	call_tool / _native_tool_result_to_mcp_result
规划原语	src/fast_agent/agents/tool_call_planning.py	plan_tool_calls / PlannedToolCall
迭代上限 / 并行判定	src/fast_agent/constants.py	DEFAULT_MAX_ITERATIONS / should_parallelize_tool_calls