Haystack Experimental Agent 完全指南：Tool-Using Agent 与 Human-in-the-Loop 确认策略实战-育师

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(name="calculator", description="A tool for performing mathematical calculations.", ...) search_tool = Tool(name="search", description="A tool for searching the web.", ...) agent = Agent( chat_generator=OpenAIChatGenerator(), tools=[calculator_tool, search_tool], confirmation_strategies={ calculator_tool.name: HumanInTheLoopStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ), search_tool.name: HumanInTheLoopStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ), }, ) # 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 f"Deleted {path}." hook = ConfirmationHook( confirmation_strategies={ "delete_file": BlockingConfirmationStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ) } ) agent = Agent(chat_generator=OpenAIChatGenerator(), 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_generator`	`ChatGenerator`	必填	Agent 使用的聊天生成器，其`run`方法必须支持`tools`参数，否则构造时抛出`TypeError`（见 agent.py）
`tools`	`ToolsType \\| None`	`None`	可供 Agent 使用的`Tool`对象列表或`Toolset`；为`None`时 Agent 退化为纯 ChatGenerator
`system_prompt`	`str \\| None`	`None`	Agent 的系统提示词，可为普通字符串或 Jinja2 消息模板
`exit_conditions`	`list[str] \\| None`	`["text"]`	退出条件列表：包含`"text"`表示模型生成无工具调用文本即返回；也可填入工具名，表示该工具执行完毕后返回。不合法值会抛`ValueError`
`state_schema`	`dict[str, Any] \\| None`	`None`	工具共享的运行时状态 schema，每个键对应一个类型配置（含`"type"`与可选的`"handler"`）；工具可通过`inputs_from_state`/`outputs_to_state`读写
`max_agent_steps`	`int`	`100`	最大步数上限，一个 step = 一次生成 + 该轮所有工具调用；超限即停止并返回当前状态
`streaming_callback`	`StreamingCallbackT \\| None`	`None`	LLM 流式输出回调；同一回调也可配置为在工具调用时输出工具结果
`raise_on_tool_invocation_failure`	`bool`	`False`	工具调用失败时是否抛异常；为`False`时异常被转换为聊天消息回传给 LLM 继续处理
`confirmation_strategies`	`dict[str, ConfirmationStrategy] \\| None`	`None`	实验版扩展：按工具名映射 human-in-the-loop 确认策略
`tool_invoker_kwargs`	`dict[str, Any] \\| None`	`None`	透传给底层`ToolInvoker`的额外关键字参数
`chat_message_store`	`ChatMessageStore \\| None`	`None`	用于存取聊天历史的存储组件（实验版扩展）
`memory_store`	`MemoryStore \\| None`	`None`	用于存取记忆的存储组件（实验版扩展）

关于`state_schema`的源码细节

标准 Agent 在构造时会为state_schema自动补充几个保留键（见 agent.py）：

messages：类型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.py）；
break_point：AgentBreakpoint，可为针对"chat_generator"的Breakpoint或针对"tool_invoker"的ToolBreakpoint；触发时会抛出BreakpointException；
snapshot：先前保存的 Agent 执行快照字典，包含从上次中断处恢复执行所需的全部信息；
system_prompt：若提供则覆盖默认系统提示词；
tools：本次运行的临时工具集——可以是Tool列表、Toolset，也可以是工具名字符串列表（此时从 Agent 初始化时配置的工具中按名选取，见_select_tools实现 agent.py）；
confirmation_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_criteria：search_memories的参数字典，可含filters（过滤条件）、query（检索查询，若传入则忽略传给 Agent 的用户查询）、top_k（返回记忆条数）、include_memory_metadata（是否把记忆元数据放进ChatMessage）；
kwargs：与state_schema中定义键匹配的额外数据，会被注入运行时状态。

返回字典

run返回的字典包含：

"messages"：整个运行期间交换的全部消息；
"last_message"：最后一条消息（从源码看，它是messages列表的最后一个元素，见 agent.py）；
state_schema中定义的所有额外键，外加运行元数据：step_count、token_usage、tool_call_counts、exit_reason。

exit_reason的取值（用于下游路由，如配合ConditionalRouter）："text"（模型产出无工具调用的完整回复）、"length"/"content_filter"（模型产出不完整回复）、触发工具退出条件的工具名（此时last_message为该工具的结果）、"max_agent_steps"（达到步数上限），或钩子通过stop_run状态键提供的自定义原因。

异常

RuntimeError：Agent 组件在调用run()前未完成 warm-up；
BreakpointException：触发了 Agent 断点（break_point）。

run_async

run_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.py：

1. ConfirmationPolicy（确认策略：何时询问）

ConfirmationPolicy协议定义should_ask(tool_name, tool_description, tool_params) -> bool，仓库内置实现位于 haystack/hooks/human_in_the_loop/policies.py：

策略	行为
`AlwaysAskPolicy`	总是询问：`should_ask`恒返回`True`
`NeverAskPolicy`	从不询问：`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是一个数据类，字段包括action（"confirm"/"reject"/"modify"）、feedback（用户反馈文本）、new_tool_params（修改后的参数）。

3. ConfirmationStrategy（确认策略对象：组合决策）

ConfirmationStrategy协议定义run(...) -> ToolExecutionDecision与run_async(...)。核心实现是BlockingConfirmationStrategy（strategies.py），其run流程为：

调用confirmation_policy.should_ask(...)，返回False则直接生成execute=True的ToolExecutionDecision；
需要确认时调用confirmation_ui.get_user_confirmation(...)；
将 UI 结果回传给策略的update_after_confirmation供其记忆学习；
根据action生成决策：
- "reject"→execute=False，附带模板生成的拒绝反馈（默认模板REJECTION_FEEDBACK_TEMPLATE = "Tool execution for '{tool_name}' was rejected by the user."，可自定义reject_template）；
- "modify"→execute=True，final_tool_params替换为用户修改后的参数，并附带修改说明模板（MODIFICATION_FEEDBACK_TEMPLATE，含{tool_name}与{final_tool_params}占位符）；
- "confirm"→execute=True，原样执行。

BlockingConfirmationStrategy支持三种反馈模板定制：reject_template、modify_template、user_feedback_template（后者含{feedback}占位符）。

决策的应用：拒绝、修改与历史重写

策略产生的ToolExecutionDecision会被_apply_tool_execution_decisions（strategies.py）应用到对话历史上：

reject：向历史插入"assistant 工具调用消息 + 带error=True的 tool 结果消息"对，把拒绝反馈喂回给 LLM；
modify：在工具调用消息前插入一条 user 消息解释参数被修改的原因（否则 LLM 不知道参数为何变化，可能再次用原参数调用）；
最终由_update_chat_history将拒绝/修改消息插入到对话中最后一个 user 或 tool 消息之后，保证待执行工具调用始终位于消息列表末尾。

此外，策略查找支持通配符"*"与元组键：confirmation_strategies的键可以是单个工具名、元组（多个工具共享一个策略）或"*"（兜底默认）；更具体的键优先（见_get_confirmation_strategy，strategies.py）。序列化时元组键会被编码为 JSON 数组字符串（如("a", "b")→'["a", "b"]'），反序列化时再还原（见_serialize_confirmation_strategies/_deserialize_confirmation_strategies）。

六、异步/服务端场景：BreakpointConfirmationStrategy 与快照恢复

当 Agent 运行在无法立即与用户交互的异步环境中（例如 Web 后端、任务队列），BlockingConfirmationStrategy的同步阻塞方式不再适用。实验版为此提供了BreakpointConfirmationStrategy（experimental_agents_api.md）。

工作原理

该策略的设计目标是"先暂停、后审批"：

当某个工具执行需要确认时，run()并不返回决策，而是总是抛出HITLBreakpointException（见 experimental_agents_api.md）；
Agent 捕获该异常后，将自己的当前状态（包括工具调用详情）序列化为快照文件保存；
外部系统可以利用该快照通知用户审阅并确认工具执行；
用户作出决定后，通过Agent.run(snapshot=...)从保存的快照处恢复执行。

HITLBreakpointException

def __init__(message: str, tool_name: str, snapshot_file_path: str, tool_call_id: str | None = None) -> None

message：异常消息；
tool_name：被暂停执行的工具名；
snapshot_file_path：已保存的 pipeline 快照文件路径；
tool_call_id：可选，工具调用的唯一标识，用于将用户决策与具体某次工具调用关联追踪。

BreakpointConfirmationStrategy

def __init__(snapshot_file_path: str) -> None

snapshot_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_only=True（默认）时，只处理触发断点的那一个工具调用并重建其参数——非常适合"把相关工具调用与描述呈现给人类确认"的场景；
breakpoint_tool_only=False时返回快照中所有工具调用；
返回值是(工具调用字典列表, 工具描述字典)的元组。

七、序列化与反序列化：to_dict / from_dict

Agent 与 HITL 组件都实现了标准的 Haystack 序列化协议，便于 YAML/JSON 化保存与加载（可用于 Pipeline YAML 编排、快照持久化）：

Agent

to_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 / Hook

BlockingConfirmationStrategy.to_dict序列化confirmation_policy、confirmation_ui与三个反馈模板；from_dict通过deserialize_component_inplace还原策略与 UI；
BreakpointConfirmationStrategy同样提供to_dict/from_dict；
ConfirmationHook.to_dict会将确认策略字典整体序列化（元组键转 JSON 数组字符串），from_dict负责还原。

组合使用建议

在 Pipeline 化场景中，推荐把"确认策略注册 + 工具定义 + 序列化配置"集中管理：ConfirmationHook与 Agent 均支持to_dict/from_dict，可以将整个带 HITL 的 Agent 保存为 YAML，实现"配置即代码"式的可复现部署。

八、生产实践要点与边界

1. 选择确认策略与 UI 的匹配

交互式 CLI / 笔记本环境：BlockingConfirmationStrategy+RichConsoleUI（富提示）或SimpleConsoleUI（简单提示）；
Web 后端 / 消息队列场景：BreakpointConfirmationStrategy+ 快照 + 外部审批通道，配合run(snapshot=...)恢复；
高频安全工具（删除、写库、外发请求）建议AlwaysAskPolicy；低风险纯计算工具用NeverAskPolicy避免打扰；AskOncePolicy适合"首次确认、后续信任"的场景。

2. 请求级上下文传递

confirmation_strategy_context（Agent 的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_conditions，agent.py）；
标准 Agent 支持tool_concurrency_limit（默认 4）控制并行工具执行数量；
无工具时 Agent 即 ChatGenerator，一次生成即退出——这是"零工具"场景下的兜底行为，可以放心用于纯对话需求。

4. 步数预算与流式

max_agent_steps（默认 100）是硬性预算，超限后exit_reason="max_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 源码目录，可以进一步掌握每个参数的底层行为。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Haystack Experimental Agent 完全指南：Tool-Using Agent 与 Human-in-the-Loop 确认策略实战