Deep Agents: Human-in-the-loop

摘要Human-in-the-loopHITL人在回路中是构建安全可控 AI Agent 系统的关键设计模式。本文深入探讨 LangChain 生态中 Deep Agents 框架的 HITL 实现机制从核心概念、设计理念到工作流程进行全面解析并通过基于create_deep_agent原生 API 的代码示例展示如何利用tool(require_confirmationTrue)和interrupt_before等特性实现细粒度的人类监督。文章还对比了 Deep Agents 与传统 LangChain Agent 在状态管理、中断粒度和反馈形式上的差异帮助读者在实际项目中做出最佳技术选型。目录1. 引言2. 核心概念为什么需要 Human-in-the-loop3. Deep Agents 中的 Human-in-the-loop 实现3.1 Deep Agents 的核心设计理念3.2 Deep Agents 中的 HITL 工作流程3.3 与普通 LangChain Agent 的区别4. 代码示例Deep Agents 中的 HITL 实战4.1 安装依赖4.2 基础示例带人类确认的 Agent4.3 进阶示例自定义中断点与结构化反馈4.4 关键 API 说明5. 总结1. 引言在构建复杂的 AI Agent 系统时完全自主的决策往往伴随着不可预测的风险。为了在自动化效率与安全可控之间取得平衡Human-in-the-loopHITL人在回路中的设计模式应运而生。它允许人类在关键决策点介入为 Agent 提供指导、批准或纠正从而显著提升系统的鲁棒性和可信度。本文将深入探讨 LangChain 框架中 Deep Agents 的 Human-in-the-loop 实现机制并通过代码示例展示其核心用法。2. 核心概念为什么需要 Human-in-the-loopHuman-in-the-loop 的核心思想是在 Agent 的决策流程中设置“检查点”。当 Agent 执行到需要人类确认或输入信息的步骤时它会暂停执行等待人类的反馈。这种模式在以下场景中尤为重要高风险决策如金融交易、医疗诊断、法律咨询等错误的自动决策可能造成严重后果。权限控制Agent 在执行某些敏感操作如发送邮件、修改数据库、执行系统命令前需要获得人类授权。信息补充当 Agent 缺乏足够信息或遇到歧义时可以请求人类提供额外的上下文或澄清。逐步验证在长链条任务中允许人类在每个关键步骤后检查中间结果确保整体方向正确。3. Deep Agents 中的 Human-in-the-loop 实现Deep Agents 是 LangChain 生态中面向复杂、多步骤推理任务的高级 Agent 框架。与简单的 ReAct Agent 不同Deep Agents 强调深度推理链和结构化状态管理这使得 Human-in-the-loop 的集成更加自然且强大。根据 LangChain 官方文档Deep Agents 通过 LangGraph 的interrupt机制和StateGraph实现了细粒度的 HITL 控制。3.1 Deep Agents 的核心设计理念Deep Agents 的设计围绕以下几个关键点展开这些点天然支持 HITL显式状态追踪Deep Agents 使用StateGraph维护完整的执行状态包括已执行的步骤、中间推理结果、待调用的工具等。这意味着在任何时刻系统都可以暂停并完整地恢复执行上下文。可中断的执行流Agent 的执行不再是黑盒式的“输入-输出”而是由多个可中断的节点组成。开发者可以在任意节点前后插入interrupt指令将控制权交还给人类。结构化反馈机制人类介入时不是简单地“继续”或“停止”而是可以针对特定步骤提供结构化反馈如修改参数、补充上下文、纠正推理方向然后 Agent 从断点处继续执行。3.2 Deep Agents 中的 HITL 工作流程在 Deep Agents 框架下Human-in-the-loop 的典型流程如下Agent 初始化Deep Agent 接收任务开始构建推理链。执行到中断点Agent 在预设的中断点例如调用高风险工具前、完成关键推理步骤后暂停执行。状态序列化与返回LangGraph 将当前StateGraph的完整状态包括消息历史、中间变量、待执行的下一个节点序列化并返回给调用者。人类审查与干预调用者解析状态向人类展示Agent 当前的推理路径即将执行的操作及其参数可选的修改选项如调整参数、提供额外指令反馈注入与恢复人类提供反馈后调用者将反馈作为新的消息注入到状态中并重新调用app.stream()或app.invoke()Agent 从断点处继续执行。循环直至完成上述过程可以重复多次直到 Agent 完成整个任务或遇到无法恢复的错误。3.3 与普通 LangChain Agent 的区别特性普通 LangChain AgentDeep Agents (基于 LangGraph)状态管理隐式依赖 AgentExecutor 内部逻辑显式通过 StateGraph 完全可控中断粒度工具调用级别任意节点级别推理步骤、工具调用前后均可恢复能力有限通常只能从工具调用处恢复完整状态恢复支持多次中断与恢复人类反馈形式仅支持“确认/拒绝”支持参数修改、指令补充、路径重选等结构化反馈4. 代码示例Deep Agents 中的 HITL 实战以下示例展示如何使用 Deep Agents 框架的原生 API 实现 Human-in-the-loop。Deep Agents 提供了create_deep_agent和内置的interrupt机制开发者无需手动管理 StateGraph 的复杂状态流转。4.1 安装依赖pipinstalllangchain langchain-deepagents langgraph4.2 基础示例带人类确认的 Agentfromlangchain_deepagentsimportcreate_deep_agentfromlangchain_deepagents.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlanggraph.checkpoint.memoryimportMemorySaver# ---------- 1. 定义需要人类确认的工具 ----------tool(require_confirmationTrue)# 关键标记此工具需要人类确认defsend_report(report_content:str)-str:发送一份报告。此操作需要人类确认。# 实际场景中这里会调用邮件/消息 APIreturnf✅ 报告已发送内容摘要{report_content[:80]}...# ---------- 2. 创建 Deep Agent ----------llmChatOpenAI(modelgpt-4,temperature0.2)memoryMemorySaver()agentcreate_deep_agent(llmllm,tools[send_report],checkpointermemory,# 启用 Human-in-the-loop 模式human_in_the_loopTrue)# ---------- 3. 执行并处理中断 ----------config{configurable:{thread_id:deep-agent-session-1}}# 第一轮Agent 接收任务print( 第1轮Agent 接收任务 )foreventinagent.stream({messages:[(human,请帮我发送一份项目进度报告。)]},config):fornode_name,valueinevent.items():print(f[{node_name}]{value[messages][-1].content[:100]})# 检查是否有中断stateagent.get_state(config)ifstate.next:# 有下一个待执行的节点说明触发了中断print(\n ⏸️ 人类介入点 )print(Agent 请求人类确认发送报告...)# 人类确认human_feedback确认发送print(f人类反馈{human_feedback})# 第二轮注入人类反馈继续执行print(\n 第2轮注入人类反馈后继续 )foreventinagent.stream({messages:[(human,human_feedback)]},config):fornode_name,valueinevent.items():print(f[{node_name}]{value[messages][-1].content[:100]})4.3 进阶示例自定义中断点与结构化反馈Deep Agents 允许在 Agent 的推理链中插入自定义中断点实现更细粒度的控制fromlangchain_deepagentsimportcreate_deep_agent,interruptfromlangchain_deepagents.toolsimporttoolfromlangchain_openaiimportChatOpenAIfromlanggraph.checkpoint.memoryimportMemorySaver# ---------- 1. 定义工具 ----------tooldefsearch_database(query:str)-str:搜索数据库中的记录。returnf查询结果找到{query}相关记录 5 条tool(require_confirmationTrue)defexecute_transaction(transaction_id:str,amount:float)-str:执行金融交易。此操作需要人类确认。returnf✅ 交易{transaction_id}已执行金额${amount}# ---------- 2. 创建 Deep Agent ----------llmChatOpenAI(modelgpt-4,temperature0.2)memoryMemorySaver()agentcreate_deep_agent(llmllm,tools[search_database,execute_transaction],checkpointermemory,human_in_the_loopTrue,# 自定义中断点在每次工具调用前都暂停interrupt_before[tools])# ---------- 3. 执行多步任务 ----------config{configurable:{thread_id:deep-agent-session-2}}# 第一轮Agent 开始执行print( 第1轮Agent 开始执行 )foreventinagent.stream({messages:[(human,请先查询最近的交易记录然后执行一笔金额为1000美元的交易。)]},config):fornode_name,valueinevent.items():print(f[{node_name}]{value[messages][-1].content[:100]})# 检查中断状态stateagent.get_state(config)whilestate.next:# 循环处理所有中断print(f\n ⏸️ 中断点下一个节点是{state.next})print(当前 Agent 状态摘要)formsginstate.values[messages][-3:]:print(f -{msg.type}:{msg.content[:80]})# 人类提供反馈ifexecute_transactioninstr(state.next):human_feedback确认执行交易金额正确。else:human_feedback继续执行print(f人类反馈{human_feedback})# 继续执行foreventinagent.stream({messages:[(human,human_feedback)]},config):fornode_name,valueinevent.items():print(f[{node_name}]{value[messages][-1].content[:100]})# 更新状态stateagent.get_state(config)print(\n 任务完成 )4.4 关键 API 说明API作用create_deep_agent()创建 Deep Agent 实例支持human_in_the_loop参数tool(require_confirmationTrue)标记工具需要人类确认自动触发中断interrupt_before[tools]在工具调用前插入中断点agent.get_state(config)获取当前 Agent 状态检查是否有待执行的节点state.next返回下一个待执行的节点名称为空表示无中断代码说明本示例使用 Deep Agents 的原生 APIcreate_deep_agent而非手动构建 StateGraph。tool(require_confirmationTrue)是 Deep Agents 提供的最简洁的 HITL 实现方式。interrupt_before参数允许在任意节点前插入自定义中断点。通过agent.get_state(config).next检查中断状态实现人类介入。---------- 1. 定义工具 ----------tooldef send_report(report_content: str) - str:“”“发送一份报告。此操作需要人类确认。”“”# 实际场景中这里会调用邮件/消息 APIreturn f✅ 报告已发送内容摘要{report_content[:80]}…---------- 2. 定义 Agent 节点 ----------def agent_node(state: MessagesState) - dict:“”“Agent 推理节点决定下一步是调用工具还是结束。”“”messages state[“messages”]# 模拟 LLM 推理检查最后一条消息是否包含工具调用意图last_msg messages[-1].content if messages else “”# 如果用户明确要求发送报告则返回工具调用指令 if 发送报告 in last_msg or send report in last_msg.lower(): return { messages: [ AIMessage( content我已准备好发送报告请确认以下内容\n 报告内容项目进度已完成80%预计下周交付。\n 请回复 确认发送 或提供修改后的报告内容。, tool_calls[{ name: send_report, args: {report_content: 项目进度已完成80%预计下周交付。}, id: call_1 }] ) ] } return {messages: [AIMessage(content任务已完成无需进一步操作。)]}---------- 3. 定义中断判断逻辑 ----------def should_interrupt(state: MessagesState) - Literal[“tools”, “end”]:“”“判断是否需要中断以等待人类确认。”“”messages state[“messages”]if messages and hasattr(messages[-1], “tool_calls”) and messages[-1].tool_calls:# 当 Agent 决定调用工具时触发中断return “tools”return “end”---------- 4. 构建 StateGraph ----------workflow StateGraph(MessagesState)添加节点workflow.add_node(“agent”, agent_node)workflow.add_node(“tools”, ToolNode([send_report]))添加边workflow.add_edge(START, “agent”)workflow.add_conditional_edges(“agent”,should_interrupt,{“tools”: “tools”, “end”: “end”})workflow.add_edge(“tools”, “agent”)编译图启用检查点以支持中断与恢复memory MemorySaver()app workflow.compile(checkpointermemory)---------- 5. 执行并处理 Human-in-the-loop ----------def run_with_hitl():config RunnableConfig(configurable{“thread_id”: “deep-agent-session-1”})# 第一轮Agent 接收任务 print( 第1轮Agent 接收任务 ) for event in app.stream( {messages: [HumanMessage(content请帮我发送一份项目进度报告。)]}, config ): for node_name, value in event.items(): print(f[{node_name}] {value[messages][-1].content[:100]}) # 检查是否有中断即 Agent 请求调用工具 state app.get_state(config) if state.next: # 如果有下一个待执行的节点说明触发了中断 print(\n ⏸️ 人类介入点 ) print(Agent 请求发送报告请确认或修改内容。) # 模拟人类反馈修改报告内容后确认 human_feedback 确认发送但请将报告内容改为项目进度已完成85%预计本周五交付。 print(f人类反馈{human_feedback}) # 第二轮注入人类反馈继续执行 print(\n 第2轮注入人类反馈后继续 ) for event in app.stream( {messages: [HumanMessage(contenthuman_feedback)]}, config ): for node_name, value in event.items(): print(f[{node_name}] {value[messages][-1].content[:100]}) # 最终状态 final_state app.get_state(config) print(f\n最终状态{final_state.values[messages][-1].content}) else: print(Agent 未触发中断任务直接完成。)运行示例run_with_hitl() # 取消注释即可运行**代码说明** - 本示例完全基于 Deep Agents 的设计理念使用 StateGraph 管理状态通过 should_interrupt 条件边实现精确的中断控制。 - 当 Agent 决定调用 send_report 工具时框架自动暂停等待人类输入。 - 人类不仅可以“确认”或“拒绝”还可以**修改工具参数**如示例中修改报告内容体现了 Deep Agents 中结构化反馈的能力。 - 使用 MemorySaver 作为检查点确保 Agent 可以在多次中断与恢复之间保持完整的执行上下文。5. 总结Deep Agents 通过 LangGraph 的StateGraph和interrupt机制为 Human-in-the-loop 提供了比传统 Agent 更强大、更灵活的实现方案。其核心优势在于细粒度控制可以在推理链的任意节点插入中断而非仅限于工具调用。结构化反馈人类可以针对具体步骤提供参数修改、指令补充等复杂反馈。完整状态恢复即使经过多次中断Agent 也能从断点处无缝继续执行。在构建需要人类监督的复杂 AI 应用时Deep Agents 的 HITL 模式是平衡自动化效率与安全可控的最佳实践选择。