1. 项目概述当AI需要“刹车”时在构建基于LangChain的自动化AI应用时我们常常会陷入一种“效率至上”的迷思追求更快的响应、更少的干预、更流畅的端到端流程。然而当AI代理Agent开始处理涉及资金转账、内容发布、数据删除或关键决策等敏感操作时这种“全自动”模式就变得异常危险。想象一下一个客服Agent未经确认就擅自为用户办理了退款或者一个内容生成Agent自动将未审核的文案发布到了官网——这些都不是天方夜谭而是缺乏“安全护栏”的AI系统必然会导致的灾难。这正是“人机协同”Human-in-the-Loop, HITL中间件存在的核心价值。它不是要拖慢AI的效率而是为AI的“自由发挥”装上可控的“方向盘”和“刹车”。在LangChain 1.x的生态中HumanInTheLoopMiddleware正是实现这一理念的关键组件。它允许开发者在Agent执行链的特定环节插入人工审批节点将敏感操作的最终决策权交还给人类从而在自动化与安全性、合规性之间找到最佳平衡点。简单来说这个项目就是关于如何利用LangChain的中间件机制为你的AI Agent构建一个可配置、可扩展的“安全审批工作流”。无论你是开发金融领域的自动交易顾问、电商领域的智能客服还是企业内部的数据处理自动化工具只要涉及敏感操作HITL都是你必须认真考虑的设计模式。2. HITL中间件核心设计思路拆解2.1 为什么是“中间件”在深入代码之前理解“中间件”在LangChain中的角色至关重要。它不是链Chain的一部分也不是一个单独的工具Tool。中间件更像是一个“拦截器”或“装饰器”它包裹在现有的运行逻辑如Agent执行外部在特定生命周期节点如工具调用前、调用后注入自定义逻辑。选择中间件来实现HITL而非直接修改工具或Agent的逻辑有以下几个核心优势非侵入式你无需重写已有的工具函数或Agent的推理逻辑。只需将中间件“附加”到现有的Runnable如Agent上即可为其增加审批能力。这符合开闭原则对原有代码影响最小。可插拔审批逻辑可以被轻松地启用、禁用或替换。在开发、测试和生产环境中你可以灵活配置是否需要人工介入或者介入的粒度如何。关注点分离审批的流程管理如展示信息、等待输入、记录决策与核心的业务工具逻辑是分离的。这使得代码更清晰也便于团队协作——业务开发者和安全合规开发者可以各司其职。HumanInTheLoopMiddleware正是这样一个标准的LangChainRunnableMiddleware。它的核心职责是在Agent准备调用一个被标记为需要审批的工具时暂停自动化流程将工具的名称、参数以及上下文信息提交给一个“审批处理器”并等待处理器的返回结果再决定是继续、修改还是终止操作。2.2 四种决策模式不只是“通过”或“拒绝”一个粗糙的审批系统可能只提供“批准”和“拒绝”两个按钮。但真实的业务场景要复杂得多。HumanInTheLoopMiddleware的设计考虑到了这种复杂性它定义了四种标准的决策结果几乎涵盖了所有需要人工干预的场景继续CONTINUE审核人完全同意Agent的提议。中间件将放行使用Agent原本准备调用的工具和参数继续执行。这是最直接的“绿灯”。修改后继续MODIFIED_CONTINUE审核人认为Agent的意图正确但参数或细节需要调整。例如Agent要发送一封邮件但收件人或措辞需要修改。审核人可以提供修改后的参数中间件将使用新参数调用原工具。这实现了“指导式”的人机协作。跳过SKIP审核人认为当前步骤不应该执行但整个Agent的任务可以继续。中间件将跳过这个特定的工具调用并返回一个预设的“跳过结果”给AgentAgent可以基于这个结果进行后续推理。例如在信息查询链中某一步查询被判定为冗余可以跳过而不影响最终答案的生成。终止STOP审核人认为出现了严重问题整个Agent任务必须立即停止。中间件将抛出一个特定异常HumanRejectedException彻底中断执行流。这用于处理那些一旦执行就可能造成不可逆后果的操作。这四种决策构成了一个完整的控制矩阵让人类监督者不仅能说“行”或“不行”还能进行“微调”和“路径修正”极大地提升了人机协作的灵活性和有效性。2.3 条件拦截精准控制审批范围为每一个工具调用都请求人工审批是不现实的这会让人工成为瓶颈。因此条件拦截机制是HITL是否实用的关键。你需要在中间件中定义一个should_checkpoint函数或类似逻辑来动态判断当前这次工具调用是否需要触发审批。常见的拦截条件包括基于工具名称只对send_email,transfer_funds,delete_database等高风险工具进行拦截。基于参数内容检查工具调用参数中的敏感信息。例如当转账金额超过某个阈值如amount 10000或邮件内容包含特定关键词时才触发审批。基于运行上下文结合当前会话的上下文信息。例如对于VIP客户某些操作可以自动通过或者在某项任务的初始阶段需要严格审批后续步骤可以放宽。基于用户或环境根据当前登录用户的角色如管理员、普通员工或当前运行环境生产环境、测试环境来决定。在LangChain的实现中这通常通过为工具绑定元数据metadata或在中间件的判断逻辑中访问RunnableConfig来实现。精准的条件拦截确保了人工干预用在“刀刃”上在保障安全的同时最大化自动化流程的效率。3. 核心细节解析与实操要点3.1 审批处理器Approval Handler的实现中间件本身不处理与人的交互它只负责“请求审批”和“接收决策”。与人类交互的具体工作由一个独立的审批处理器Approval Handler来完成。这是你需要重点实现的部分它决定了审批请求如何呈现、如何接收人工输入。审批处理器通常需要实现一个核心方法例如get_decision它接收工具调用信息并返回上述四种决策之一。其实现方式多种多样命令行交互最简单的方式在终端打印信息并等待用户输入。适用于脚本和本地测试。# 示例简单的命令行处理器 def cli_approval_handler(tool_name, tool_args): print(f[审批请求] 工具 {tool_name} 将被调用。) print(f参数: {tool_args}) response input(请决定 (c:继续, m:修改, s:跳过, x:终止): ).strip().lower() # ... 根据response返回对应的决策对象Web API端点在生产环境中最实用的方式。中间件向一个内部API发送审批请求该API会创建一个待办任务如在管理后台生成一条审批记录并等待通过轮询或Webhook前端用户操作后返回结果。这可以与现有的OA、工单系统集成。消息队列与回调适用于异步、长耗时的流程。中间件将审批事件发布到消息队列如Redis, RabbitMQ由独立的审批服务消费并处理人工交互处理完成后通过回调URL通知中间件继续执行。状态检查点Checkpointer集成这是LangChain提供的一个更高级、更集成的模式。Checkpointer用于持久化链的执行状态。当触发审批时中间件可以将当前状态保存到检查点并返回一个“等待”信号。外部系统可以读取这个检查点展示给审批人审批人做出决策后再通过该检查点恢复执行。这种方式特别适合需要暂停很长时间或需要复杂上下文展示的场景。注意审批处理器的实现必须考虑超时和错误处理。如果一个审批请求长时间未得到响应例如审批人下班了系统应该有超时机制并执行默认操作如转为拒绝或通知管理员。同时处理器与中间件的通信需要健壮避免因为网络问题导致整个流程挂起。3.2 工具标记与元数据传递如何告诉中间件哪些工具需要审批有两种主流模式显式列表在初始化中间件时传入一个需要审批的工具名称列表。这种方式简单直接但不够灵活。middleware HumanInTheLoopMiddleware( approval_handlermy_handler, tools_to_intercept[send_email, make_payment] )元数据Metadata驱动这是更优雅和强大的方式。在为工具Tool或tool装饰的函数定义时可以为其添加自定义的元数据。from langchain.tools import tool tool(metadata{requires_approval: True, approval_threshold: 5000}) def transfer_funds(amount: float, account: str): 向指定账户转账。 # ... 工具实现然后在中间件的should_checkpoint函数中可以读取工具的metadata属性来进行动态判断。这种方式可以将审批规则如阈值与工具定义放在一起管理起来更清晰。3.3 状态管理与上下文保持当流程被人工审批中断时一个关键问题是如何保持Agent的“记忆”和“状态”例如一个多轮对话的Agent在审批邮件内容时其之前的对话历史不能丢失。LangChain的Runnable和RunnableConfig机制在这里发挥了作用。RunnableConfig中可以包含callbacks、metadata和tags等信息这些信息会在整个调用链中传递。HumanInTheLoopMiddleware作为中间件可以访问和修改这个config。一种最佳实践是将关键的会话ID、用户标识等信息传入config.metadata。当审批处理器需要向人类展示上下文时它可以利用这个ID去查询完整的会话历史。同样在做出“修改后继续”的决策时修改后的参数也需要能够无缝地集成回当前的执行上下文中确保Agent的后续推理基于更新后的信息。4. 实操过程与核心环节实现下面我将通过一个模拟“智能财务助手”Agent的完整例子展示如何从零搭建一个集成HITL中间件的系统。这个Agent可以帮助用户查询余额和进行转账其中转账操作需要人工审批。4.1 环境准备与工具定义首先安装必要依赖并定义两个工具一个安全的查询工具和一个需要审批的转账工具。# 假设的依赖pip install langchain langchain-openai from langchain.tools import tool from typing import Optional # 工具1查询余额无需审批 tool def check_balance(account_id: str) - str: 查询指定账户的余额。这是一个只读操作很安全。 # 模拟数据库查询 balance_db {acc_001: 15000.0, acc_002: 5000.0} balance balance_db.get(account_id, 0.0) return f账户 {account_id} 的当前余额为 {balance} 元。 # 工具2转账需要审批 tool(metadata{requires_approval: True}) # 使用元数据标记 def transfer_funds(from_account: str, to_account: str, amount: float, note: Optional[str] ) - str: 执行转账操作。这是一个高风险操作需要人工审批。 Args: from_account: 转出账户 to_account: 转入账户 amount: 转账金额 note: 转账备注 # 注意真实的工具实现不会在这里直接转账 # 它应该只在审批通过后由中间件或后续流程调用。 # 这里我们只返回一个模拟的成功消息。 return f成功从 {from_account} 向 {to_account} 转账 {amount} 元。备注{note}4.2 实现一个命令行审批处理器我们实现一个简单的处理器它在命令行与用户交互。from enum import Enum from pydantic import BaseModel class HumanDecision(str, Enum): CONTINUE continue MODIFIED_CONTINUE modified_continue SKIP skip STOP stop class Decision(BaseModel): 审批决策结果模型 decision: HumanDecision modified_args: Optional[dict] None # 仅在 MODIFIED_CONTINUE 时使用 skip_result: Optional[str] None # 仅在 SKIP 时使用 class CliApprovalHandler: 命令行审批处理器 def get_decision(self, tool_name: str, tool_args: dict, runnable_config: dict) - Decision: 获取人工决策 print(f\n{*50}) print(f[HITL 审批请求]) print(f工具: {tool_name}) print(f参数: {tool_args}) print(f上下文元数据: {runnable_config.get(metadata, {})}) print(f{*50}) while True: cmd input(\n请输入决策 (c:继续, m:修改并继续, s:跳过, x:终止): ).strip().lower() if cmd c: return Decision(decisionHumanDecision.CONTINUE) elif cmd m: print(请输入修改后的参数JSON格式例如{\amount\: 100}) try: modified_input input(修改: ) modified_args eval(modified_input) # 生产环境请使用json.loads if not isinstance(modified_args, dict): raise ValueError return Decision(decisionHumanDecision.MODIFIED_CONTINUE, modified_argsmodified_args) except: print(输入格式错误请重新选择。) elif cmd s: skip_reason input(请输入跳过原因将作为结果返回给Agent: ) return Decision(decisionHumanDecision.SKIP, skip_resultf操作被人工跳过。原因{skip_reason}) elif cmd x: return Decision(decisionHumanDecision.STOP) else: print(无效输入请重新输入。)4.3 构建自定义的HITL中间件现在我们基于LangChain的RunnableMiddleware来构建自己的中间件。from langchain.schema.runnable import Runnable, RunnableConfig from langchain.schema.runnable.config import patch_config from typing import Any, Callable import json class HumanRejectedException(Exception): 人工拒绝异常 pass class HumanInTheLoopMiddleware(Runnable): 人机协同中间件 def __init__(self, runnable: Runnable, approval_handler: CliApprovalHandler): self.runnable runnable self.handler approval_handler def _should_intercept(self, tool_name: str, tool_metadata: dict) - bool: 判断是否需要拦截。这里根据工具的元数据判断。 # 这是一个简单的示例检查元数据中是否有 requires_approval 标记 return tool_metadata.get(requires_approval, False) # 更复杂的逻辑可以在这里实现例如检查参数金额等。 def invoke(self, input: dict, config: Optional[RunnableConfig] None, **kwargs) - Any: # 调用原始的runnable例如Agent但我们需要在工具调用时进行拦截。 # 由于标准Runnable可能不直接暴露工具调用点这里展示一个概念性流程。 # 在实际中你可能需要包装一个特定的AgentExecutor或使用更底层的hook。 # 为了演示我们假设 input 包含了用户查询 user_query input.get(input, ) # 模拟Agent的决策过程解析用户意图选择工具 if 转账 in user_query: tool_to_call transfer_funds tool_args {from_account: acc_001, to_account: acc_002, amount: 8000, note: 季度货款} tool_metadata {requires_approval: True} else: # 非敏感操作直接执行 return self.runnable.invoke(input, config, **kwargs) # 检查是否需要拦截 if self._should_intercept(tool_to_call, tool_metadata): print(f检测到需审批的工具调用: {tool_to_call}) # 获取人工决策 decision self.handler.get_decision(tool_to_call, tool_args, config or {}) if decision.decision HumanDecision.CONTINUE: print(审批通过继续执行原操作。) # 调用原始工具这里模拟 return f[模拟执行] {tool_to_call} with args {tool_args} elif decision.decision HumanDecision.MODIFIED_CONTINUE: print(f审批通过使用修改后的参数执行: {decision.modified_args}) # 使用修改后的参数调用工具 modified_args decision.modified_args or tool_args return f[模拟执行] {tool_to_call} with modified args {modified_args} elif decision.decision HumanDecision.SKIP: print(f操作被跳过。返回结果: {decision.skip_result}) # 返回跳过结果Agent应能处理这个结果并继续 return decision.skip_result elif decision.decision HumanDecision.STOP: print(操作被人工终止。) raise HumanRejectedException(该操作已被人工审核员拒绝并终止。) else: # 无需审批直接执行 return self.runnable.invoke(input, config, **kwargs) # 实际开发中你需要将中间件绑定到AgentExecutor上并重写其调用逻辑以捕获工具调用事件。 # LangChain社区或未来版本可能提供更直接的集成方式。4.4 组装并运行智能体将上述组件组装起来创建一个简单的代理流程。from langchain.agents import AgentExecutor, create_react_agent from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain.schema import SystemMessage # 1. 定义工具列表 tools [check_balance, transfer_funds] # 2. 创建LLM和提示词 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) prompt_template PromptTemplate.from_template( 你是一个财务助手。请根据用户问题决定使用哪个工具。\n 可用工具\n{工具描述}\n\n 用户问题{input}\n\n 请只输出工具名称和参数JSON格式。 ) # 注意这是一个极度简化的Agent逻辑。真实场景应使用LangChain的标准Agent框架。 # 3. 创建审批处理器和中间件 approval_handler CliApprovalHandler() # 4. 创建一个简单的可运行对象这里简化实际应为AgentExecutor class SimpleAgent(Runnable): def invoke(self, input: dict, configNone, **kwargs): query input.get(input, ) # 简单的意图识别 if 余额 in query: return check_balance.invoke({account_id: acc_001}) elif 转账 in query: # 这里本应调用transfer_funds但会被中间件拦截 # 为了演示我们返回一个信号表示Agent决定调用转账工具 return {needs_approval: True, tool: transfer_funds, args: {amount: 8000}} else: return 我不确定如何处理这个请求。 base_agent SimpleAgent() # 用中间件包装基础Agent agent_with_hitl HumanInTheLoopMiddleware(runnablebase_agent, approval_handlerapproval_handler) # 5. 运行测试 print(测试1查询余额应直接返回无审批) result1 agent_with_hitl.invoke({input: 我的账户余额是多少}) print(f结果: {result1}\n) print(测试2发起转账应触发审批流程) try: result2 agent_with_hitl.invoke({input: 我要向acc_002转账8000元付货款。}) print(f结果: {result2}) except HumanRejectedException as e: print(f流程被终止: {e})运行上述代码当查询余额时会直接得到结果。当发起转账时程序会暂停在命令行中弹出审批请求等待你输入决策c, m, s, x并根据你的输入执行相应操作。5. 常见问题与排查技巧实录在实际集成HITL中间件的过程中你会遇到一些典型问题。以下是我在多个项目中总结的经验和解决方案。5.1 审批流程“卡住”或无响应问题现象Agent执行到需要审批的工具时程序挂起没有任何日志输出或者审批请求没有按预期出现。排查思路检查条件拦截函数首先确认_should_intercept函数逻辑是否正确。打印工具的名称和元数据确保需要审批的工具确实被识别到。一个常见的错误是元数据的键名不匹配或值为False。确认中间件绑定确保HumanInTheLoopMiddleware正确包装了你的AgentExecutor或Runnable。在LangChain中中间件的绑定顺序有时会影响行为。确保它在执行链的正确位置。检查审批处理器阻塞如果你的审批处理器是同步的如命令行输入并且它在等待一个永远不会到来的输入例如在无头服务器环境中程序就会卡住。在生产环境中必须使用异步的、带超时机制的处理器如基于Web API的处理器。查看执行流在Agent的callbacks中增加日志特别是on_tool_start回调确认工具调用事件是否被触发。这能帮你确定问题是出在工具调用未被捕获还是出在捕获后的审批逻辑里。5.2 决策后状态不一致或上下文丢失问题现象审批人选择“修改后继续”但Agent后续的推理似乎没有基于修改后的参数或者整个会话的状态在审批后重置了。解决方案确保参数正确传递在MODIFIED_CONTINUE决策中你返回的modified_args必须完整替换原始的工具调用参数。中间件需要负责用新参数重新构造工具调用。检查中间件中调用工具的那段代码确认传入的是decision.modified_args。利用RunnableConfig传递状态将重要的会话上下文如session_id,conversation_history放在config.metadata中。在审批处理器里读取这些信息用于展示在决策后确保这些config被原封不动地或更新后传递回后续的执行步骤。不要依赖全局变量。考虑使用Checkpointer对于复杂的、多步骤的链LangChain的Checkpointer是管理状态的最佳实践。当审批触发时将当前链状态序列化保存。审批完成后从检查点恢复。这能完美解决状态丢失问题但架构复杂度较高。5.3 如何设计高效的审批界面与流程痛点审批请求信息不全审批人无法决策审批流程太慢成为系统瓶颈。实操心得信息聚合审批处理器传递给前端的不能仅仅是工具名和参数。应该聚合相关的上下文。例如对于“发送邮件”工具除了邮件内容还应附上生成这封邮件的原始用户请求、之前的对话记录等。这需要你在中间件中能访问到更丰富的运行时上下文RunnableConfig和调用栈信息是关键。分级审批与自动规则不是所有操作都需要人工。实现一个规则引擎低风险操作如金额小于100元自动通过中等风险操作如金额在100-5000元发送通知给负责人设定一个超时如30分钟超时后自动通过或拒绝高风险操作5000元必须人工即时审批。这可以通过在_should_intercept函数和审批处理器中实现复杂逻辑来完成。提供默认选项与批量处理在审批界面提供“批准所有类似请求”或“拒绝并阻止该用户此类操作”的选项。对于高频、低差异的审批这能极大提升效率。与现有系统集成不要另建一套审批系统。将审批事件推送至公司现有的IM工具如钉钉、飞书、Slack审批机器人或生成工单接入OA系统。利用这些系统的通知、待办和移动端能力。5.4 性能与并发考量问题大量并发请求导致审批队列堆积中间件引入的延迟过高。优化技巧异步非阻塞确保整个审批流程是异步的。中间件在发出审批请求后不应同步等待而应挂起当前任务例如返回一个Awaitable或将状态保存到数据库释放工作线程去处理其他请求。这通常需要与异步框架如FastAPI、Celery深度集成。轻量级判断_should_intercept函数的逻辑要尽可能轻量避免复杂的数据库查询或网络调用。它应该是一个快速的、基于内存的判断。设置超时与降级为每个审批请求设置严格的超时时间如2分钟。超时后执行预设的“安全默认操作”例如拒绝操作并记录告警。这能防止整个系统因个别审批卡死。缓存决策对于同一用户、同一类型、参数相似的重复操作可以在短时间内缓存审批结果例如用户A在10分钟内再次请求发送内容相似的邮件自动通过并记录日志。这需要仔细设计缓存键和失效策略避免安全漏洞。最后我想分享一个深刻的体会引入HITL机制技术实现只占一半另一半是流程和人的定义。你必须和业务方、风控团队一起清晰地定义出“什么操作需要审批”、“由谁审批”、“审批的SLA服务等级协议是多久”、“超时了怎么办”。将这些规则明确化、文档化并映射到中间件的配置和审批处理器的逻辑中才能让人机协同真正为业务赋能而不是成为一个摆设或瓶颈。在开发初期不妨将所有的工具调用都日志记录即使不拦截运行一段时间后分析日志再和数据、业务团队一起确定哪些是真正的“敏感操作”这样制定的拦截规则才最有说服力。