Haystack Agent 组件深度解析基于 ChatGenerator 与 State 的工具型 Agent 架构指南【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文是 Haystack 官方参考文档 agents-api 的技术解读与实战手册。文章以该文档定义的Agent与State两大核心类为主线结合当前仓库源码haystack/components/agents/agent.py、haystack/components/agents/state/state.py深入展开你将掌握 Agent 的构造参数、退出条件、运行循环、序列化机制以及 State 状态容器的 schema 合并语义并能在真实项目中用几十行代码搭建一个可调工具、可共享上下文的 LLM Agent。一、Agent 是什么一个工具型 LLM 组件的整体定位根据参考文档的定义Agent是一个实现工具使用型 Agent的 Haystack 组件其核心特性是与聊天模型供应商无关只要 ChatGenerator 的run()方法支持tools参数就可以作为 Agent 的推理引擎例如OpenAIChatGenerator等。循环处理消息并执行工具组件会不断思考 → 调用工具 → 观察工具结果 → 再思考直到满足某个退出条件exit condition才停止。退出条件可配置既可以在模型直接生成文本回复时退出text也可以在指定的某个工具被执行完后退出工具名多个退出条件可同时指定。无工具即退化为纯 LLM当不传入任何工具时Agent 的行为等价于一个 ChatGenerator产生一条回复后立即退出。从当前仓库的类注释haystack/components/agents/agent.py可以看到同样定位的描述A tool-using Agent powered by a large language model. The Agent processes messages and calls tools until it meets an exit condition.该组件在haystack.components.agents包下导出与State一起对外可见见 haystack/components/agents/init.pyfrom haystack.components.agents import Agent from haystack.components.agents.state import State二、Agent 构造参数详解参考文档给出了Agent.__init__的完整签名def __init__(*, chat_generator: ChatGenerator, tools: Optional[Union[list[Tool], Toolset]] None, system_prompt: Optional[str] None, exit_conditions: Optional[list[str]] None, state_schema: Optional[dict[str, Any]] None, max_agent_steps: int 100, streaming_callback: Optional[StreamingCallbackT] None, raise_on_tool_invocation_failure: bool False, tool_invoker_kwargs: Optional[dict[str, Any]] None) - None各参数含义如下表参数与说明以参考文档为准参数类型默认值说明chat_generatorChatGenerator必填Agent 使用的聊天生成器实例必须支持 tools其run()方法需接受tools参数。toolslist[Tool]/ToolsetNoneAgent 可使用的工具列表或工具集。system_promptstrNone系统提示词用于约束 Agent 行为。exit_conditionslist[str][text]退出条件列表。可包含text模型生成无工具调用的消息时退出或工具名该工具执行完成后退出。state_schemadict[str, Any]None工具运行期共享状态的 schema供工具读写。max_agent_stepsint100Agent 运行的最大步数上限达到后停止并返回当前状态。streaming_callbackStreamingCallbackTNoneLLM 流式输出时的回调同一回调也可配置为在工具被调用时输出工具结果。raise_on_tool_invocation_failureboolFalse工具调用失败时是否抛出异常。为False时异常会被转换为一条聊天消息交给 LLM 继续处理。tool_invoker_kwargsdict[str, Any]None透传给 ToolInvoker 的额外关键字参数。异常约束TypeError当传入的chat_generator的run()方法不支持tools参数时抛出。ValueError当exit_conditions不合法时抛出。从当前仓库源码看构造时的工具支持校验发生在__init__中haystack/components/agents/agent.pyAgent通过inspect.signature(chat_generator.run)检查tools是否在参数列表里若传入了工具而生成器不支持会立刻抛出TypeError并给出明确提示。2.1 当前仓库中的参数演进从当前仓库源码可以确认Agent.__init__在后续版本中进一步扩展这些属于文档之后的演进供参考user_prompt可复用的用户提示模板支持 Jinja2 模板变量运行时追加到传入的 messages 之后。required_variables声明user_prompt/system_prompt中必须由运行期提供的模板变量缺省为*全部必填设为None则全部可选。tool_concurrency_limit并行执行工具调用的最大并发数默认4设为1可关闭并行工具执行。hooks在before_run、before_llm、before_tool、after_tool、on_exit、after_run等钩子点上注册的 Hook 列表钩子接收实时State并可通过修改它影响运行流程。三、快速上手文档最小示例与完整可运行示例3.1 参考文档的最小示例参考文档给出的最小用法如下from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.tools.tool import Tool tools [Tool(namecalculator, description...), Tool(namesearch, description...)] agent Agent( chat_generatorOpenAIChatGenerator(), toolstools, exit_conditions[search], ) # Run the agent result agent.run( messages[ChatMessage.from_user(Find information about Haystack)] ) assert messages in result # Contains conversation history要点exit_conditions[search]表示当模型调用了名为search的工具且执行成功后Agent 即结束本次运行。返回的result至少包含messages整个对话历史assert语句验证了这一点。3.2 完整实战示例搜索 计算的 Agent参考文档的最小示例只展示了骨架。结合源码类注释中的完整示例haystack/components/agents/agent.py下面是一个查询法国小费习惯 → 用计算器算小费的端到端可运行示例。它演示了用tool装饰器定义工具、通过Annotated描述参数、通过system_prompt约束行为以及用streaming_callback打印流式输出from typing import Annotated, Literal from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack.tools import tool tool def search(query: Annotated[str, The search query]) - str: Search for information on the web. # 实际项目中此处应调用真实搜索 API return In France, a 15% service charge is typically included, but leaving 5-10% extra is appreciated. tool def calculator( operation: Annotated[Literal[multiply, percentage], The mathematical operation to perform], a: Annotated[float, First number], b: Annotated[float, Second number], ) - float: Perform mathematical calculations. if operation multiply: return a * b elif operation percentage: return (a / 100) * b return 0 agent Agent( system_prompt( You are a helpful assistant. Use the search tool to find information about a users question and the calculator tool to perform math. ), chat_generatorOpenAIChatGenerator(), tools[search, calculator], streaming_callbackprint_streaming_chunk, ) result agent.run( messages[ChatMessage.from_user(Calculate the appropriate tip for an €85 meal in France)] ) # 获取最终回复 print(result[last_message].text)tool装饰器会把普通函数转换为Tool对象name、description来自函数名与 docstringparameters则根据带Annotated类型注解的函数签名自动生成符合 JSON Schema 的参数定义。Tool数据类的完整字段定义见 haystack/tools/tool.pyname、description、parameters、function同步函数、async_function协程函数、outputs_to_string、inputs_from_state、outputs_to_state。3.3 使用 Toolset 组织工具当工具数量变多时可以用Toolset把相关工具打包成一个集合传给 Agenthaystack/tools/toolset.pyfrom typing import Annotated from haystack.tools import tool, Toolset from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator tool def add(a: Annotated[int, first number], b: Annotated[int, second number]) - int: Add two numbers. return a b tool def subtract(a: Annotated[int, first number], b: Annotated[int, second number]) - int: Subtract b from a. return a - b math_toolset Toolset([add, subtract]) agent Agent(chat_generatorOpenAIChatGenerator(), toolsmath_toolset)Toolset实现了集合接口__iter__、__contains__、__len__、__getitem__因此可以像工具列表一样被Agent与聊天生成器消费。同时它也是动态工具加载的基类通过子类化Toolset并覆写warm_up()在启动时把工具赋给self.tools与to_dict()/from_dict()序列化端点描述符而非工具实例可以实现从 OpenAPI URL、MCP 服务器等外部来源动态加载工具。四、运行 Agentrun 与 run_async4.1 run 方法参考文档给出的Agent.run签名def run(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] None, *, break_point: Optional[AgentBreakpoint] None, snapshot: Optional[AgentSnapshot] None, system_prompt: Optional[str] None, tools: Optional[Union[list[Tool], Toolset, list[str]]] None, **kwargs: Any) - dict[str, Any]参数说明以参考文档为准参数说明messages要处理的 HaystackChatMessage列表。streaming_callbackLLM 流式输出时的回调同一回调可配置为在工具被调用时输出工具结果。break_point一个AgentBreakpoint可以是针对chat_generator的Breakpoint或针对tool_invoker的ToolBreakpoint。snapshot之前保存的 Agent 执行快照包含从上次中断处恢复执行所需的全部信息。system_prompt本次运行的系统提示词提供时覆盖默认系统提示词。tools本次运行使用的工具Tool列表、Toolset或工具名字符串列表。传工具名时会从 Agent 构造时配置的工具中按名选取。kwargs传给 State schema 的额外数据键必须与state_schema中定义的一致。异常RuntimeError调用run()前未对 Agent 执行warm_up()。BreakpointException触发了 agent breakpoint。返回值字典messagesAgent 运行期间交换的全部消息列表。last_message运行期间交换的最后一条消息。以及state_schema中定义的任何额外键。4.2 run_async 方法async def run_async(messages: list[ChatMessage], streaming_callback: Optional[StreamingCallbackT] None, *, break_point: Optional[AgentBreakpoint] None, snapshot: Optional[AgentSnapshot] None, system_prompt: Optional[str] None, tools: Optional[Union[list[Tool], Toolset, list[str]]] None, **kwargs: Any) - dict[str, Any]run_async是run的异步版本遵循相同逻辑但在可能的地方使用异步操作例如若 ChatGenerator 提供了run_async方法则优先调用它。对于同步的生成器当前仓库通过_execute_component_async将调用派发到工作线程执行asyncio.to_thread并保持 tracing span 上下文工具的invoke_async在无async_function时也会回退到asyncio.to_thread运行同步函数见 haystack/tools/tool.py。4.3 运行结果的运行时元数据当前仓库参考文档的返回值说明聚焦于messages、last_message与state_schema键。从当前仓库源码haystack/components/agents/agent.py可以看到Agent 会在运行期自动向 State 中写入一批运行元数据键并以输出形式暴露可用于下游路由与监控输出键类型含义step_countint已执行的步数。一步 一次 chat-generator 调用 该次调用中模型请求的所有工具调用若有的执行。token_usagedict本次运行中所有 LLM 调用的 token 用量聚合由每条 LLM 消息的meta[usage]累加而来。tool_call_countsdict[str, int]各工具被调用的次数映射。exit_reasonstrAgent 停止的原因可用于ConditionalRouter之类的下游路由。取值包括text模型返回了无工具调用的完整回复、length/content_filter模型返回了不完整回复、满足工具退出条件的工具名此时last_message是该工具的结果、max_agent_steps达到步数上限、或钩子通过stop_run状态键提供的自定义原因。这些键由常量_RUN_METADATA_STATE_KEYS定义属于保留键用户不能在自己的state_schema中重定义它们。Agent 内部还有一组完全私有的_INTERNAL_STATE_KEYS如continue_run、stop_run、tools、hook_context、context_tokens它们不暴露为输入或输出。五、退出条件exit_conditions机制详解参考文档指出退出条件可以是text模型生成无工具调用的消息时返回也可以是工具名该工具执行完成后返回多个条件可同时指定默认是[text]。从源码看退出判定有两套逻辑haystack/components/agents/agent.py1. 模型退出原因判定_get_model_exit_reason当最后一条消息来自 assistant 且不含工具调用时若meta[finish_reason]为length返回length若为content_filter返回content_filter若消息含文本返回text空响应且无终止原因时不退出——这保留了 Agent 对 ChatGenerator 丢弃的畸形工具调用的恢复能力。2. 工具退出条件判定_check_exit_conditions当exit_conditions不只包含text时遍历 LLM 消息中的每个工具调用只要模型调用了至少一个在exit_conditions中列出的工具且该工具执行未出错就返回该工具名作为退出原因若被调用的退出条件工具执行出错则取消退出即使同一步中有其他退出条件工具成功多个退出条件工具在同一步被并行调用时返回第一个遇到的工具名。测试用例test/components/agents/test_agent.py大量验证了这类组合场景例如exit_conditions[text, weather_tool]、exit_conditions[weather_tool, search]等配置。六、StateAgent 与工具共享的运行时状态容器参考文档用专门一节介绍了State。它是在 Agent 及其工具执行期间存储共享信息的容器例如文档、上下文和中间结果都可以放入其中使 Agent 与工具能够读写同一份上下文。6.1 schema 结构与合并语义State内部包装了一个由schema定义的_data字典每个 schema 条目形如parameter_name: { type: SomeType, # 期望的类型 handler: Optional[Callable[[Any, Any], Any]] # 合并/更新函数 }handler 控制set()时的合并方式列表类型默认使用merge_lists拼接列表其他类型默认使用replace_values用新值覆盖旧值。参考文档特别强调一个messages字段类型list[ChatMessage]会被自动加入 schema这正是 Agent 能持续读写同一对话上下文的机制。6.2 State 使用示例参考文档示例from haystack.components.agents.state import State my_state State( schema{gh_repo_name: {type: str}, user_name: {type: str}}, data{gh_repo_name: my_repo, user_name: my_user_name} )6.3 State 的完整 API参考文档给出了以下方法签名与说明__init__def __init__(schema: dict[str, Any], data: Optional[dict[str, Any]] None)schema参数名到类型与 handler 配置的映射。type必须是合法的 Python 类型handler必须是可调用对象或None。handler 为None时使用类型默认 handler列表类型为haystack.agents.state.state_utils.merge_lists其他类型为haystack.agents.state.state_utils.replace_values。data可选的初始数据字典。get(key, defaultNone)按 key 读取值未找到时返回default。set(key, value, handler_overrideNone)按 schema 规则写入或合并值。合并规则为若给了handler_override则使用它否则使用 schema 中该 key 定义的 handler。datapropertyState 当前的全部数据。has(key)判断 key 是否存在于 state 中返回布尔值。to_dict()将 State 序列化为字典。from_dict(data)classmethod从字典反序列化回 State 对象。6.4 源码级深入校验、默认 handler 与读写语义从 haystack/components/agents/state/state.py 可以确认几个重要实现细节schema 校验构造时_validate_schema会检查每个条目必须有type字段、type必须是合法 Python 类型支持普通类、list[str]等泛型、Union/Optional联合类型、handler 必须可调用或为None特别地messages键的类型必须是list[ChatMessage]。默认 handler 注入构造时会对未显式指定 handler 的条目按类型补上默认 handler列表用merge_lists其余用replace_values实现位于 haystack/components/agents/state/state_utils.py。merge_lists会把非列表值包装成列表后拼接current为None视为空列表replace_values则直接返回新值。set()的健壮性若写入的 key 不在 schema 中set()会抛出ValueErrorget()返回的是值的深拷贝避免外部意外修改内部数据。Agent 集成在Agent.__init__中用户提供的state_schema会被浅拷贝进resolved_state_schema若没有messages键则自动补充{type: list[ChatMessage], handler: merge_lists}再叠加上运行元数据键与内部键最终据此生成组件的输入/输出 socket见 haystack/components/agents/agent.py。序列化to_dict()序列化 schema 中的类型serialize_type与 handler 可调用serialize_callable并用字段级回退保证单个不可序列化的值不会拖垮整个 State 的序列化from_dict()相应地进行反序列化见 haystack/components/agents/state/state.py。6.5 工具与 State 的双向数据流State 之所以重要是因为工具可以通过Tool的inputs_from_state与outputs_to_state字段直接与 Agent 共享数据haystack/tools/tool.pyinputs_from_state把 State 键映射到工具参数名。例如{repository: repo}表示把 State 中的repository值传给工具的repo参数。工具构造时会校验这些参数名确实存在于工具的函数签名或 JSON schema 中。outputs_to_state把工具输出映射回 State 键并可附带 handler。例如{documents: {source: docs, handler: custom_handler}}表示把工具结果字典中docs字段经custom_handler处理后写入 State 的documents键省略source时整个工具结果传给 handler。在工具执行层haystack/components/agents/tool_calling.py_merge_tool_outputs_into_state负责把工具输出按outputs_to_state配置写入 State_build_tool_result_message/_process_tool_output则根据outputs_to_string把工具结果转换为字符串默认对非字符串结果做 JSON 序列化失败时回退到str()或按raw_result原样返回用于图片等TextContent/ImageContent场景。七、warm_up 与序列化7.1 warm_updef warm_up() - None参考文档说明其作用为Warm up the Agent。从源码看haystack/components/agents/agent.pywarm_up()会依次完成三件事预热全部工具warm_up_tools、预热钩子warm_up_hooks、预热底层 chat generator若其有warm_up方法。run()内部会在正式执行前调用warm_up()。工具与 Toolset 的warm_up()约定是幂等的——例如动态加载工具的 Toolset 子类应通过自身状态做保护if self._client is not None: return因为预热可能在每次运行前被调用。7.2 to_dict 与 from_dictdef to_dict() - dict[str, Any] def from_dict(cls, data: dict[str, Any]) - Agent参考文档说明to_dict将组件序列化为字典from_dict从字典反序列化。这正是 Haystack Pipeline 的 YAML 序列化机制的基础——Agent 可以被嵌入 Pipeline并整体导出/导入。从源码看haystack/components/agents/agent.pyto_dict()会序列化chat_generator组件字典、tools工具或 Toolset 的序列化形式、system_prompt、exit_conditions、state_schema类型与 handler 可调用均被序列化、max_agent_steps、streaming_callback可调用序列化、raise_on_tool_invocation_failure等from_dict()则按需反序列化 chat generator、state_schema、streaming_callback、tools、hooks。State自身的to_dict()/from_dict()与此同理见 haystack/components/agents/state/state.py。八、运行循环的源码级剖析虽然参考文档没有画出运行循环细节但从当前仓库源码haystack/components/agents/agent.py可以完整还原 Agent 的一次运行过程run()首先调用warm_up()然后通过_initialize_fresh_execution构建执行上下文把kwargs中与state_schema匹配的键填充进State初始化messages、step_count、token_usage、tool_call_counts、exit_reason等状态解析流式回调并组装 chat generator 与工具执行的输入。进入while exe_context.counter self.max_agent_steps主循环每轮执行_run_step先重新展平工具让SearchableToolset这类动态工具集能持续暴露新发现的工具并做重名检查运行before_llm钩子检查stop_run状态键是否要求提前终止调用chat_generator.run(messages..., tools...)把回复写入 State并记录 token 用量与上下文 token 数若无工具或模型产出了无工具调用的终止性回复则按退出原因结束本步走on_exit钩子判定是否继续否则运行before_tool钩子重新读取 State 中最后一条消息的待执行工具调用钩子可以改写、拒绝这些调用调用_run_tool并发执行工具并发上限为tool_concurrency_limit把工具结果消息写回 State记录各工具调用次数运行after_tool钩子最后检查工具退出条件满足则设置exit_reason并结束。若循环因max_agent_steps耗尽而自然结束未break则设置exit_reason max_agent_steps并记录警告日志。运行after_run钩子从 State 中剔除内部键后构建返回值messages、last_message取消息列表最后一条、运行元数据以及state_schema中定义的所有键。整个循环还被 Haystack 的 tracing 体系包裹每次运行会产生haystack.agent.runspan每步产生haystack.agent.stepspanLLM 调用与工具执行分别有各自的子 span便于在监控系统中观测 Agent 的逐步行为。九、注意事项与边界条件聊天生成器必须支持工具为 Agent 配置工具时chat_generator.run()必须接受tools参数否则构造时即抛出TypeError。若在运行期传入工具而生成器不支持同样会抛错。无工具即纯 LLMtools为空时Agent 生成一条回复后立即退出行为等价于 ChatGenerator。步数上限是硬边界max_agent_steps默认 100保证即使模型反复请求工具调用也不会无限循环达到上限时返回当前状态并以max_agent_steps作为exit_reason。一步的定义是一次 LLM 调用 该次调用请求的全部工具调用执行。工具失败的可恢复性raise_on_tool_invocation_failureFalse默认时工具调用失败不会中断运行而是把异常转换为聊天消息交回给 LLM让模型可以换一种方式继续设为True则直接抛出。State 的键约束写入 State 的 key 必须存在于 schema 中否则set()抛ValueErrormessages类型必须是list[ChatMessage]运行元数据键与内部键为保留键不可在state_schema中重定义。快照与断点通过run(snapshot...)可以从保存的执行快照恢复运行break_point可在 chat_generator 或 tool_invoker 处暂停执行这为调试与人工介入Human-in-the-Loop提供了基础。Warm-up 前置条件虽然run()内部会自动预热但若绕过run()直接依赖组件状态如把 Agent 嵌入 Pipeline应保证 Pipeline 在运行前完成 warm-up。十、总结参考文档围绕两个核心类勾勒了 Haystack Agent 的完整能力Agent工具使用、退出条件、同步/异步运行、序列化与Stateschema 化共享状态、类型化合并语义。结合当前仓库源码可以看到Agent 本质上是一个LLM 循环控制器通过 State 把对话消息、工具调用、运行元数据与用户自定义上下文统一管理通过exit_conditions精确控制终止时机并通过Tool/Toolset机制把任意函数、组件乃至外部服务接入模型推理回路。无论是构建 RAG Agent、多工具编排 Agent还是将 Agent 嵌入 Haystack PipelineAgentState都是值得首先掌握的组件组合。相关测试test/components/agents/test_agent.py、test/components/agents/test_state_class.py提供了大量可对照的行为示例可作为进一步学习的起点。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考