Haystack Experimental Agent 完全指南Tool-Using Agent 与 Human-in-the-Loop 确认策略实战【免费下载链接】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_experimental.components.agents.Agent是 Haystack 生态中面向实验特性提供的工具型智能体组件它内置循环调用 LLM → 请求工具 → 执行工具 → 检查退出条件的主循环并在此基础上扩展了对 human-in-the-loop人在回路确认策略的原生支持——你可以在工具真正执行前插入总是询问从不询问仅询问一次等确认策略甚至借助BreakpointConfirmationStrategy将执行暂停并序列化为快照交给异步环境中的用户事后审批。读完本文你将掌握该 Agent 的完整构造参数、运行与恢复机制、HITL 三要素策略、UI、决策的协作方式以及如何将它与 Haystack 标准组件、状态模式和序列化能力整合进生产级 RAG / 多工具工作流。说明本文面向的 API 参考文档位于 experimental_agents_api.md文中的实现证据均来自当前仓库的haystack/源码。一、Agent 是什么定位与核心行为haystack_experimental.components.agents.agent.Agent是一个实现了带工具调用能力的智能体的 Haystack 组件其核心特点是chat model 提供方无关provider-agnostic——只要传入的 ChatGenerator 支持tools参数即可驱动 Agent 完成多轮工具调用。从源码看标准 Agent 类的定义位于 haystack/components/agents/agent.py其 docstring 明确描述了行为Agent 处理消息并调用工具直到满足一个退出条件exit condition退出条件既可以是模型产出了一段不再附带工具调用的文本也可以是执行了某个指定工具可以同时指定多个退出条件当不传任何工具时Agent 退化为一个 ChatGenerator生成一条回复后立即结束。在实验版中该 Agent 扩展了 Haystack 标准 Agent专门增加了 human-in-the-loop 确认策略支持见文档中对 Agent 类的 NOTE 说明。一次 Agent 运行的核心循环从 agent.py 的_run_step实现可以还原标准主循环每个step包含将当前可用的工具列表重新展平写入运行时状态state.data[tools]供before_tool类钩子如ConfirmationHook读取执行before_llm钩子然后调用chat_generator.run(messages..., tools...)获得 LLM 回复若模型回复是无工具调用的终结性文本或finish_reason为length/content_filter则触发text退出否则执行before_tool钩子——human-in-the-loop 确认逻辑正是在这一步注入——再从state.data[messages]中重读待执行的工具调用调用工具执行器_run_tool写入工具结果消息再执行after_tool钩子检查工具退出条件决定继续循环还是停止。整个循环受max_agent_steps限制超过步数上限时 Agent 停止并返回当前状态exit_reason记为max_agent_steps相关常量见 agent.py。二、快速上手一个带确认策略的最小示例文档给出的完整示例可在haystack_experimental中直接运行如下from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.tools.tool import Tool from haystack_experimental.components.agents import Agent from haystack_experimental.components.agents.human_in_the_loop import ( HumanInTheLoopStrategy, AlwaysAskPolicy, NeverAskPolicy, SimpleConsoleUI, ) calculator_tool Tool(namecalculator, descriptionA tool for performing mathematical calculations., ...) search_tool Tool(namesearch, descriptionA tool for searching the web., ...) agent Agent( chat_generatorOpenAIChatGenerator(), tools[calculator_tool, search_tool], confirmation_strategies{ calculator_tool.name: HumanInTheLoopStrategy( confirmation_policyNeverAskPolicy(), confirmation_uiSimpleConsoleUI() ), search_tool.name: HumanInTheLoopStrategy( confirmation_policyAlwaysAskPolicy(), confirmation_uiSimpleConsoleUI() ), }, ) # Run the agent result agent.run( messages[ChatMessage.from_user(Find information about Haystack)] ) assert messages in result # Contains conversation history这段代码演示了两个关键点confirmation_strategies按工具名tool.name映射策略calculator使用NeverAskPolicy从不询问、直接执行search使用AlwaysAskPolicy每次执行前都向用户确认agent.run()返回的字典中一定包含messages键保存完整的对话历史。更贴近当前仓库实际 API 的等价写法是使用BlockingConfirmationStrategy位于 haystack/hooks/human_in_the_loop/strategies.py并通过ConfirmationHook注册到 Agent 的before_tool钩子点上from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.hooks.human_in_the_loop import ( AlwaysAskPolicy, BlockingConfirmationStrategy, ConfirmationHook, NeverAskPolicy, SimpleConsoleUI, ) from haystack.tools import tool tool def delete_file(path: str) - str: Delete the file at the given path. return fDeleted {path}. hook ConfirmationHook( confirmation_strategies{ delete_file: BlockingConfirmationStrategy( confirmation_policyNeverAskPolicy(), confirmation_uiSimpleConsoleUI() ) } ) agent Agent(chat_generatorOpenAIChatGenerator(), tools[delete_file], hooks{before_tool: [hook]})ConfirmationHook的完整定义见 haystack/hooks/human_in_the_loop/hooks.py它被限制只能注册在before_tool钩子点allowed_hook_points (before_tool,)若注册到其他钩子点Agent 在构造时会直接抛出ValueError。三、Agent 构造参数全解析Agent.__init__的完整签名来自实验版 API 文档def __init__(*, chat_generator: ChatGenerator, tools: ToolsType | None None, system_prompt: str | None None, exit_conditions: list[str] | None None, state_schema: dict[str, Any] | None None, max_agent_steps: int 100, streaming_callback: StreamingCallbackT | None None, raise_on_tool_invocation_failure: bool False, confirmation_strategies: dict[str, ConfirmationStrategy] | None None, tool_invoker_kwargs: dict[str, Any] | None None, chat_message_store: ChatMessageStore | None None, memory_store: MemoryStore | None None) - None各参数含义结合仓库源码逐项说明参数类型默认值说明chat_generatorChatGenerator必填Agent 使用的聊天生成器其run方法必须支持tools参数否则构造时抛出TypeError见 agent.pytoolsToolsType \| NoneNone可供 Agent 使用的Tool对象列表或Toolset为None时 Agent 退化为纯 ChatGeneratorsystem_promptstr \| NoneNoneAgent 的系统提示词可为普通字符串或 Jinja2 消息模板exit_conditionslist[str] \| None[text]退出条件列表包含text表示模型生成无工具调用文本即返回也可填入工具名表示该工具执行完毕后返回。不合法值会抛ValueErrorstate_schemadict[str, Any] \| NoneNone工具共享的运行时状态 schema每个键对应一个类型配置含type与可选的handler工具可通过inputs_from_state/outputs_to_state读写max_agent_stepsint100最大步数上限一个 step 一次生成 该轮所有工具调用超限即停止并返回当前状态streaming_callbackStreamingCallbackT \| NoneNoneLLM 流式输出回调同一回调也可配置为在工具调用时输出工具结果raise_on_tool_invocation_failureboolFalse工具调用失败时是否抛异常为False时异常被转换为聊天消息回传给 LLM 继续处理confirmation_strategiesdict[str, ConfirmationStrategy] \| NoneNone实验版扩展按工具名映射 human-in-the-loop 确认策略tool_invoker_kwargsdict[str, Any] \| NoneNone透传给底层ToolInvoker的额外关键字参数chat_message_storeChatMessageStore \| NoneNone用于存取聊天历史的存储组件实验版扩展memory_storeMemoryStore \| NoneNone用于存取记忆的存储组件实验版扩展关于state_schema的源码细节标准 Agent 在构造时会为state_schema自动补充几个保留键见 agent.pymessages类型list[ChatMessage]handler 为merge_lists用于保存对话历史运行元数据键step_count、token_usage、tool_call_counts、exit_reason仅作为输出暴露不可在用户自定义 schema 中重复定义内部控制键continue_run、stop_run、tools、hook_context、context_tokens纯内部状态既不是输入也不是输出。如果用户自定义的state_schema中使用了这些保留键构造时会抛出ValueError并列出保留键清单。四、run 与 run_async参数、返回与异常run 方法签名def run(messages: list[ChatMessage], streaming_callback: StreamingCallbackT | None None, *, generation_kwargs: dict[str, Any] | None None, break_point: AgentBreakpoint | None None, snapshot: AgentSnapshot | None None, system_prompt: str | None None, tools: ToolsType | list[str] | None None, confirmation_strategy_context: dict[str, Any] | None None, chat_message_store_kwargs: dict[str, Any] | None None, memory_store_kwargs: dict[str, Any] | None None, **kwargs: Any) - dict[str, Any]关键参数messages待处理的ChatMessage列表通常以ChatMessage.from_user(...)开头streaming_callback运行时覆盖初始化时的流式回调select_streaming_callback会优先取运行时值generation_kwargs透传给 LLM 的额外生成参数按 key 合并时以运行时传入值为准见 agent.pybreak_pointAgentBreakpoint可为针对chat_generator的Breakpoint或针对tool_invoker的ToolBreakpoint触发时会抛出BreakpointExceptionsnapshot先前保存的 Agent 执行快照字典包含从上次中断处恢复执行所需的全部信息system_prompt若提供则覆盖默认系统提示词tools本次运行的临时工具集——可以是Tool列表、Toolset也可以是工具名字符串列表此时从 Agent 初始化时配置的工具中按名选取见_select_tools实现 agent.pyconfirmation_strategy_context传给确认策略的请求级资源字典适用于 Web/服务端环境用于传递 WebSocket 连接、异步队列、Redis pub/sub 客户端等非阻塞交互所需对象chat_message_store_kwargs传给ChatMessageStore的关键字参数例如chat_history_id与last_k用于按历史 ID 取最近 K 条消息memory_store_kwargs传给MemoryStore的参数包含user_id/run_id/agent_id按用户 / 运行 / 智能体维度检索并追加记忆search_criteriasearch_memories的参数字典可含filters过滤条件、query检索查询若传入则忽略传给 Agent 的用户查询、top_k返回记忆条数、include_memory_metadata是否把记忆元数据放进ChatMessagekwargs与state_schema中定义键匹配的额外数据会被注入运行时状态。返回字典run返回的字典包含messages整个运行期间交换的全部消息last_message最后一条消息从源码看它是messages列表的最后一个元素见 agent.pystate_schema中定义的所有额外键外加运行元数据step_count、token_usage、tool_call_counts、exit_reason。exit_reason的取值用于下游路由如配合ConditionalRoutertext模型产出无工具调用的完整回复、length/content_filter模型产出不完整回复、触发工具退出条件的工具名此时last_message为该工具的结果、max_agent_steps达到步数上限或钩子通过stop_run状态键提供的自定义原因。异常RuntimeErrorAgent 组件在调用run()前未完成 warm-upBreakpointException触发了 Agent 断点break_point。run_asyncrun_async与run逻辑一致只是尽可能使用异步路径当 ChatGenerator 提供run_async时直接调用否则通过_execute_component_async调度到线程中执行见 agent.py。它接受的参数与run相同streaming_callback为异步回调也抛出同样的异常。五、HITL 确认机制的三层架构human-in-the-loop 确认机制围绕三个抽象展开它们的协议定义位于 haystack/hooks/human_in_the_loop/types/protocol.py1. ConfirmationPolicy确认策略何时询问ConfirmationPolicy协议定义should_ask(tool_name, tool_description, tool_params) - bool仓库内置实现位于 haystack/hooks/human_in_the_loop/policies.py策略行为AlwaysAskPolicy总是询问should_ask恒返回TrueNeverAskPolicy从不询问should_ask恒返回False直接放行AskOncePolicy每个工具相同参数仅询问一次内部记录已确认的(tool_name, tool_params)对update_after_confirmation在用户选择 confirm 后记忆该组合后续同参数调用不再打扰用户2. ConfirmationUI确认界面如何呈现ConfirmationUI协议定义get_user_confirmation(tool_name, tool_description, tool_params) - ConfirmationUIResult。仓库在 haystack/hooks/human_in_the_loop/user_interfaces.py 提供了两种实现SimpleConsoleUI简单的控制台询问文档示例中使用RichConsoleUI基于rich库的富控制台界面需要pip install rich见该文件的LazyImport以面板展示工具名、描述与参数支持y确认/n拒绝可附反馈/m修改参数逐字段提示输入非字符串类型按 JSON 解析三种交互并带线程锁_ui_lock保证并发安全。ConfirmationUIResult是一个数据类字段包括actionconfirm/reject/modify、feedback用户反馈文本、new_tool_params修改后的参数。3. ConfirmationStrategy确认策略对象组合决策ConfirmationStrategy协议定义run(...) - ToolExecutionDecision与run_async(...)。核心实现是BlockingConfirmationStrategystrategies.py其run流程为调用confirmation_policy.should_ask(...)返回False则直接生成executeTrue的ToolExecutionDecision需要确认时调用confirmation_ui.get_user_confirmation(...)将 UI 结果回传给策略的update_after_confirmation供其记忆学习根据action生成决策reject→executeFalse附带模板生成的拒绝反馈默认模板REJECTION_FEEDBACK_TEMPLATE Tool execution for {tool_name} was rejected by the user.可自定义reject_templatemodify→executeTruefinal_tool_params替换为用户修改后的参数并附带修改说明模板MODIFICATION_FEEDBACK_TEMPLATE含{tool_name}与{final_tool_params}占位符confirm→executeTrue原样执行。BlockingConfirmationStrategy支持三种反馈模板定制reject_template、modify_template、user_feedback_template后者含{feedback}占位符。决策的应用拒绝、修改与历史重写策略产生的ToolExecutionDecision会被_apply_tool_execution_decisionsstrategies.py应用到对话历史上reject向历史插入assistant 工具调用消息 带errorTrue的 tool 结果消息对把拒绝反馈喂回给 LLMmodify在工具调用消息前插入一条 user 消息解释参数被修改的原因否则 LLM 不知道参数为何变化可能再次用原参数调用最终由_update_chat_history将拒绝/修改消息插入到对话中最后一个 user 或 tool 消息之后保证待执行工具调用始终位于消息列表末尾。此外策略查找支持通配符*与元组键confirmation_strategies的键可以是单个工具名、元组多个工具共享一个策略或*兜底默认更具体的键优先见_get_confirmation_strategystrategies.py。序列化时元组键会被编码为 JSON 数组字符串如(a, b)→[a, b]反序列化时再还原见_serialize_confirmation_strategies/_deserialize_confirmation_strategies。六、异步/服务端场景BreakpointConfirmationStrategy 与快照恢复当 Agent 运行在无法立即与用户交互的异步环境中例如 Web 后端、任务队列BlockingConfirmationStrategy的同步阻塞方式不再适用。实验版为此提供了BreakpointConfirmationStrategyexperimental_agents_api.md。工作原理该策略的设计目标是先暂停、后审批当某个工具执行需要确认时run()并不返回决策而是总是抛出HITLBreakpointException见 experimental_agents_api.mdAgent 捕获该异常后将自己的当前状态包括工具调用详情序列化为快照文件保存外部系统可以利用该快照通知用户审阅并确认工具执行用户作出决定后通过Agent.run(snapshot...)从保存的快照处恢复执行。HITLBreakpointExceptiondef __init__(message: str, tool_name: str, snapshot_file_path: str, tool_call_id: str | None None) - Nonemessage异常消息tool_name被暂停执行的工具名snapshot_file_path已保存的 pipeline 快照文件路径tool_call_id可选工具调用的唯一标识用于将用户决策与具体某次工具调用关联追踪。BreakpointConfirmationStrategydef __init__(snapshot_file_path: str) - Nonesnapshot_file_path快照保存目录路径。其run方法签名与BlockingConfirmationStrategy一致def run( *, tool_name: str, tool_description: str, tool_params: dict[str, Any], tool_call_id: str | None None, confirmation_strategy_context: dict[str, Any] | None None ) - ToolExecutionDecision行为特点接收的工具描述来自工具自身的description字段供外部 UI 展示confirmation_strategy_context参数保留但不使用仅用于接口兼容无论输入什么run都会抛出HITLBreakpointException永不正常返回run_async直接委托给同步run()支持to_dict/from_dict序列化from_dict通过deserialize_component_inplace还原组件。从快照提取工具调用信息配合断点暂停实验模块还提供工具函数def get_tool_calls_and_descriptions_from_snapshot( agent_snapshot: AgentSnapshot, breakpoint_tool_only: bool True ) - tuple[list[dict], dict[str, str]]从AgentSnapshot中提取工具调用及其描述breakpoint_tool_onlyTrue默认时只处理触发断点的那一个工具调用并重建其参数——非常适合把相关工具调用与描述呈现给人类确认的场景breakpoint_tool_onlyFalse时返回快照中所有工具调用返回值是(工具调用字典列表, 工具描述字典)的元组。七、序列化与反序列化to_dict / from_dictAgent 与 HITL 组件都实现了标准的 Haystack 序列化协议便于 YAML/JSON 化保存与加载可用于 Pipeline YAML 编排、快照持久化Agentto_dict() - dict[str, Any]序列化组件包含chat_generator通过component_to_dict、tools、system_prompt、exit_conditions、state_schema通过_schema_to_dict规范化、max_agent_steps、streaming_callback可调用对象通过serialize_callable、raise_on_tool_invocation_failure、hooks_serialize_hooks_dictionary等from_dict(cls, data) - Agent类方法反序列化会依次还原chat_generator、state_schema、streaming_callback、tools、hooks等组件见 agent.py。ConfirmationStrategy / HookBlockingConfirmationStrategy.to_dict序列化confirmation_policy、confirmation_ui与三个反馈模板from_dict通过deserialize_component_inplace还原策略与 UIBreakpointConfirmationStrategy同样提供to_dict/from_dictConfirmationHook.to_dict会将确认策略字典整体序列化元组键转 JSON 数组字符串from_dict负责还原。组合使用建议在 Pipeline 化场景中推荐把确认策略注册 工具定义 序列化配置集中管理ConfirmationHook与 Agent 均支持to_dict/from_dict可以将整个带 HITL 的 Agent 保存为 YAML实现配置即代码式的可复现部署。八、生产实践要点与边界1. 选择确认策略与 UI 的匹配交互式 CLI / 笔记本环境BlockingConfirmationStrategyRichConsoleUI富提示或SimpleConsoleUI简单提示Web 后端 / 消息队列场景BreakpointConfirmationStrategy 快照 外部审批通道配合run(snapshot...)恢复高频安全工具删除、写库、外发请求建议AlwaysAskPolicy低风险纯计算工具用NeverAskPolicy避免打扰AskOncePolicy适合首次确认、后续信任的场景。2. 请求级上下文传递confirmation_strategy_contextAgent 的run参数与hook_context标准 Agent 的run参数用于在 Web 环境传递每次请求独有的资源WebSocket、异步队列、Redis 客户端。在标准ConfirmationHook中该上下文通过state.data.get(hook_context)读取——注意源码特意通过state.data而非state.get读取因为state.get会做深拷贝可能破坏不可拷贝的资源对象见 hooks.py。3. 退出条件与并发工具工具退出条件会在该工具调用成功且未报错时触发如果同一轮中退出条件工具出错则取消退出、继续循环见_check_exit_conditionsagent.py标准 Agent 支持tool_concurrency_limit默认 4控制并行工具执行数量无工具时 Agent 即 ChatGenerator一次生成即退出——这是零工具场景下的兜底行为可以放心用于纯对话需求。4. 步数预算与流式max_agent_steps默认 100是硬性预算超限后exit_reasonmax_agent_steps可在下游用ConditionalRouter路由到继续/压缩上下文/提示用户等分支。流式输出通过streaming_callback实现同一回调也可用于流式展示工具结果。结语haystack_experimental.components.agents.Agent在标准 Haystack Agent 的基础上把工具调用主循环与human-in-the-loop 确认机制完整打通ConfirmationPolicy决定是否询问、ConfirmationUI负责交互呈现、ConfirmationStrategy产出可执行的ToolExecutionDecision而BreakpointConfirmationStrategy 快照恢复则为异步服务端场景提供了非阻塞审批路径。配合state_schema共享状态、run/run_async双通道和完整序列化协议它可以被平滑嵌入生产级 RAG、多工具智能体与 Web 服务管线。深入阅读 experimental_agents_api.md 以及 agent.py、human_in_the_loop 源码目录可以进一步掌握每个参数的底层行为。【免费下载链接】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），仅供参考