Haystack 实验性 Agents API 完全指南：工具调用、退出条件与 Human-in-the-Loop 确认策略-育师

Haystack 实验性 Agents API 完全指南：工具调用、退出条件与 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 官方参考文档（docs-website/reference_versioned_docs/version-2.22/experiments-api/experimental_agents_api.md）展开，围绕实验性的haystack_experimental.components.agents.Agent组件，系统讲解其工具调用机制、退出条件、运行时状态（State）、流式回调、断点（Breakpoint）与快照（Snapshot），以及以HumanInTheLoopStrategy、BreakpointConfirmationStrategy为代表的人机协同确认策略。读完本文，你将掌握如何构建一个"感知工具、按需停止、可人工审批工具执行"的 Agent，并理解其在当前 Haystack 仓库（haystack/components/agents/agent.py、haystack/hooks/human_in_the_loop/）中的底层实现脉络。

一、Agent 组件是什么

根据参考文档，haystack_experimental.components.agents.agent.Agent是一个实现了工具使用型 Agent的 Haystack 组件，核心特点包括：

与模型提供商无关的聊天模型支持：只要 Chat Generator 的run方法支持tools参数，即可接入；
工具循环执行：组件持续处理消息、执行工具，直到满足某个退出条件（exit condition）；
多退出条件：既可以在模型直接返回文本时退出，也可以在指定的工具执行完毕后退出，多个条件可以同时指定；
无工具时退化为普通聊天模型：当不传入任何工具时，Agent 的表现与ChatGenerator一致——生成一次回复后立即退出；
扩展了 Haystack 核心 Agent：文档明确指出该类在 HaystackAgent组件之上扩展了 human-in-the-loop（人机协同）确认策略支持。当前仓库中，核心Agent的实现位于 haystack/components/agents/agent.py，而人机协同确认机制已沉淀为ConfirmationHook与BlockingConfirmationStrategy等正式组件（见 haystack/hooks/human_in_the_loop/）。

最小使用示例

参考文档给出的标准用法如下：

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

这段示例展示了两个关键能力：为不同工具注册不同的确认策略（计算器工具从不询问、搜索工具每次都询问），以及通过result["messages"]获取完整的对话历史。从仓库源码看，这种"按工具差异化审批"的设计在正式版中对应ConfirmationHook的confirmation_strategies字典（见 haystack/hooks/human_in_the_loop/hooks.py），其键既可以是单个工具名，也可以是共享同一策略的工具名元组，还可以是应用于兜底的通配符"*"——更具体的键优先匹配。

二、Agent 初始化参数详解

参考文档给出了Agent.__init__的完整签名，以下逐一说明每个参数的含义、默认值与源码层面的影响：

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`参数）
`tools`	`ToolsType \\| None`	`None`	Agent 可用的`Tool`对象列表或一个`Toolset`
`system_prompt`	`str \\| None`	`None`	Agent 的系统提示词
`exit_conditions`	`list[str] \\| None`	`["text"]`	使 Agent 返回的条件列表。包含`"text"`表示生成无工具调用的消息时返回；也可包含工具名，表示该工具执行完毕后返回
`state_schema`	`dict[str, Any] \\| None`	`None`	工具使用的运行时状态（State）的 schema
`max_agent_steps`	`int`	`100`	Agent 运行的最大步数上限，超出后停止并返回当前状态
`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`	Agent 存取聊天消息历史的存储
`memory_store`	`MemoryStore \\| None`	`None`	Agent 存取记忆的存储

异常：当chat_generator的run方法不支持tools参数时抛出TypeError；exit_conditions不合法时抛出ValueError。

源码层面的关键实现细节

在 haystack/components/agents/agent.py 中，Agent.__init__对上述参数做了进一步细化与校验，可作为理解实验 API 的参照：

工具能力自检：初始化时通过inspect.signature(chat_generator.run).parameters判断生成器是否接受tools参数（源码第 460 行），若传入工具但不支持则立即抛出TypeError；
退出条件默认值：exit_conditions is None时回退为["text"]（第 469-470 行）；
状态 schema 保留键：step_count、token_usage、tool_call_counts、exit_reason等运行元数据键，以及continue_run、stop_run、tools、hook_context、context_tokens等内部控制键是保留的，用户不得在state_schema中重定义（第 77-99、472-480 行）；
工具并发上限：正式版还提供了tool_concurrency_limit（默认 4）与tool_streaming_callback_passthrough，前者控制并行执行工具调用的最大数量，设为 1 即禁用并行（第 393-394 行）。

三、Agent.run：驱动工具循环的核心入口

Agent.run是 Agent 的主循环入口，签名如下：

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对象列表；
streaming_callback：LLM 流式输出回调，与初始化时传入的回调二选一，运行期优先级更高；
generation_kwargs：传给 LLM 的额外生成参数，会覆盖初始化时传入的同名参数；
break_point：AgentBreakpoint，可以是针对"chat_generator"的Breakpoint，或针对"tool_invoker"的ToolBreakpoint；
snapshot：此前保存的 Agent 执行快照字典，包含从断点处恢复执行所需的全部信息；
system_prompt：运行期系统提示词，提供时覆盖默认值；
tools：本次运行使用的工具，可以是Tool列表、Toolset或工具名字符串列表（按名称从 Agent 原始配置的工具中选取）；
confirmation_strategy_context：向确认策略传递请求级资源的字典，在 Web/服务端场景尤为有用，例如传递 WebSocket 连接、异步队列、Redis pub/sub 客户端，使策略可以进行非阻塞式用户交互；
chat_message_store_kwargs：传给ChatMessageStore的关键字参数，例如chat_history_id与last_k用于按历史 ID 和最近条数检索聊天历史；
memory_store_kwargs：传给MemoryStore的关键字参数，可包含：
- user_id：检索/写入记忆的用户 ID；
- run_id：检索/写入记忆的运行 ID；
- agent_id：检索/写入记忆的 Agent ID；
- search_criteria：search_memories方法的参数字典，可包含filters（记忆检索过滤器）、query（检索查询，注意：一旦传入，Agent 的用户查询在记忆检索时会被忽略）、top_k（返回记忆条数）、include_memory_metadata（是否在ChatMessage中包含记忆元数据）；
kwargs：传入 State schema 的额外数据，键必须与state_schema定义匹配。

返回值

run返回一个字典，包含以下键：

"messages"：Agent 运行期间交换的全部消息列表；
"last_message"：运行期间交换的最后一条消息；
state_schema中定义的任何其他键。

异常：未warm_up就调用run()会抛RuntimeError；Agent 断点被触发时抛BreakpointException。

底层运行循环（源码视角）

从当前仓库 haystack/components/agents/agent.py 的run实现（第 826-907 行）可以清晰看到这个循环的骨架：

预热：调用self.warm_up()预热工具、钩子与聊天生成器（第 872 行）；
初始化执行上下文：_initialize_fresh_execution构建State、选定工具、解析流式回调，并初始化step_count、token_usage、tool_call_counts、exit_reason等运行元数据（第 874-882 行）；
主循环：在counter < max_agent_steps的条件下反复执行_run_step，每步一次聊天生成调用加上该轮所有工具调用；当_run_step返回False时跳出循环（第 889-891 行）；
步数上限兜底：若循环正常耗尽（未break），说明是max_agent_steps触顶，此时记录exit_reason = "max_agent_steps"并输出警告日志（第 892-899 行）；
组装结果：剔除内部状态键后，从messages中取最后一条作为last_message返回（第 900-905 行）。

四、Agent.run_async：异步版本

run_async是run的异步版本，遵循相同逻辑，但尽可能使用异步操作——例如优先调用ChatGenerator.run_async（若可用）。其参数与run基本一致（streaming_callback为异步回调、generation_kwargs同样覆盖初始化参数），差异仅在于：

需先调用warm_up_async()而非warm_up()；
未预热即调用抛RuntimeError；断点触发抛BreakpointException；
返回值与run完全一致（"messages"、"last_message"+state_schema键）。

在源码中，run_async（第 909-993 行）与run的差异点集中在：预热阶段调用warm_up_async（第 958 行）、钩子调用替换为_run_hooks_async（第 972、986 行）、步骤执行替换为_run_step_async（第 976 行）。这也与仓库中OpenAIChatGenerator等组件普遍提供run_async的趋势一致（参见 releasenotes 中的add-run-async-to-*系列说明）。

五、序列化：to_dict 与 from_dict

实验 Agent 与 Haystack 其他组件一样支持完整的序列化/反序列化，便于 YAML 描述、管道持久化与远程传输：

def to_dict() -> dict[str, Any]

将组件序列化为字典并返回。对应的类方法：

@classmethod def from_dict(cls, data: dict[str, Any]) -> "Agent"

从字典反序列化出Agent实例，参数data为待反序列化的字典。

从源码看（haystack/components/agents/agent.py 第 633-679 行），to_dict通过default_to_dict序列化chat_generator、工具、提示词、退出条件、state_schema、流式回调（经serialize_callable）、钩子等全部初始化参数；from_dict则依次反序列化聊天生成器、状态 schema、流式回调、工具与钩子，最终交给default_from_dict完成实例重建。

六、Human-in-the-Loop：让工具执行接受人工审批

实验 Agent 的亮点在于为工具调用引入了人工确认层。参考文档将其划分为三个模块：human_in_the_loop.breakpoint、human_in_the_loop.errors与human_in_the_loop.strategies。

6.1 确认策略（HumanInTheLoopStrategy）

实验 API 中，confirmation_strategies参数接受dict[str, ConfirmationStrategy]，将每个工具名映射到其确认策略（如示例中的HumanInTheLoopStrategy(confirmation_policy=..., confirmation_ui=...)）。策略由两部分组合而成：

确认策略（ConfirmationPolicy）：决定"何时询问"。当前仓库 haystack/hooks/human_in_the_loop/policies.py 提供三种现成实现：
- AlwaysAskPolicy：每次都询问（should_ask恒返回True）；
- NeverAskPolicy：从不询问（should_ask恒返回False）；
- AskOncePolicy：对同一工具的相同参数只询问一次（内部记录已确认的tool_name -> tool_params映射，避免重复打断）。
确认 UI（ConfirmationUI）：决定"如何询问"。见 haystack/hooks/human_in_the_loop/user_interfaces.py：
- SimpleConsoleUI：基于标准输入输出的纯文本交互，支持y/n/m（确认/拒绝/修改），无需额外依赖；
- RichConsoleUI：基于rich库的富文本面板交互，同样支持确认/拒绝/修改三选一，修改参数时非字符串类型按 JSON 解析（需要pip install rich）。

三种交互结果由 haystack/hooks/human_in_the_loop/dataclasses.py 中的ConfirmationUIResult承载：action（"confirm"/"reject"/"modify"）、可选feedback（用户反馈文本）、可选new_tool_params（修改后的参数）。

6.2 决策处理流程（源码纵深）

从 haystack/hooks/human_in_the_loop/strategies.py 的BlockingConfirmationStrategy.run（第 66-139 行）可以看到一次完整的审批闭环：

先调用confirmation_policy.should_ask(...)判断是否需要询问，若不需要则直接放行（execute=True）；
需要询问时调用confirmation_ui.get_user_confirmation(...)收集用户意见；
将结果回传给confirmation_policy.update_after_confirmation(...)，供AskOncePolicy这类有状态策略记录学习；
根据action分派：
- reject：不执行，生成拒绝反馈文本（模板"Tool execution for '{tool_name}' was rejected by the user."），可拼接用户反馈；
- modify：用new_tool_params替换原参数后执行，并生成参数修改说明（模板"The parameters for tool '{tool_name}' were updated by the user to: ..."）；
- confirm：按原参数直接执行。

最终产出ToolExecutionDecision（tool_name、execute、tool_call_id、feedback、final_tool_params）。tool_call_id用于将决策与具体工具调用一一关联，避免并行工具调用时错配。

6.3 ConfirmationHook：接入 Agent 主循环的桥梁

在当前仓库中，确认逻辑以before_tool钩子的形式挂载到核心 Agent 上（haystack/hooks/human_in_the_loop/hooks.py）：

hook = ConfirmationHook( confirmation_strategies={ "delete_file": BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=RichConsoleUI() ), "*": BlockingConfirmationStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ), } ) agent = Agent(chat_generator=OpenAIChatGenerator(), tools=[delete_file], hooks={"before_tool": [hook]})

要点：

allowed_hook_points = ("before_tool",)，即该钩子只能在工具执行前的钩子点注册，Agent 会在构造时校验并拒绝其他位置（对应 haystack/components/agents/agent.py 中_validate_hooks的钩子点限制逻辑）；
运行时从state.data读取可用工具（"tools"）与请求级上下文（"hook_context"），只处理最后一条含工具调用的消息；
通过confirmation_strategy_context（实验 API 的run参数）/ 正式版的hook_context传递 WebSocket、队列等每请求资源，实现非阻塞交互——这正是实验文档中confirmation_strategy_context参数的设计意图。

七、BreakpointConfirmationStrategy：无法即时交互时的审批方案

实验 API 中还有一种专门为"无法即时交互"场景设计的策略：BreakpointConfirmationStrategy。它不阻塞等待用户输入，而是通过抛出断点异常来暂停执行，将状态序列化保存，随后再异步通知用户审批。

7.1 构造与运行

def __init__(snapshot_file_path: str) -> None

参数snapshot_file_path为快照保存目录路径。

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

该方法总是抛出HITLBreakpointException，不会返回。confirmation_strategy_context参数仅用于接口兼容，此策略并不使用它。run_async是run的异步包装，同样总是抛异常。

7.2 HITLBreakpointException

该异常（模块human_in_the_loop.errors）在工具执行被ConfirmationStrategy暂停时抛出，构造参数：

message：异常消息；
tool_name：被暂停执行的工具名；
snapshot_file_path：已保存的管道快照文件路径；
tool_call_id（可选）：工具调用的唯一标识，用于将审批决策关联回具体的工具调用。

抛出后，Agent 捕获异常并序列化其当前状态（含工具调用细节），这些信息可用于通知用户审阅并确认工具执行。

7.3 从快照提取工具调用信息

配套的辅助函数：

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）只处理导致断点的那个工具调用并重建其参数，适合"将相关工具调用及其描述呈现给人工确认后再执行"的场景；设为False则返回全部工具调用。返回值是一个二元组：工具调用字典列表 + 工具名到描述的字典。

八、实践：组装一个带人工审批的 Agent 工作流

综合以上 API，一个完整的"工具调用 + 差异化审批"工作流可以这样组织：

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, AskOncePolicy, SimpleConsoleUI, ) calculator_tool = Tool(name="calculator", description="A tool for performing mathematical calculations.", ...) send_email_tool = Tool(name="send_email", description="Send an email to a recipient.", ...) search_tool = Tool(name="search", description="A tool for searching the web.", ...) agent = Agent( chat_generator=OpenAIChatGenerator(), tools=[calculator_tool, send_email_tool, search_tool], exit_conditions=["text", "send_email"], # 文本回复或发送邮件后退出 max_agent_steps=50, # 限制步数上限，防止失控循环 confirmation_strategies={ # 计算器：从不打扰用户 calculator_tool.name: HumanInTheLoopStrategy( confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI() ), # 搜索：每次都确认 search_tool.name: HumanInTheLoopStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI() ), # 发送邮件：同一参数只确认一次 send_email_tool.name: HumanInTheLoopStrategy( confirmation_policy=AskOncePolicy(), confirmation_ui=SimpleConsoleUI() ), }, ) result = agent.run( messages=[ChatMessage.from_user("Find the latest Haystack release and email it to me")], generation_kwargs={"temperature": 0.3}, # 运行期覆盖生成参数 ) assert "messages" in result assert "last_message" in result

在 Web 服务场景中，还可以通过confirmation_strategy_context传入每请求的 WebSocket 连接或异步队列，让审批以非阻塞方式推送给前端用户，而run_async则保证整个 Agent 循环不会占用事件循环。

九、小结与进一步阅读

实验性 Agents API 为 Haystack 提供了完整的"模型规划 → 工具执行 → 人工把关"能力：exit_conditions与max_agent_steps双保险控制循环生命周期，state_schema让工具间共享运行时状态，HumanInTheLoopStrategy（策略 + UI）与BreakpointConfirmationStrategy（异常 + 快照）覆盖了"可即时交互"与"不可即时交互"两类审批场景，run_async则保证高并发服务场景下的可用性。

若想深入了解上述机制在当前仓库中的正式实现，建议继续阅读：

Agent 主循环与状态管理：haystack/components/agents/agent.py、haystack/components/agents/state/state.py；
人机协同确认机制：haystack/hooks/human_in_the_loop/hooks.py、haystack/hooks/human_in_the_loop/strategies.py、haystack/hooks/human_in_the_loop/policies.py、haystack/hooks/human_in_the_loop/user_interfaces.py；
协议定义：haystack/hooks/human_in_the_loop/types/protocol.py、haystack/hooks/human_in_the_loop/dataclasses.py；
对应测试：test/hooks/human_in_the_loop/test_hooks.py、test/hooks/human_in_the_loop/test_strategies.py、test/hooks/human_in_the_loop/test_policies.py。

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