1. 项目概述为什么我们需要在AI流程中引入“人”的审批在构建基于LangChain的自动化AI应用时我们常常会面临一个两难境地一方面我们希望流程能全自动、高效地运行减少人工干预另一方面我们又必须对某些关键或敏感操作保持绝对的控制权避免AI“自作主张”带来的风险。比如一个自动化的客户服务Agent如果被授权可以直接从数据库删除用户记录或者一个营销文案生成工具如果未经审核就向全网发布内容这其中的潜在风险是巨大的。这就是HITLHuman-In-The-Loop人机协同机制的核心价值所在。它不是在否定AI的能力而是在关键的决策节点上巧妙地引入人类的判断力形成一道安全阀。LangChain 1.x版本中的HumanInTheLoopMiddleware中间件正是为此而生。它允许我们在Agent执行工具调用的链条中插入一个“暂停”点将待执行的操作特别是工具调用提交给人类审批只有获得批准后流程才会继续。简单来说这个项目就是教你如何给你的LangChain智能体Agent装上“刹车”和“方向盘”让它在执行敏感操作前必须停下来等你点头。这不仅仅是技术实现更是一种负责任AI开发的工程实践。无论你是开发内部数据查询工具、金融风控助手还是内容审核流水线掌握HITL都能让你的应用更加可靠、合规。2. 核心组件与原理深度拆解要理解并实现HITL我们需要先吃透几个核心概念它们共同构成了人机协同的骨架。2.1 HumanInTheLoopMiddleware流程的“交通警察”HumanInTheLoopMiddleware是LangChain提供的一个标准中间件。中间件Middleware的设计模式在软件开发中很常见它允许你在核心处理逻辑的前后插入自定义代码。在LangChain的Agent执行上下文中中间件可以拦截、修改或响应Agent执行过程中的特定事件。这个中间件主要拦截的是AgentAction事件。当Agent决定要调用一个工具Tool时会产生一个AgentAction对象其中包含了要调用哪个工具、传入什么参数等信息。HITL中间件的作用就是捕获这个AgentAction暂停当前执行流将动作详情工具名、参数以某种方式如打印到控制台、发送到Webhook、写入数据库呈现给人类审批者然后等待一个“继续”或“终止”的指令。它的工作流程可以类比为一个严格的报销审批系统员工Agent提交报销单AgentAction系统中间件将单据转给经理人类经理审核后选择“批准”或“驳回”系统再根据批示决定是打款还是将驳回原因反馈给员工。2.2 Checkpointer状态的“存档点”HITL流程涉及“暂停”和“恢复”这就要求系统必须有能力保存和加载执行状态。这就是Checkpointer的用武之地。你可以把它想象成游戏中的存档点。当中间件拦截到一个需要人工审批的AgentAction时它需要将当前Agent的完整状态包括之前的对话历史、已执行的动作、中间变量等序列化并保存起来。等到人工审批完成系统再根据一个唯一的标识如checkpoint_id加载之前保存的状态并从暂停点继续执行。LangChain提供了基础的BaseCheckpointSaver接口你可以根据需要实现内存存储、文件存储或数据库存储。对于简单的演示或单次会话使用内存存储MemorySaver就足够了但对于生产环境你可能需要实现一个基于Redis或PostgreSQL的持久化Checkpointer以支持多用户、长时间运行的会话。2.3 决策逻辑人类的“指挥棒”当人类审批者看到待审批的操作后他需要做出决策。LangChain的HITL中间件通常支持几种基本的决策类型这构成了人机交互的协议继续Proceed批准该操作。中间件将允许Agent执行这个被拦截的工具调用。忽略Ignore跳过此操作。中间件会告诉Agent这个工具调用失败了或返回一个空结果促使Agent重新思考并可能选择其他路径。修改Modify修改操作参数后继续。审批者可以调整工具调用的输入参数然后以修改后的参数继续执行。这需要前端界面提供参数编辑能力。终止Terminate直接终止整个Agent运行。用于当审批者发现Agent行为完全偏离预期需要紧急停止的场景。在实际实现中我们通常会将这几种决策封装成一个枚举类并通过一个回调函数或消息队列将决策结果传回给等待中的中间件。3. 四种决策模式的实战应用与代码实现理论讲完了我们直接上代码看看如何具体实现这四种决策。我们将构建一个模拟“数据库查询助手”的Agent它有一个高危工具delete_user_record任何删除操作都必须经过人工审批。3.1 基础环境搭建与工具定义首先安装依赖并定义工具。# 假设使用 OpenAI 模型你需要设置自己的 API_KEY import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain_core.tools import tool from langchain_core.prompts import PromptTemplate # 1. 定义一个危险的工具删除用户记录 tool def delete_user_record(user_id: str) - str: 根据用户ID删除其所有记录。这是一个危险操作需要谨慎使用。 # 这里应该是真实的数据库删除逻辑例如 # db.execute(f“DELETE FROM users WHERE id {user_id}”) return f“用户 {user_id} 的记录已删除模拟操作。” # 2. 定义一个安全的工具查询用户信息 tool def query_user_info(user_id: str) - str: 根据用户ID查询用户基本信息。 return f“用户 {user_id} 的信息姓名‘测试用户’注册于2023-01-01模拟数据。” # 将工具放入列表 tools [delete_user_record, query_user_info] # 3. 创建LLM和Agent llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) prompt PromptTemplate.from_template(“”” 你是一个数据库助手。请根据用户问题谨慎地使用工具来回答问题。 你可以使用的工具有{tools}。 用户问题{input} 思考过程{agent_scratchpad} “””) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)现在如果我们直接运行agent_executor.invoke({“input”: “删除用户ID为123的记录”})AI会直接调用删除工具这很危险。接下来我们引入HITL。3.2 实现自定义HITL中间件与Checkpointer我们将实现一个简单的、基于控制台输入的HITL中间件。在生产环境中你需要将其替换为WebSocket、HTTP回调或集成到你的任务管理系统中。from langchain_core.agents import AgentAction from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List, Optional from enum import Enum import json # 定义人类决策枚举 class HumanDecision(Enum): PROCEED “proceed” # 继续执行 IGNORE “ignore” # 忽略此工具调用 MODIFY “modify” # 修改参数后执行 TERMINATE “terminate” # 终止整个流程 # 一个简单的内存Checkpointer class SimpleMemoryCheckpointer: def __init__(self): self._storage {} def save(self, checkpoint_id: str, state: Dict[str, Any]): self._storage[checkpoint_id] json.dumps(state) def load(self, checkpoint_id: str) - Optional[Dict[str, Any]]: data self._storage.get(checkpoint_id) return json.loads(data) if data else None # 自定义HITL回调处理器一种实现中间件逻辑的方式 class HumanInTheLoopCallbackHandler(BaseCallbackHandler): def __init__(self, checkpointer: SimpleMemoryCheckpointer, sensitive_tools: List[str]): self.checkpointer checkpointer self.sensitive_tools sensitive_tools # 需要审批的工具名列表 self.current_checkpoint_id None def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: # 检查当前动作的工具是否在敏感列表内 if action.tool in self.sensitive_tools: print(f“\n⚠️ [HITL 拦截] 检测到敏感工具调用”) print(f“ 工具: {action.tool}”) print(f“ 参数: {action.tool_input}”) print(f“ 日志: {action.log}”) # 生成一个检查点ID并保存当前状态这里简化了状态获取实际应从kwargs[‘run_id’]等获取更多上下文 import uuid self.current_checkpoint_id str(uuid.uuid4()) # 注意这里需要保存完整的Agent状态实际项目中状态结构更复杂 snapshot { “agent_action”: action.dict(), “run_id”: kwargs.get(‘run_id’), # 应包含更多的链条状态... } self.checkpointer.save(self.current_checkpoint_id, snapshot) print(f“ 状态已保存至检查点: {self.current_checkpoint_id}”) # 模拟人工审批流程控制台输入 decision self._await_human_decision(action) # 根据决策处理 return self._handle_decision(decision, action) # 非敏感工具直接放行 return None def _await_human_decision(self, action: AgentAction) - HumanDecision: “”“模拟等待人类决策。在实际中这里可能是等待一个API回调或从消息队列中读取结果。”“” print(“\n请做出审批决策”) print(“ 1 - 批准执行 (Proceed)”) print(“ 2 - 忽略此调用 (Ignore)”) print(“ 3 - 修改参数 (Modify)”) print(“ 4 - 终止任务 (Terminate)”) while True: choice input(“请输入选择 (1/2/3/4): “).strip() if choice ‘1’: return HumanDecision.PROCEED elif choice ‘2’: return HumanDecision.IGNORE elif choice ‘3’: return HumanDecision.MODIFY elif choice ‘4’: return HumanDecision.TERMINATE else: print(“输入无效请重新选择。”) def _handle_decision(self, decision: HumanDecision, original_action: AgentAction): “”“根据人类决策返回相应的指令以控制Agent执行流。”“” if decision HumanDecision.PROCEED: print(“[审批结果] 已批准。继续执行工具调用。”) # 返回NoneLangChain会继续正常执行这个AgentAction return None elif decision HumanDecision.IGNORE: print(“[审批结果] 已忽略。模拟工具调用失败促使Agent重新规划。”) # 抛出一个特殊的异常或返回一个错误结果让Agent认为工具调用失败 # 这里我们通过返回一个包含错误信息的字典来模拟 raise ValueError(f“Human denied the call to tool ‘{original_action.tool}’. Agent must reconsider.”) elif decision HumanDecision.MODIFY: new_input input(f“请输入修改后的参数 (原参数: {original_action.tool_input}): “).strip() print(f“[审批结果] 参数已修改为: {new_input}。将以新参数执行。”) # 创建一个修改后的AgentAction modified_action AgentAction( tooloriginal_action.tool, tool_inputnew_input, logoriginal_action.log “\n[参数已被人工修改]” ) # 需要替换原来的action这里通过抛出包含新action的异常是一种方式依赖具体LangChain版本 # 更稳健的做法是在自定义的AgentExecutor中处理 return modified_action elif decision HumanDecision.TERMINATE: print(“[审批结果] 任务被人工终止。”) # 终止整个执行 raise KeyboardInterrupt(“Agent execution terminated by human.”)3.3 集成与运行测试现在我们将这个HITL回调处理器集成到AgentExecutor中。# 初始化Checkpointer和HITL Handler checkpointer SimpleMemoryCheckpointer() hitl_handler HumanInTheLoopCallbackHandler( checkpointercheckpointer, sensitive_tools[“delete_user_record”] # 只对删除工具进行拦截 ) # 创建带有HITL回调的Executor agent_executor_with_hitl AgentExecutor( agentagent, toolstools, verboseTrue, callbacks[hitl_handler] # 传入我们的回调处理器 ) # 测试场景1触发敏感操作删除 print(“ 测试1尝试删除用户 ) try: result agent_executor_with_hitl.invoke({“input”: “请删除用户ID为‘U1001’的记录。”}) print(“最终结果:”, result) except (ValueError, KeyboardInterrupt) as e: print(f“流程因审批决策而中断: {e}”) # 测试场景2安全操作查询应不受影响 print(“\n 测试2查询用户信息应直接通过 ) result agent_executor_with_hitl.invoke({“input”: “查询用户ID为‘U1001’的信息。”}) print(“查询结果:”, result[‘output’])运行上述代码当Agent尝试调用delete_user_record时流程会被暂停并在控制台等待你的输入。输入1批准你会看到删除操作被执行输入2Agent会收到一个“失败”信号它可能会尝试其他方法或直接回答无法操作输入3你可以修改要删除的user_id输入4整个Agent会话会立即停止。实操心得状态管理的复杂性上面的示例极大地简化了状态保存Checkpoint的逻辑。在实际项目中AgentExecutor的内部状态可能非常复杂包含多轮对话历史、工具输出、中间思维链等。直接使用BaseCallbackHandler的on_agent_action钩子来完整实现HITL和状态恢复可能会比较棘手。更成熟的做法是使用LangChain的RunnableWithMessageHistory或直接基于LangGraphLangChain的新编排框架来构建流程它们的检查点机制更为完善和强大。对于生产级应用我强烈建议评估LangGraph它原生支持在状态图的任何节点设置检查点并等待外部信号如人工审批后再继续架构上更清晰。4. 条件拦截更精细化的控制策略我们之前的实现是“工具黑名单”模式即列出所有需要审批的工具。但在更复杂的场景下我们可能需要基于动态条件来决定是否触发拦截。这就是条件拦截。4.1 基于参数内容的拦截例如我们可能只拦截“删除管理员账户”的操作而对删除普通用户放行。我们需要在中间件里解析工具输入参数。class ConditionalHITLHandler(HumanInTheLoopCallbackHandler): def __init__(self, checkpointer: SimpleMemoryCheckpointer, sensitive_tools: List[str]): super().__init__(checkpointer, sensitive_tools) def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: # 基础检查工具名 if action.tool not in self.sensitive_tools: return None # 进阶基于参数的条件判断 tool_input action.tool_input if action.tool “delete_user_record”: user_id tool_input.get(“user_id”) if isinstance(tool_input, dict) else tool_input # 假设‘admin’开头的用户ID是管理员 if isinstance(user_id, str) and user_id.startswith(“admin”): print(f“⚠️ [条件拦截] 尝试删除管理员账户 ‘{user_id}’触发强制审批。”) return super().on_agent_action(action, **kwargs) else: print(f“[条件放行] 删除普通用户 ‘{user_id}’无需审批。”) return None # 直接放行 # 对于其他敏感工具仍执行默认审批 return super().on_agent_action(action, **kwargs)4.2 基于执行上下文的拦截有时拦截与否取决于整个对话历史或之前的操作。例如同一个“发送邮件”工具如果是在客服流程中自动发送确认函可以放行但如果是在营销流程中首次向用户发送推广邮件则需要审批。def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: if action.tool “send_email”: # 我们需要访问之前的对话历史或Agent的“记忆” # 在callback中可以通过kwargs[‘parent_run_id’]等获取运行树进而查询历史消息 # 这里是一个简化示例假设我们能从某个地方拿到对话摘要 conversation_summary self._get_conversation_topic(kwargs.get(‘run_id’)) if “营销” in conversation_summary and “首次” in conversation_summary: print(“⚠️ [上下文拦截] 营销场景首次发送邮件需审批。”) return super().on_agent_action(action, **kwargs) return None注意事项性能与状态依赖条件拦截增加了每次工具调用前的计算开销。复杂的条件判断如调用另一个LLM分析对话历史会显著影响延迟。此外条件判断严重依赖于能否从回调上下文中获取到准确的运行时状态如完整的对话历史。在LangChain 1.x中这有时并不直接可能需要通过自定义AgentExecutor或使用LangGraph来更优雅地访问和传递这些上下文信息。5. 生产级架构思考与常见问题排查将HITL从Demo推向生产你会遇到一系列新的挑战。下面分享一些实战经验和避坑指南。5.1 生产架构设计一个典型的生产级HITL系统包含以下组件AI工作流引擎运行LangChain Agent或LangGraph。负责执行并在预设节点暂停。审批任务队列如RabbitMQ、Redis Stream或数据库任务表。当工作流暂停时向此队列投递一个审批任务。审批管理后台一个Web应用从队列中拉取任务以友好界面展示工具、参数、对话上下文呈现给审批员并收集决策结果。决策回调接口审批后台通过HTTP Webhook或消息队列将决策结果回传给工作流引擎。持久化Checkpointer使用Redis或数据库存储检查点状态支持高可用和会话恢复。关键点工作流引擎和审批后台应解耦。引擎投递任务后即可释放资源或挂起等待回调。这避免了长期占用连接也便于水平扩展。5.2 常见问题与解决方案实录问题1审批超时了怎么办Agent暂停等待审批不能无限期等下去。你需要设置一个超时时间例如30分钟。超时后系统可以自动执行一个默认决策如“忽略”或“终止”并将任务标记为“超时关闭”通知相关人员。解决方案在投递审批任务时设置一个expires_at时间戳。后台有一个定时任务扫描超时未处理的任务执行默认策略并通知工作流引擎。问题2审批过程中原始用户又发送了新消息这涉及到会话管理。如果检查点保存了完整的会话状态新消息应该被放入一个等待队列或者直接回复用户“您的请求正在处理中请稍候”。更复杂的处理是允许新消息开启一个全新的、独立的Agent会话但这需要仔细设计以避免状态混乱。问题3如何让审批者了解完整的上下文仅仅提供工具名和参数是不够的。审批者需要知道“AI为什么想这么做”。因此在保存检查点时必须连同导致该动作的Agent思考过程scratchpad和最近的几条对话历史一并保存并在审批界面中清晰展示。问题4HITL中间件导致Agent执行异常如何调试LangChain的中间件/回调系统有时错误信息不直观。建议开启verboseTrue观察日志流。在回调函数的各个阶段添加详细的打印日志。重点检查on_agent_action方法的返回值。返回None通常表示放行如果希望阻止默认行为可能需要抛出特定异常或返回一个包含AgentFinish的列表这取决于具体版本和架构。最稳妥的方法是参考LangChain官方关于创建自定义AgentExecutor的文档。问题5在LangGraph中如何更优雅地实现HITLLangGraph通过“状态图”和“节点”的概念使得HITL的实现变得直观。你可以定义一个human_review_node它的状态包含需要审批的proposed_action。这个节点执行后会进入一个暂停状态waiting直到外部系统如你的审批后台通过Graph.update_state()方法向该节点的状态中写入review_decision。然后图会根据这个决策值通过条件边conditional_edge路由到不同的下一个节点如execute_action_node或skip_action_node。这是比在回调中“拦截”更声明式、更易管理的方式。最后我想强调的是引入HITL不仅仅是添加一个技术组件它意味着你的AI应用从“全自动”转向了“人机协同”的新范式。你需要重新思考用户体验审批延迟、运营成本需要多少审批员、以及流程设计什么操作需要审批、审批粒度如何。从最简单的“高危工具全拦截”开始逐步迭代出符合你业务需求的、精细化的协同规则才是稳健之道。
1. 项目概述为什么我们需要在AI流程中引入“人”的审批在构建基于LangChain的自动化AI应用时我们常常会面临一个两难境地一方面我们希望流程能全自动、高效地运行减少人工干预另一方面我们又必须对某些关键或敏感操作保持绝对的控制权避免AI“自作主张”带来的风险。比如一个自动化的客户服务Agent如果被授权可以直接从数据库删除用户记录或者一个营销文案生成工具如果未经审核就向全网发布内容这其中的潜在风险是巨大的。这就是HITLHuman-In-The-Loop人机协同机制的核心价值所在。它不是在否定AI的能力而是在关键的决策节点上巧妙地引入人类的判断力形成一道安全阀。LangChain 1.x版本中的HumanInTheLoopMiddleware中间件正是为此而生。它允许我们在Agent执行工具调用的链条中插入一个“暂停”点将待执行的操作特别是工具调用提交给人类审批只有获得批准后流程才会继续。简单来说这个项目就是教你如何给你的LangChain智能体Agent装上“刹车”和“方向盘”让它在执行敏感操作前必须停下来等你点头。这不仅仅是技术实现更是一种负责任AI开发的工程实践。无论你是开发内部数据查询工具、金融风控助手还是内容审核流水线掌握HITL都能让你的应用更加可靠、合规。2. 核心组件与原理深度拆解要理解并实现HITL我们需要先吃透几个核心概念它们共同构成了人机协同的骨架。2.1 HumanInTheLoopMiddleware流程的“交通警察”HumanInTheLoopMiddleware是LangChain提供的一个标准中间件。中间件Middleware的设计模式在软件开发中很常见它允许你在核心处理逻辑的前后插入自定义代码。在LangChain的Agent执行上下文中中间件可以拦截、修改或响应Agent执行过程中的特定事件。这个中间件主要拦截的是AgentAction事件。当Agent决定要调用一个工具Tool时会产生一个AgentAction对象其中包含了要调用哪个工具、传入什么参数等信息。HITL中间件的作用就是捕获这个AgentAction暂停当前执行流将动作详情工具名、参数以某种方式如打印到控制台、发送到Webhook、写入数据库呈现给人类审批者然后等待一个“继续”或“终止”的指令。它的工作流程可以类比为一个严格的报销审批系统员工Agent提交报销单AgentAction系统中间件将单据转给经理人类经理审核后选择“批准”或“驳回”系统再根据批示决定是打款还是将驳回原因反馈给员工。2.2 Checkpointer状态的“存档点”HITL流程涉及“暂停”和“恢复”这就要求系统必须有能力保存和加载执行状态。这就是Checkpointer的用武之地。你可以把它想象成游戏中的存档点。当中间件拦截到一个需要人工审批的AgentAction时它需要将当前Agent的完整状态包括之前的对话历史、已执行的动作、中间变量等序列化并保存起来。等到人工审批完成系统再根据一个唯一的标识如checkpoint_id加载之前保存的状态并从暂停点继续执行。LangChain提供了基础的BaseCheckpointSaver接口你可以根据需要实现内存存储、文件存储或数据库存储。对于简单的演示或单次会话使用内存存储MemorySaver就足够了但对于生产环境你可能需要实现一个基于Redis或PostgreSQL的持久化Checkpointer以支持多用户、长时间运行的会话。2.3 决策逻辑人类的“指挥棒”当人类审批者看到待审批的操作后他需要做出决策。LangChain的HITL中间件通常支持几种基本的决策类型这构成了人机交互的协议继续Proceed批准该操作。中间件将允许Agent执行这个被拦截的工具调用。忽略Ignore跳过此操作。中间件会告诉Agent这个工具调用失败了或返回一个空结果促使Agent重新思考并可能选择其他路径。修改Modify修改操作参数后继续。审批者可以调整工具调用的输入参数然后以修改后的参数继续执行。这需要前端界面提供参数编辑能力。终止Terminate直接终止整个Agent运行。用于当审批者发现Agent行为完全偏离预期需要紧急停止的场景。在实际实现中我们通常会将这几种决策封装成一个枚举类并通过一个回调函数或消息队列将决策结果传回给等待中的中间件。3. 四种决策模式的实战应用与代码实现理论讲完了我们直接上代码看看如何具体实现这四种决策。我们将构建一个模拟“数据库查询助手”的Agent它有一个高危工具delete_user_record任何删除操作都必须经过人工审批。3.1 基础环境搭建与工具定义首先安装依赖并定义工具。# 假设使用 OpenAI 模型你需要设置自己的 API_KEY import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain_core.tools import tool from langchain_core.prompts import PromptTemplate # 1. 定义一个危险的工具删除用户记录 tool def delete_user_record(user_id: str) - str: 根据用户ID删除其所有记录。这是一个危险操作需要谨慎使用。 # 这里应该是真实的数据库删除逻辑例如 # db.execute(f“DELETE FROM users WHERE id {user_id}”) return f“用户 {user_id} 的记录已删除模拟操作。” # 2. 定义一个安全的工具查询用户信息 tool def query_user_info(user_id: str) - str: 根据用户ID查询用户基本信息。 return f“用户 {user_id} 的信息姓名‘测试用户’注册于2023-01-01模拟数据。” # 将工具放入列表 tools [delete_user_record, query_user_info] # 3. 创建LLM和Agent llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) prompt PromptTemplate.from_template(“”” 你是一个数据库助手。请根据用户问题谨慎地使用工具来回答问题。 你可以使用的工具有{tools}。 用户问题{input} 思考过程{agent_scratchpad} “””) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue)现在如果我们直接运行agent_executor.invoke({“input”: “删除用户ID为123的记录”})AI会直接调用删除工具这很危险。接下来我们引入HITL。3.2 实现自定义HITL中间件与Checkpointer我们将实现一个简单的、基于控制台输入的HITL中间件。在生产环境中你需要将其替换为WebSocket、HTTP回调或集成到你的任务管理系统中。from langchain_core.agents import AgentAction from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List, Optional from enum import Enum import json # 定义人类决策枚举 class HumanDecision(Enum): PROCEED “proceed” # 继续执行 IGNORE “ignore” # 忽略此工具调用 MODIFY “modify” # 修改参数后执行 TERMINATE “terminate” # 终止整个流程 # 一个简单的内存Checkpointer class SimpleMemoryCheckpointer: def __init__(self): self._storage {} def save(self, checkpoint_id: str, state: Dict[str, Any]): self._storage[checkpoint_id] json.dumps(state) def load(self, checkpoint_id: str) - Optional[Dict[str, Any]]: data self._storage.get(checkpoint_id) return json.loads(data) if data else None # 自定义HITL回调处理器一种实现中间件逻辑的方式 class HumanInTheLoopCallbackHandler(BaseCallbackHandler): def __init__(self, checkpointer: SimpleMemoryCheckpointer, sensitive_tools: List[str]): self.checkpointer checkpointer self.sensitive_tools sensitive_tools # 需要审批的工具名列表 self.current_checkpoint_id None def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: # 检查当前动作的工具是否在敏感列表内 if action.tool in self.sensitive_tools: print(f“\n⚠️ [HITL 拦截] 检测到敏感工具调用”) print(f“ 工具: {action.tool}”) print(f“ 参数: {action.tool_input}”) print(f“ 日志: {action.log}”) # 生成一个检查点ID并保存当前状态这里简化了状态获取实际应从kwargs[‘run_id’]等获取更多上下文 import uuid self.current_checkpoint_id str(uuid.uuid4()) # 注意这里需要保存完整的Agent状态实际项目中状态结构更复杂 snapshot { “agent_action”: action.dict(), “run_id”: kwargs.get(‘run_id’), # 应包含更多的链条状态... } self.checkpointer.save(self.current_checkpoint_id, snapshot) print(f“ 状态已保存至检查点: {self.current_checkpoint_id}”) # 模拟人工审批流程控制台输入 decision self._await_human_decision(action) # 根据决策处理 return self._handle_decision(decision, action) # 非敏感工具直接放行 return None def _await_human_decision(self, action: AgentAction) - HumanDecision: “”“模拟等待人类决策。在实际中这里可能是等待一个API回调或从消息队列中读取结果。”“” print(“\n请做出审批决策”) print(“ 1 - 批准执行 (Proceed)”) print(“ 2 - 忽略此调用 (Ignore)”) print(“ 3 - 修改参数 (Modify)”) print(“ 4 - 终止任务 (Terminate)”) while True: choice input(“请输入选择 (1/2/3/4): “).strip() if choice ‘1’: return HumanDecision.PROCEED elif choice ‘2’: return HumanDecision.IGNORE elif choice ‘3’: return HumanDecision.MODIFY elif choice ‘4’: return HumanDecision.TERMINATE else: print(“输入无效请重新选择。”) def _handle_decision(self, decision: HumanDecision, original_action: AgentAction): “”“根据人类决策返回相应的指令以控制Agent执行流。”“” if decision HumanDecision.PROCEED: print(“[审批结果] 已批准。继续执行工具调用。”) # 返回NoneLangChain会继续正常执行这个AgentAction return None elif decision HumanDecision.IGNORE: print(“[审批结果] 已忽略。模拟工具调用失败促使Agent重新规划。”) # 抛出一个特殊的异常或返回一个错误结果让Agent认为工具调用失败 # 这里我们通过返回一个包含错误信息的字典来模拟 raise ValueError(f“Human denied the call to tool ‘{original_action.tool}’. Agent must reconsider.”) elif decision HumanDecision.MODIFY: new_input input(f“请输入修改后的参数 (原参数: {original_action.tool_input}): “).strip() print(f“[审批结果] 参数已修改为: {new_input}。将以新参数执行。”) # 创建一个修改后的AgentAction modified_action AgentAction( tooloriginal_action.tool, tool_inputnew_input, logoriginal_action.log “\n[参数已被人工修改]” ) # 需要替换原来的action这里通过抛出包含新action的异常是一种方式依赖具体LangChain版本 # 更稳健的做法是在自定义的AgentExecutor中处理 return modified_action elif decision HumanDecision.TERMINATE: print(“[审批结果] 任务被人工终止。”) # 终止整个执行 raise KeyboardInterrupt(“Agent execution terminated by human.”)3.3 集成与运行测试现在我们将这个HITL回调处理器集成到AgentExecutor中。# 初始化Checkpointer和HITL Handler checkpointer SimpleMemoryCheckpointer() hitl_handler HumanInTheLoopCallbackHandler( checkpointercheckpointer, sensitive_tools[“delete_user_record”] # 只对删除工具进行拦截 ) # 创建带有HITL回调的Executor agent_executor_with_hitl AgentExecutor( agentagent, toolstools, verboseTrue, callbacks[hitl_handler] # 传入我们的回调处理器 ) # 测试场景1触发敏感操作删除 print(“ 测试1尝试删除用户 ) try: result agent_executor_with_hitl.invoke({“input”: “请删除用户ID为‘U1001’的记录。”}) print(“最终结果:”, result) except (ValueError, KeyboardInterrupt) as e: print(f“流程因审批决策而中断: {e}”) # 测试场景2安全操作查询应不受影响 print(“\n 测试2查询用户信息应直接通过 ) result agent_executor_with_hitl.invoke({“input”: “查询用户ID为‘U1001’的信息。”}) print(“查询结果:”, result[‘output’])运行上述代码当Agent尝试调用delete_user_record时流程会被暂停并在控制台等待你的输入。输入1批准你会看到删除操作被执行输入2Agent会收到一个“失败”信号它可能会尝试其他方法或直接回答无法操作输入3你可以修改要删除的user_id输入4整个Agent会话会立即停止。实操心得状态管理的复杂性上面的示例极大地简化了状态保存Checkpoint的逻辑。在实际项目中AgentExecutor的内部状态可能非常复杂包含多轮对话历史、工具输出、中间思维链等。直接使用BaseCallbackHandler的on_agent_action钩子来完整实现HITL和状态恢复可能会比较棘手。更成熟的做法是使用LangChain的RunnableWithMessageHistory或直接基于LangGraphLangChain的新编排框架来构建流程它们的检查点机制更为完善和强大。对于生产级应用我强烈建议评估LangGraph它原生支持在状态图的任何节点设置检查点并等待外部信号如人工审批后再继续架构上更清晰。4. 条件拦截更精细化的控制策略我们之前的实现是“工具黑名单”模式即列出所有需要审批的工具。但在更复杂的场景下我们可能需要基于动态条件来决定是否触发拦截。这就是条件拦截。4.1 基于参数内容的拦截例如我们可能只拦截“删除管理员账户”的操作而对删除普通用户放行。我们需要在中间件里解析工具输入参数。class ConditionalHITLHandler(HumanInTheLoopCallbackHandler): def __init__(self, checkpointer: SimpleMemoryCheckpointer, sensitive_tools: List[str]): super().__init__(checkpointer, sensitive_tools) def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: # 基础检查工具名 if action.tool not in self.sensitive_tools: return None # 进阶基于参数的条件判断 tool_input action.tool_input if action.tool “delete_user_record”: user_id tool_input.get(“user_id”) if isinstance(tool_input, dict) else tool_input # 假设‘admin’开头的用户ID是管理员 if isinstance(user_id, str) and user_id.startswith(“admin”): print(f“⚠️ [条件拦截] 尝试删除管理员账户 ‘{user_id}’触发强制审批。”) return super().on_agent_action(action, **kwargs) else: print(f“[条件放行] 删除普通用户 ‘{user_id}’无需审批。”) return None # 直接放行 # 对于其他敏感工具仍执行默认审批 return super().on_agent_action(action, **kwargs)4.2 基于执行上下文的拦截有时拦截与否取决于整个对话历史或之前的操作。例如同一个“发送邮件”工具如果是在客服流程中自动发送确认函可以放行但如果是在营销流程中首次向用户发送推广邮件则需要审批。def on_agent_action(self, action: AgentAction, **kwargs: Any) - Any: if action.tool “send_email”: # 我们需要访问之前的对话历史或Agent的“记忆” # 在callback中可以通过kwargs[‘parent_run_id’]等获取运行树进而查询历史消息 # 这里是一个简化示例假设我们能从某个地方拿到对话摘要 conversation_summary self._get_conversation_topic(kwargs.get(‘run_id’)) if “营销” in conversation_summary and “首次” in conversation_summary: print(“⚠️ [上下文拦截] 营销场景首次发送邮件需审批。”) return super().on_agent_action(action, **kwargs) return None注意事项性能与状态依赖条件拦截增加了每次工具调用前的计算开销。复杂的条件判断如调用另一个LLM分析对话历史会显著影响延迟。此外条件判断严重依赖于能否从回调上下文中获取到准确的运行时状态如完整的对话历史。在LangChain 1.x中这有时并不直接可能需要通过自定义AgentExecutor或使用LangGraph来更优雅地访问和传递这些上下文信息。5. 生产级架构思考与常见问题排查将HITL从Demo推向生产你会遇到一系列新的挑战。下面分享一些实战经验和避坑指南。5.1 生产架构设计一个典型的生产级HITL系统包含以下组件AI工作流引擎运行LangChain Agent或LangGraph。负责执行并在预设节点暂停。审批任务队列如RabbitMQ、Redis Stream或数据库任务表。当工作流暂停时向此队列投递一个审批任务。审批管理后台一个Web应用从队列中拉取任务以友好界面展示工具、参数、对话上下文呈现给审批员并收集决策结果。决策回调接口审批后台通过HTTP Webhook或消息队列将决策结果回传给工作流引擎。持久化Checkpointer使用Redis或数据库存储检查点状态支持高可用和会话恢复。关键点工作流引擎和审批后台应解耦。引擎投递任务后即可释放资源或挂起等待回调。这避免了长期占用连接也便于水平扩展。5.2 常见问题与解决方案实录问题1审批超时了怎么办Agent暂停等待审批不能无限期等下去。你需要设置一个超时时间例如30分钟。超时后系统可以自动执行一个默认决策如“忽略”或“终止”并将任务标记为“超时关闭”通知相关人员。解决方案在投递审批任务时设置一个expires_at时间戳。后台有一个定时任务扫描超时未处理的任务执行默认策略并通知工作流引擎。问题2审批过程中原始用户又发送了新消息这涉及到会话管理。如果检查点保存了完整的会话状态新消息应该被放入一个等待队列或者直接回复用户“您的请求正在处理中请稍候”。更复杂的处理是允许新消息开启一个全新的、独立的Agent会话但这需要仔细设计以避免状态混乱。问题3如何让审批者了解完整的上下文仅仅提供工具名和参数是不够的。审批者需要知道“AI为什么想这么做”。因此在保存检查点时必须连同导致该动作的Agent思考过程scratchpad和最近的几条对话历史一并保存并在审批界面中清晰展示。问题4HITL中间件导致Agent执行异常如何调试LangChain的中间件/回调系统有时错误信息不直观。建议开启verboseTrue观察日志流。在回调函数的各个阶段添加详细的打印日志。重点检查on_agent_action方法的返回值。返回None通常表示放行如果希望阻止默认行为可能需要抛出特定异常或返回一个包含AgentFinish的列表这取决于具体版本和架构。最稳妥的方法是参考LangChain官方关于创建自定义AgentExecutor的文档。问题5在LangGraph中如何更优雅地实现HITLLangGraph通过“状态图”和“节点”的概念使得HITL的实现变得直观。你可以定义一个human_review_node它的状态包含需要审批的proposed_action。这个节点执行后会进入一个暂停状态waiting直到外部系统如你的审批后台通过Graph.update_state()方法向该节点的状态中写入review_decision。然后图会根据这个决策值通过条件边conditional_edge路由到不同的下一个节点如execute_action_node或skip_action_node。这是比在回调中“拦截”更声明式、更易管理的方式。最后我想强调的是引入HITL不仅仅是添加一个技术组件它意味着你的AI应用从“全自动”转向了“人机协同”的新范式。你需要重新思考用户体验审批延迟、运营成本需要多少审批员、以及流程设计什么操作需要审批、审批粒度如何。从最简单的“高危工具全拦截”开始逐步迭代出符合你业务需求的、精细化的协同规则才是稳健之道。
