Haystack Agents API 完全指南构建工具调用型 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 仓库docs-website/reference_versioned_docs/version-2.19/haystack-api/agents_api.md为骨架深入讲解haystack.components.agents中Agent与State两大核心类的完整 API如何配置工具、设定退出条件、管理跨工具共享状态、同步与异步运行并结合 agent.py、state.py 等源码实现剖析其底层执行机制。读完本文你将能够在 Haystack 2.19 中独立搭建带工具调用的多轮 Agent、自定义运行状态 schema并理解其运行循环与退出机制的工作原理。Agent 组件是什么Agent是一个 Haystack 组件实现了一个支持工具调用的智能体tool-using agent并提供与具体模型供应商无关provider-agnostic的聊天模型支持。它的核心工作方式是接收消息列表list[ChatMessage]作为输入循环执行调用聊天生成器 → 解析工具调用 → 执行工具 → 把工具结果写回对话这一流程直到满足一个**退出条件exit condition**才返回结果。退出条件可以由两种方式触发模型直接返回一段文本回答不包含工具调用或者调用了某个被特别指定的工具。多个退出条件可以同时指定Agent 会在任一条件满足时停止。一个特殊但常见的使用场景当你不给 Agent 配置任何工具时它退化为一个普通的ChatGenerator——生成一次回复后立即退出。这使 Agent 既可以作为完整的多工具智能体使用也可以作为普通对话组件嵌入 pipeline。官方用法示例以下示例来自关联文档展示了最基础的用法——配置两个工具calculator与search并把search设为退出条件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]时Agent 会一直循环执行直到模型调用search工具完成搜索后立即返回——此时last_message就是search工具的执行结果。Agent 构造函数参数详解Agent.__init__的完整签名如下摘自 agent.py 源码实际实现比文档多了user_prompt、required_variables、tool_concurrency_limit等参数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 使用的聊天生成器实例其run方法必须支持tools参数。toolslist[Tool]/ToolsetNoneAgent 可用的工具列表或工具集Toolset。system_promptstrNoneAgent 的系统提示词。exit_conditionslist[str][text]触发 Agent 返回的条件列表。可包含text模型生成无工具调用的消息即返回或具体工具名该工具被执行后即返回。state_schemadict[str, Any]None工具运行时共享状态State的 schema 定义。max_agent_stepsint100Agent 运行的最大步数上限超过后停止并返回当前状态。一步指一次聊天生成器调用及其触发的所有工具执行。streaming_callbackStreamingCallbackTNoneLLM 流式输出回调同一回调也可用于在工具被调用时输出工具结果。raise_on_tool_invocation_failureboolFalse工具调用失败时是否抛出异常。为False时异常会被转换为聊天消息回传给 LLM让模型自行纠错。tool_invoker_kwargsdict[str, Any]None透传给底层ToolInvoker的额外关键字参数。异常情况TypeErrorchat_generator的run方法不支持tools参数。ValueErrorexit_conditions不合法。从源码看参数校验与默认行为在 agent.py 中构造函数会做如下关键校验工具支持检查通过inspect.signature(chat_generator.run).parameters判断生成器是否支持tools参数。如果配置了工具但生成器不支持立即抛出TypeError。退出条件默认值exit_conditions为None时自动设置为[text]。保留键保护step_count、token_usage、tool_call_counts、exit_reason运行元数据以及continue_run、stop_run、tools、hook_context、context_tokens内部状态键均为保留键用户自定义state_schema时不可重定义否则抛出ValueError。并发工具上限源码中的tool_concurrency_limit默认4设为1可禁用并行工具执行。Agent 的运行方法run 与 run_asyncrun 方法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定义匹配。返回值是一个字典包含messages本次运行期间交换的全部消息last_message最后一次交换的消息state_schema中定义的其他键如step_count、token_usage、tool_call_counts、exit_reason。异常RuntimeError运行前未执行warm_up()BreakpointException触发 Agent 断点时抛出。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。从 agent.py 的实现看对于仅支持同步调用的生成器会通过_execute_component_async使用asyncio.to_thread派发到线程池执行同时保持 tracing 上下文不丢失。返回键与run完全一致。从源码看运行循环run/run_async内部通过_run_step/_run_step_async执行主循环参见 agent.py 中while exe_context.counter self.max_agent_steps部分每个 step 的流程为将当前可用工具展平flatten并暴露到 State供 hook 读取执行before_llmhook若某个 hook 通过stop_run键请求停止则在调用 LLM 前结束运行调用聊天生成器把回复写入 State并记录 token 用量与上下文规模若模型产出了无工具调用的终结回复或根本未配置工具则记录exit_reason并触发on_exithook 判定是否继续否则执行before_toolhook读取待执行的工具调用消息调用_run_tool执行工具把工具结果写入消息列表执行after_toolhook通过_check_exit_conditions检查工具型退出条件是否满足步数加一循环直至退出或达到max_agent_steps。值得注意的实现细节来自 agent.py 的_get_model_exit_reason与_check_exit_conditions当最后一条消息的finish_reason为length或content_filter时Agent 会以对应原因退出便于调用方区分部分回复与完整回答工具型退出条件只有在被调用且未出错时才触发退出若被指定为退出条件的工具执行出错则取消本次退出让 Agent 继续循环。StateAgent 与工具之间的共享状态容器State是 Agent 及其工具在运行期间存储共享信息的容器例如文档、上下文和中间结果。它由haystack.components.agents.state导出。设计原理State内部封装了一个由schema定义的_data字典。schema 中每一项的格式为parameter_name: { type: SomeType, # expected type handler: Optional[Callable[[Any, Any], Any]] # merge/update function }handler决定了使用set()方法时新值与旧值的合并策略列表类型默认使用merge_lists拼接两个列表其他类型默认使用replace_values直接用新值覆盖旧值。State会自动向 schema 追加一个messages字段类型为list[ChatMessage]其 handler 为merge_lists见 state.py 的__init__。这使得 Agent 与其工具能够读写同一个上下文。官方用法示例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} )方法速查方法签名说明__init__(schema: dict[str, Any], data: Optional[dict[str, Any]] None)以 schema 和可选初始数据初始化 State。schema 的type必须是合法的 Python 类型handler必须是可调用对象或None为None时按类型选用默认 handler。get(key: str, default: Any None) - Any按键取值键不存在时返回default。源码中通过deepcopy返回避免外部修改影响内部状态。set(key: str, value: Any, handler_override: Optional[Callable] None) - None按键写入或合并值。优先使用handler_override否则使用 schema 中为该键定义的 handler。若键不在 schema 中抛出ValueError。dataproperty - dict[str, Any]返回 State 的全部当前数据。has(key: str) - bool检查某个键是否存在于 State 中。to_dict() - dict[str, Any]将 State 对象序列化为字典schema 中的类型与 handler 会分别经过类型/可调用对象序列化单个不可序列化的字段只会被跳过不会导致整体失败。from_dictclassmethod (data: dict[str, Any]) - State从字典反序列化回 State 对象。从源码看 schema 校验与默认 handler在 state.py 中_validate_schema会检查每个参数定义必须包含type键且必须是合法的 Python 类型普通类、泛型如list[str]、Union/Optional 均可_is_valid_type的实现见 state_utils.pyhandler必须可调用或为None若定义messages键其类型必须为list[ChatMessage]。默认 handler 的选定逻辑为_is_list_type(definition[type])为真时使用merge_lists否则使用replace_values。两个默认 handler 的实现位于 state_utils.pymerge_lists(current, new)把current与new都规范化为列表后拼接。current为None时视为空列表单个值会自动包装成列表。replace_values(current, new)直接返回new即覆盖旧值。序列化与反序列化Agent与State都支持to_dict/from_dict便于在 Haystack pipeline 中保存、加载与 YAML 描述。Agent.to_dict()返回包含全部初始化参数的序列化字典chat_generator、tools、state_schema、streaming_callback、hooks等均会被递归序列化见 agent.py 的to_dict实现。Agent.from_dict(cls, data)从字典反序列化会先就地反序列化chat_generator、tools、state_schema把序列化的类型/回调还原为 Python 对象再调用默认反序列化逻辑。state_schema为None时安全处理为{}相关行为在 test_agent.py 的test_from_dict_state_schema_none中有验证。State.to_dict()返回{schema: ..., data: ...}结构State.from_dict()则反向恢复。对应测试见 test_state_class.py 与 test_agent.py覆盖了 schema 解析、保留键保护、退出条件校验、序列化往返等场景。完整实战搜索 计算的多工具 Agent综合官方文档示例与 agent.py 中的进阶示例下面是一个更完整的可运行脚本——Agent 先搜索法国小费习俗再用计算器算出 €85 餐费的建议小费最后以纯文本回答退出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. 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, max_agent_steps10, ) result agent.run( messages[ChatMessage.from_user(Calculate the appropriate tip for an €85 meal in France)] ) print(result[last_message].text) # 最终回答 print(result[tool_call_counts]) # 各工具调用次数 print(result[exit_reason]) # 退出原因如 text要点回顾tools中的tool装饰器会从函数签名自动生成工具 schema工具类定义见 tool.py未显式设置exit_conditions时默认[text]模型给出不含工具调用的完整回答即退出运行结果中的tool_call_counts与exit_reason由 Agent 自动写入 State 并作为输出返回可用于下游路由判断例如配合ConditionalRouter按退出原因分流。使用 user_prompt 模板变量当 Agent 需要嵌入 pipeline 或频繁更换输入时可以用 Jinja2 模板定义user_prompt模板变量自动成为run方法的输入此功能在 2.19 源码中已支持agent Agent( chat_generatorOpenAIChatGenerator(), tools[translate], system_promptYou are a helpful translation assistant., user_prompt{% message roleuser%} Translate the following document to {{ language }}: {{ document }} {% endmessage %}, ) result agent.run( messages[], languageFrench, documentThe weather is lovely today and the sun is shining., )进阶机制hooks 与运行时动态工具从 agent.py 的类文档与实现可以看出2.19 的 Agent 还内置了强大的扩展机制Hook 点before_run每次运行一次可改写初始消息或播种状态、before_llm每次 LLM 调用前、before_tool工具执行前可改写或拒绝工具调用、after_tool工具执行后可改写工具结果消息如脱敏、截断、摘要、on_exit即将退出时可通过state.set(continue_run, True)让 Agent 继续运行、after_run运行结束、构建返回值之前。动态工具集_select_tools支持在run时以工具名列表选取初始化配置中的部分工具且每次运行都会重新 spawn 工具集避免并发运行时共享同一 Toolset 导致状态串扰。Tracing每次运行会创建haystack.agent.run与haystack.agent.step等 trace span记录最大步数、工具列表、退出条件与状态 schema 等标签便于观测与调试。总结本文围绕agents_api.md完整梳理了 Haystack 2.19 中Agent与State的 API 与实现Agent通过生成 → 工具调用 → 结果回写的循环实现多步智能体退出条件既可以是纯文本回答也可以是特定工具的调用不配工具时Agent即普通ChatGenerator一套组件覆盖两种形态State以 schema handler 机制实现跨工具的共享上下文列表默认拼接、其余默认覆盖且自动维护messages字段run/run_async提供同步与异步两种执行路径返回messages、last_message及全部 state_schema 键序列化接口让 Agent 可以无缝嵌入 Haystack pipeline配合 hooks、动态工具与 tracing 可构建生产级的工具调用智能体。如需深入阅读源码可重点查看 agent.pyAgent 实现与运行循环、state.py 与 state_utils.pyState 与合并策略、tool.pyTool 定义以及 test_agent.py、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），仅供参考