数据截至 (上游 commit e50a5bce6439)
第 4 章 Agent:在引擎之上手写的「LLM ↔ 工具」循环
本章讲什么: Haystack 的 Agent 不是一张 Pipeline 图,而是
haystack/components/agents/agent.py里一个手写的 while 循环。看它怎么问 LLM、怎么执行工具、什么时候停;以及 2026 年中的大改:ToolInvoker 组件被移除、工具调用收编进 Agent(tool_calling.py),Agent 断点/快照机制被移除(#11202)。
1. 它要解决的小问题
管道引擎调度的是「一次性 DAG 数据流」;而 agent 要的是「LLM 看结果再决定下一步」的循环。Haystack 的选择:不改造引擎,在引擎之上手写一个循环,把 LLM 和工具都当普通对象直接调。
2. 零件清单
| 零件 | 角色 | 在哪 |
|---|---|---|
chat_generator | 调 LLM,必须支持 tools 参数 | 构造时校验,agent.py:262-268 一带 |
| 工具执行器 | 批量执行 LLM 想调的工具(原 ToolInvoker 组件,已收编) | components/agents/tool_calling.py |
State | 跨步骤累积的「记忆」(消息历史 + 自定义键) | components/agents/state/state.py:83 |
没工具时,Agent 就退化成一个普通 ChatGenerator(构造时会 warn)。
3. 核心机制:循环主体
入口是 Agent.run(agent.py:803;异步版 run_async 在 :905),主体是 while exe_context.counter < self.max_agent_steps(:948),每步调 _run_step(:968,异步版同构)。一步里发生的事:
① 每步先重铺工具列表(:973-976):flatten_tools_or_toolsets 每步重算——SearchableToolset 这类动态工具集在后续步骤发现的新工具能冒出来;重名在这里就炸,不带病上路。
② 问 LLM(:980-994):先跑 BEFORE_LLM 钩子,然后直接 chat_generator.run(messages, tools=…)(:989——早期版本借道管道引擎的单组件执行器,现已直调),回复存进 state["messages"],顺手记录 token 用量。
③ 文本退出判断(:997-1001):没工具可用,或最后一条消息是「非空 assistant 文本且没有任何 tool_call」(_is_text_exit,:160)——退出,exit_reason = "text"。要求非空文本是为了不让空回复误触发退出。
④ 执行工具(:1003-1018):先跑 BEFORE_TOOL 钩子,然后从 State 里重读待执行调用(:1006)——因为钩子(比如 ConfirmationHook 人工确认)可能刚改过它们(agent.py:167-172 注释);再 _run_tool(**inputs) 执行(:1014),结果消息回灌 state,_record_tool_calls 记账,跑 AFTER_TOOL 钩子。
⑤ 显式退出条件(:1021-1024):若 exit_conditions 不是默认的 ["text"] 而是某些工具名,_check_exit_conditions(:1096)检查 LLM 是否调了这些工具且没出错——是则带着工具名退出。exit_reason 的三种取值写在 run_async 的 docstring 里:"text" / 工具名 / "max_agent_steps"(:924-927)。
钩子系统是这版 Agent 的新骨架:BEFORE_RUN / BEFORE_LLM / BEFORE_TOOL / AFTER_TOOL / AFTER_RUN 五个挂点(:945、:980、:1003、:1018、:959),人工确认、日志、护栏都从这里插。
4. State:Agent 的记忆
State(state.py:83)是一个带 schema 的键值容器。每个键有一个 type 和一个合并 handler:
- list 类型默认
merge_lists(追加); - 其他类型默认
replace_values(覆盖)。
两条默认规则的出处见 state.py:98-99 的类 docstring。这解释了为什么消息能「累积」:messages 键被注册成 list[ChatMessage] + merge_lists(agent.py:486),每轮 state.set("messages", 新消息) 是追加而非覆盖。而钩子等需要替换整段历史的场景,会显式传 handler_override=replace_values(agent.py:978 把当前工具列表整体覆盖进 state)。
Agent 的输入/输出口是动态的:遍历 state_schema,每个键都 component.set_input_type / 汇总进 set_output_types(agent.py:500-501)——这正是第 2 章动态 socket 的实战。所以工具可以通过 inputs_from_state / outputs_to_state 读写这些状态键。
5. 工具执行:按 State 读写集合排并行批次
ToolInvoker 组件已删除(上游 PR #11415),工具调用代码收进 tool_calling.py,核心是两步:
第一步——排批次(_schedule_tool_calls,:439-491):分析每个工具调用读/写哪些 State 键(_state_io_for_call,:413),做分层拓扑排序:
- 读某键的调用,必须排在写同键的调用之后(read-after-write);
- 同批内互相无依赖,并行跑;
- 依赖成环(一个工具又读又写同一键、还被调了两次)时,按调用序号确定性地破环(:483-485);
- 纯「写-写」冲突不产生依赖——没人读就不怕撞,结果稍后按调用序合并,照样确定(:452-453)。
第二步——分批执行(_run_tool,:514-600):ThreadPoolExecutor(max_workers=4)(:522、:559)逐批并行;每批开头才准备参数,让读 State 的工具看到前批写入(:561-562 注释);批内按调用序收结果合并(:578-587),返回的消息列表也保持调用序(:551-553)。出错时若 raise_on_failure=False 就把错误包成一条 tool 消息回给 LLM(_finalize_tool_result,:494-511),否则抛。
「错误变成消息回灌给 LLM」是关键设计:让模型自己看到工具报错、决定要不要重试或换路。
Tool 本身(tools/tool.py:20)= name / description / parameters(JSON schema)/ 函数。新变化:工具函数不再强制同步——function(同步)与 async_function(协程)可单给或都给(tool.py:35-44);只给同步函数时,invoke_async 会自动用 asyncio.to_thread 兜底(:43-44)。invoke 本体在 :283。
6. 断点与恢复:已从 Agent 移除
早期版本的 Agent 支持 AgentBreakpoint / ToolBreakpoint / AgentSnapshot(在 LLM 或工具处断点、快照恢复)。这套机制已在上游 PR #11202 中移除,官方文档页也一并撤下(#11955)。
现在还在的是管道级 断点:core/pipeline/breakpoint.py(第 1 章),#11883 还给同步管道加了同 run 内的断点/快照支持。Agent 层面的「停下来看看」,如今用钩子(§3)或 exit_conditions + max_agent_steps 表达。
下一章:巧妙之处、边界局限、横向对比和完整代码地图。见 05-clever-and-boundaries.md。