企业官网建设流程全解析

Haystack 人机协同（Human-in-the-Loop）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

在构建生产级 LLM Agent 时，高风险动作（发送邮件、修改数据库、发起 API 调用）往往不应由模型全权决定。Haystack 通过haystack.hooks.human_in_the_loop模块提供了一套完整的人机协同（HITL）抽象：以ConfirmationHook挂在 Agent 的before_tool钩子点上，在工具调用真正执行之前拦截请求，让人工实时confirm（确认）、reject（拒绝）或modify（修改参数）。读完本文，你将掌握该模块的全部 API（数据类、策略、UI、序列化协议），并能组合出适合控制台与 Web 服务场景的确认流程。本文以 version-2.23 的 human_in_the_loop API 参考 为主线，结合仓库源码与官方使用指南逐层展开。

在 Haystack Agent 中，HITL 如何工作

HITL 是 Agent 通用 hooks 机制 的一种应用。一个注册在before_tool钩子点上的ConfirmationHook，会截获模型请求的待执行工具调用，并通过改写 Agent 的State中的对话消息，实现对工具调用的确认、修改或拒绝。

整个系统由四个层次组成（对应模块内四个文件）：

当 Agent 即将调用工具时，执行链为：策略（Strategy）先咨询策略器（Policy）；若策略器判定需要询问，UI 便把工具名称、描述和参数展示给人工，人工可确认、拒绝或修改；随后 Agent 按人工的决策继续执行。这四个层次分别定义在haystack/hooks/human_in_the_loop/下的hooks.py、strategies.py、policies.py、user_interfaces.py，数据类集中在dataclasses.py，抽象协议则位于 types/protocol.py。

模块统一从haystack.hooks.human_in_the_loop导入，__init__.py暴露了全部公共符号：ConfirmationUIResult、ToolExecutionDecision、ConfirmationHook、AlwaysAskPolicy、AskOncePolicy、NeverAskPolicy、BlockingConfirmationStrategy、RichConsoleUI、SimpleConsoleUI（见init.py）。

核心数据类：一次人工交互的输入与输出

ConfirmationUIResult：UI 返回的人工决策

定义在 dataclasses.py，是确认 UI 交互的结果：

from haystack.hooks.human_in_the_loop import ConfirmationUIResult # 拒绝并附带原因 reject_result = ConfirmationUIResult(action="reject", feedback="参数包含外部邮箱，禁止发送") # 修改参数 modify_result = ConfirmationUIResult(action="modify", new_tool_params={"to": "internal@mycompany.com"}) # 确认 confirm_result = ConfirmationUIResult(action="confirm")

ToolExecutionDecision：策略返回的执行决策

定义在 dataclasses.py，是策略（Strategy）对「是否执行该工具调用」做出的最终决定：

该数据类还提供两个序列化方法：

• to_dict() -> dict[str, Any]：把决策转为字典表示，底层直接使用dataclasses.asdict（dataclasses.py）；
• from_dict(data: dict[str, Any]) -> "ToolExecutionDecision"：从字典反序列化，cls(**data)直接还原实例（dataclasses.py）。

确认策略器（Policies）：决定「何时询问」

策略器统一实现ConfirmationPolicy协议（protocol.py），需要提供should_ask()与可选的update_after_confirmation()，并默认通过default_to_dict/default_from_dict支持序列化。三个内置实现位于 policies.py：

AlwaysAskPolicy：每次都问

should_ask无条件返回True（policies.py），适用于发送邮件、删除数据等每次调用都需人工把关的高风险场景。

NeverAskPolicy：从不询问

should_ask无条件返回False（policies.py）。它的典型用途不是「不配策略」，而是保留策略与 UI 配置的同时临时关闭 HITL——需要恢复时只改策略器即可，无需改动 Hook 注册。

AskOncePolicy：同一参数只问一次

should_ask只在「同一工具 + 相同参数」未被确认过时返回True（policies.py）。其内部维护self._asked_tools: dict[str, Any]记录已确认的tool_name -> tool_params；update_after_confirmation在人工选择"confirm"时把该组合写入内部状态，从而后续重复调用不再打扰用户（policies.py）。

自定义策略器

继承ConfirmationPolicy即可实现按业务规则定制的策略。例如官方指南给出的「仅当to参数指向外部邮箱域时才询问」：

from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy, ConfirmationUIResult from typing import Any class AskForSensitiveParamsPolicy(ConfirmationPolicy): """Only ask when the 'to' parameter looks like an external email domain.""" def should_ask( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], ) -> bool: to = tool_params.get("to", "") return not to.endswith("@mycompany.com")

对于有状态策略，还需实现update_after_confirmation：它在用户响应后被调用，并接收完整的ConfirmationUIResult，使你可以根据结果更新内部状态。下面的策略按工具名只问一次，用户已确认过的工具不再重复询问：

from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy from haystack.hooks.human_in_the_loop import ConfirmationUIResult from typing import Any class AskOncePerToolPolicy(ConfirmationPolicy): """Ask once per tool name, regardless of parameters. Skip on repeat confirmed calls.""" def __init__(self) -> None: self._confirmed_tools: set[str] = set() def should_ask( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], ) -> bool: return tool_name not in self._confirmed_tools def update_after_confirmation( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], confirmation_result: ConfirmationUIResult, ) -> None: if confirmation_result.action == "confirm": self._confirmed_tools.add(tool_name)

确认策略（Strategy）：BlockingConfirmationStrategy

策略实现ConfirmationStrategy协议（protocol.py），负责「工具即将被调用时做什么」。内置的BlockingConfirmationStrategy会阻塞执行以收集人工反馈，完整实现位于 strategies.py。

构造参数

def __init__(*, confirmation_policy: ConfirmationPolicy, confirmation_ui: ConfirmationUI, reject_template: str = REJECTION_FEEDBACK_TEMPLATE, modify_template: str = MODIFICATION_FEEDBACK_TEMPLATE, user_feedback_template: str = USER_FEEDBACK_TEMPLATE) -> None

• confirmation_policy：确认策略器，决定何时询问；
• confirmation_ui：与用户交互的 UI；
• reject_template：拒绝反馈模板，需包含{tool_name}占位符，默认"Tool execution for '{tool_name}' was rejected by the user."；
• modify_template：修改反馈模板，需包含{tool_name}与{final_tool_params}占位符，默认"The parameters for tool '{tool_name}' were updated by the user to:\n{final_tool_params}"；
• user_feedback_template：用户附加反馈模板，需包含{feedback}占位符，默认"With user feedback: {feedback}"。

三个模板常量定义于 strategies.py。模板只会被发送回 LLM 解释「发生了什么」，各自有合理默认值，一般无需改动；只有需要定制措辞时才覆盖。

run()：同步执行决策流程

签名如下（strategies.py）：

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

内部逻辑（strategies.py）：

1. 先调用confirmation_policy.should_ask(...)；若返回False，直接返回execute=True且final_tool_params=tool_params，不打扰用户；
2. 否则调用confirmation_ui.get_user_confirmation(...)阻塞等待人工输入；
3. 把结果回传给confirmation_policy.update_after_confirmation(...)，供有状态策略学习更新；
4. 按action分流：
   • "reject"：拼装reject_template（如有feedback再拼user_feedback_template），返回execute=False的决策，把原因反馈给 LLM；
   • "modify"且携带new_tool_params：用新参数替换原参数，拼装modify_template，返回execute=True且final_tool_params为新参数的决策；
   • 其余视为"confirm"：返回execute=True、final_tool_params=tool_params。

run_async()：异步版本

签名与run()完全一致，默认实现直接委托同步run()（strategies.py），因此开箱即用；需要真正异步行为（如等待 WebSocket 事件）时再覆写。

序列化

• to_dict()：序列化为字典，其中confirmation_policy与confirmation_ui会递归调用各自的to_dict()（strategies.py）；
• from_dict()：先对init_parameters中的confirmation_policy、confirmation_ui进行组件级原地反序列化，再构造实例（strategies.py）。

test/hooks/human_in_the_loop/test_strategies.py中的TestBlockingConfirmationStrategy::test_to_dict给出了完整序列化快照，可见序列化结果包含type字段（如haystack.hooks.human_in_the_loop.strategies.BlockingConfirmationStrategy）与init_parameters两个部分。

用户界面（UIs）：控制台交互的两种实现

UI 实现ConfirmationUI协议（protocol.py），核心方法是get_user_confirmation(tool_name, tool_description, tool_params) -> ConfirmationUIResult。内置两个控制台实现位于 user_interfaces.py：

SimpleConsoleUI：纯标准库实现

使用input()/print()，零额外依赖（user_interfaces.py）。交互流程：

1. 打印工具名称、描述与参数（含(No arguments)兜底）；
2. 循环等待输入，合法选项为y/yes、n/no、m/modify，非法输入会提示重试；
3. y→action="confirm"；m→ 逐参数询问修改，字符串参数按原样输入、非字符串参数按 JSON 解析（解析失败会红色提示重试）；n→ 再询问可选的反馈消息，返回action="reject"。

两个 UI 都通过模块级_ui_lock = Lock()保证多线程场景下提示不交错（user_interfaces.py）。

RichConsoleUI：基于 rich 的样式化界面

依赖第三方库rich（pip install rich），通过LazyImport延迟加载——未安装时访问会提示Run 'pip install rich'（user_interfaces.py）。交互界面以 Panel 展示🔧 Tool Execution Request（工具名、描述、参数），用Prompt.ask提供y / n / m选择（默认y），错误输入会自动重新提示。构造时可传入自定义Console实例；序列化时因 Console 对象不可序列化，to_dict会固定存储console=None（user_interfaces.py）。

from haystack.hooks.human_in_the_loop import RichConsoleUI, SimpleConsoleUI # 依赖 rich，界面更美观 rich_ui = RichConsoleUI() # 纯标准库，零依赖 simple_ui = SimpleConsoleUI()

完整接入：把 HITL 挂到 Agent 上

ConfirmationHook（hooks.py）是这一切的入口。它被限制在before_tool钩子点（allowed_hook_points = ("before_tool",)），Agent 构造时会校验，注册到其他位置会直接报错；官方建议用一个ConfirmationHook内配置多个条目，而非注册多个 Hook。

基础示例：高危操作必须人工确认

from typing import Annotated from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.hooks.human_in_the_loop import ( AlwaysAskPolicy, BlockingConfirmationStrategy, ConfirmationHook, SimpleConsoleUI, ) from haystack.tools import tool @tool def send_email( to: Annotated[str, "The recipient email address"], subject: Annotated[str, "The email subject line"], body: Annotated[str, "The email body"], ) -> str: """Send an email to a recipient.""" return f"Email sent to {to}." strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI(), ) agent = Agent( chat_generator=OpenAIChatGenerator(model="gpt-5.4-mini"), tools=[send_email], hooks={ "before_tool": [ ConfirmationHook(confirmation_strategies={"send_email": strategy}), ], }, ) result = agent.run( messages=[ChatMessage.from_user("Send a welcome email to alice@example.com")], )

当 Agent 调用send_email时，终端会暂停并展示：

--- Tool Execution Request --- Tool: send_email Description: Send an email to a recipient. Arguments: to: alice@example.com subject: Welcome! body: Hi Alice, welcome aboard! ------------------------------ Confirm execution? (y=confirm / n=reject / m=modify):

人工输入y放行、n拒绝（可附原因）、m进入参数修改流程；拒绝或修改的说明会以模板消息回写进对话历史，让 LLM 知晓参数为何变化，避免模型再次以原参数发起调用。

多工具分组与通配符

confirmation_strategies的键支持三种形式：单个工具名、工具名元组（多个工具共享一个策略）、以及通配符"*"（应用于任何没有更具体条目的工具）。更具体的键优先，因此可以用"*"设默认值、再单独覆盖个别工具（hooks.py）：

@tool def delete_record(record_id: Annotated[str, "The ID of the record to delete"]) -> str: """Delete a record from the database.""" return f"Record {record_id} deleted." @tool def update_record( record_id: Annotated[str, "The ID of the record to update"], data: Annotated[str, "The new data as a JSON string"], ) -> str: """Update a record in the database.""" return f"Record {record_id} updated." @tool def search(query: Annotated[str, "The search query"]) -> str: """Search the knowledge base.""" return f"Results for: {query}" ask_strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI(), ) confirmation_hook = ConfirmationHook( confirmation_strategies={ # 用元组键让多个敏感工具共享同一策略 ("send_email", "delete_record", "update_record"): ask_strategy, # search 没有配置策略 —— 始终直接执行，不询问 }, ) agent = Agent( chat_generator=OpenAIChatGenerator(model="gpt-5.4-mini"), tools=[send_email, delete_record, update_record, search], hooks={"before_tool": [confirmation_hook]}, )

定制反馈文案

被拒绝或修改时，BlockingConfirmationStrategy会发消息回 LLM 解释。三个可选模板参数均可按需覆盖：

strategy = BlockingConfirmationStrategy( confirmation_policy=AlwaysAskPolicy(), confirmation_ui=SimpleConsoleUI(), reject_template="Skipping '{tool_name}' — rejected by operator.", modify_template="Updated parameters for '{tool_name}': {final_tool_params}", user_feedback_template="Reason: {feedback}", )

序列化细节：元组键与通配符如何持久化

HITL 组件全部支持to_dict/from_dict，可嵌入 Agent 的 YAML/JSON 序列化链路。一个值得注意的实现细节：映射键必须是字符串，而元组键（一组工具共享一个策略）会被编码为 JSON 数组字符串，例如("a", "b")→'["a", "b"]'；单个工具名保持原样；反序列化时再按[前缀还原为元组，同时兼容旧版内存态的 list 键（strategies.py）。ConfirmationHook.to_dict通过_serialize_confirmation_strategies完成上述编码，from_dict则反向还原（hooks.py）。

源码视角：确认流程的底层执行链

当ConfirmationHook.run被触发时，它只读取最后一条含工具调用的消息（hooks.py），随后进入strategies.py的处理管线：

1. _process_confirmation_strategies（strategies.py）：若策略表为空直接返回原历史；否则为每个工具调用收集ToolExecutionDecision；
2. _run_confirmation_strategies（strategies.py）：按工具名解析到具体Tool——找不到的工具（如模型幻觉出的名字）走_passthrough_tool_call原样放行，交由后续执行层统一抛出ToolNotFoundException（尊重raise_on_failure）；找到但无对应策略的工具直接生成execute=True决策；有策略则调用策略的run()，并包裹 tracing spanhaystack.agent.hook.human_in_the_loop.strategy（记录输入参数与决策分类）；
3. _apply_tool_execution_decisions（strategies.py）：把决策应用到消息上。拒绝的调用生成「assistant 工具调用消息 + error=True 的 tool 结果消息」对；修改的调用会在新的 assistant 工具调用消息前插入一条 user 消息解释参数为何改变，否则 LLM 不知道参数被人工改动、会再次用原参数调用；
4. _update_chat_history（strategies.py）：定位最后一条 user 消息与最后一条 tool 消息，取两者索引较大者作为插入点，把拒绝消息与确认/修改后的工具调用消息拼接回去。

最终返回的对话历史以「确认/修改后的工具调用」收尾，Agent 据此执行真正的工具调用。同步与异步路径（run/run_async、_run_confirmation_strategies/_run_confirmation_strategies_async）结构对称，异步侧通过_execute_component_async驱动策略执行（strategies.py）。上述行为均有对应测试覆盖，见 test/hooks/human_in_the_loop/ 下的test_strategies.py、test_hooks.py、test_policies.py、test_user_interfaces.py、test_dataclasses.py，以及端到端的test/components/agents/test_agent_hitl.py。

Web / 服务端场景：hook_context与自定义 UI、Strategy

控制台 UI 只适合本地或命令行环境。对于 Web 服务，协议层预留了完整的扩展点：

• confirmation_strategy_context：run()/run_async()的可选参数，用于传递请求级资源（WebSocket 连接、异步队列、Redis pub/sub 客户端等），供策略在非阻塞场景下与前端交互（strategies.py）；
• 透传方式：通过 Agent 的hook_context运行参数传入（agent.run(messages=[...], hook_context={"redis": client})），ConfirmationHook用state.data.get("hook_context")读取并传给每个策略。注意这里刻意用state.data而非state.get——后者会深拷贝，破坏 WebSocket、客户端等不可拷贝的存活资源（hooks.py）；
• 自定义 UI：实现ConfirmationUI即可接入自己的审批系统，例如发送 Webhook 等待异步审批结果：

from haystack.hooks.human_in_the_loop.types import ConfirmationUI from haystack.hooks.human_in_the_loop import ConfirmationUIResult from typing import Any class WebhookApprovalUI(ConfirmationUI): """Sends a webhook and waits for an async approval response.""" def get_user_confirmation( self, tool_name: str, tool_description: str, tool_params: dict[str, Any], ) -> ConfirmationUIResult: # 向你的系统发送审批请求并等待响应 response = send_approval_request_and_wait(tool_name, tool_params) return ConfirmationUIResult( action=response["action"], feedback=response.get("feedback"), )

一个典型的服务端模式是：策略在工具调用前发出tool_call_startSSE 事件并阻塞等待（如 RedisBLPOP），前端 Web UI 收到事件后弹出确认对话框，用户选择后把approved/rejected写回（如 RedisLPUSH），策略解除阻塞并返回ToolExecutionDecision，Agent 继续执行。这正是SimpleConsoleUI/RichConsoleUI不适合的 Web 环境下推荐的扩展方向。

注意事项小结

• HITL 只看模型产生的参数：策略呈现给人工确认的，仅是模型为一次工具调用生成的参数；通过工具inputs_from_state映射从State注入的值不参与确认展示，注入发生在真正执行阶段；
• Hook 位置受限：ConfirmationHook只能在before_tool钩子点使用（工具调用已由模型请求、尚未执行之间），Agent 会在构造时强制校验；
• 策略表为空即放行：没有为某工具配置策略时，该工具直接执行、不询问；通配符"*"可兜底所有未单独配置的工具；
• 状态型策略要注意序列化：AskOncePolicy这类带内部状态的策略，其记忆存于实例字段，跨运行持久化需结合整体 Agent 序列化链路考虑。

至此，从ConfirmationUIResult/ToolExecutionDecision两个数据类，到三层决策体系（Policy → Strategy → UI），再到ConfirmationHook与底层执行管线，你已经掌握了 Haystack HITL 模块从 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