我要投稿

彻底说清 Human-in-the-Loop：企业级 Agent 系统的关键挑战与LangGraph解法【上】

发布日期：2025-05-28 09:30:05 浏览次数： 2521

作者：AI大模型应用实践

微信搜一搜，关注“AI大模型应用实践”

在LLM Agent的自动化流程中，Human-in-the-Loop（HITL，人类参与闭环）是常见的设计模式之一。特别在要求较高的企业级场景中，HITL可以让人类在流程中对Agent运行进行适时的监督与接入，从而提高系统的准确性、可信度和用户体验。

不过在技术角度，HITL却常常成为很多人的梦魇：流程的中断、恢复、人类反馈、持久、前后台协作等，常常是HITL面临的众多难点。本文以LangGraph为框架，对企业级环境下的HITL关键挑战做解读与实践。

全文内容将涵盖：

HITL的必要性与常见模式
HITL基础应用：核心机制（LangGraph）
HITL基础应用：原理解析与注意点
HITL下的工具调用：两种模式详解
远程模式下的HITL：前后台如何协作
远程模式下的HITL：客户端的故障恢复
远程模式下的HITL：服务端的故障恢复

本篇首先介绍前四个部分。全部代码将在完结后一起提供。

01

HITL的必要性与常见模式

尽管在大部分时候我们都乐见高度自动化的AI带来的效率提高。但现实是：

LLM远非全能。甚至有相当大的可能会犯错（幻觉、推理错误），会不确定（答案信心不足），或不合规。
很多时候需要牺牲效率来换取可靠性。特别是在企业级应用场景下，一次关键错误可能会带来灾难性的影响。

HITL在企业的例子随处可见。比如：在智能客服的复杂问题回答中人工介入修正答案；企业OA中发送重要邮件与公告前需人工审批；Agent调用破坏性工具(Tool)前需要人工审批。随着Agent在B端的逐渐落地，HITL将是很多部署的标准配置。

用一张图描述HITL的几种典型模式：

这些不同的模式下，人工参与的方式与时机各有差异。可以做简单分类：

审批确认：在某些Agent的关键动作后，或达到某个指标时，需要等待人类审核，并给予批准、拒绝、或者下一步行动指示。
信息注入：在流程运行的某些环节，某些条件可能会需要人类注入更多信息。包括：任务调整、信息补充、中间结果的反馈校正等。
安全管控：本质上与审批类似。主要是针对Agent的工具使用进行管控，特别是对破坏性较大的工具进行审核、参数检查，必要时终止。

很多时候这些模式都非孤立存在，一个复杂的企业级流程可能混合着多种HITL模式。这很容易带来一些流程上的复杂性，也是一些技术难题的根源。

02

HITL基础应用：核心机制（LangGraph）

实现带有人类参与的Agent系统的关键在哪里？或许你可以想象到：

流程中断与恢复，以及为了支持它所需要的状态持久化机制。

简单说，就是需要一种机制，将流程“挂起”在特定节点（或步骤），等待人类参与和反馈，然后能从中断点恢复运行：

这要求系统能够记录中断时的上下文，并确保恢复后状态一致。很显然，你不能使用sleep等待或轮询这种糟糕的阻塞式方案。而LangGraph给出的解决方案是Interrupt（中断）、Command Resume（命令恢复）、Checkpoint（检查点）三大机制。

Interrupt（中断）

即暂停LangGraph工作流的执行，同时返回一个中断数据对象。其中含有给人类的信息，比如需要审核的内容，或者恢复时需要的元数据。典型的处理如下：

from langgraph.types import interrupt, Command
...
#这是一个Agent的某个人工参与的节点
def human_review_node(state: State):

   # 暂停执行，输出需人工审核的数据
    review_data = {"question": "请审核以下内容:", "output": state["llm_output"]}
    decision = interrupt(review_data) 

    # 恢复后将根据人工决策更新状态或跳转
    if decision == "approve":
        return Command(goto="approved_node")
    else:
        return Command(goto="rejected_node")
...

在这个Agent节点中，interrupt的作用会暂时挂起这个Agent工作流，并将review_data返回给人类处理。一旦人类给予反馈，并要求工作流继续进行，该节点就会收到反馈信息（这里的decision），进而可以恢复运行。

Command Resume（恢复）

要恢复工作流，需要获得人类反馈并注入工作流状态（State），然后发出继续执行的命令：使用 Command(resume=value) 来反馈并恢复。这个工作通常是调用Agent的客户端来完成，比如：

...调用agent客户端程序...
    result = graph.invoke(initial_state, config) #调用Agent启动工作流
    interrupt_info = result['__interrupt__'][0].value

    ...显示中断信息，人类交互与反馈...

    # 假设 thread_id 标识此次任务，再次调用invoke恢复运行即可
    user_decision = "approve"  # 这里模拟用户最后的反馈
    result = graph.invoke(Command(resume=user_decision), config={"configurable": {"thread_id": thread_id}})
...

在这里通过Command对象的resume信息将人类反馈再送回Agent，变成之前Agent发起的interrupt调用的返回值（即上面代码中的decision）。

Checkpoint（检查点）

为了实现“断点续跑”，必须要实现Agent的状态持久化，用来在恢复时“重建现场”。这种机制也有利于Agent发生故障时的轨迹重放。这需要你首先创建一个检查点管理器：

...
# 初始化 PostgreSQL 检查点保存器
with PostgresSaver.from_conn_string("postgresql://postgres:yourpassword@localhost/postgres?sslmode=disable") as checkpointer:
        checkpointer.setup()
        graph = builder.compile(checkpointer=checkpointer)

这里创建了一个基于Postgres的checkpointer，并将其交给Agent。该checkpointer首次通过setup创建数据库对象；后续在每个node运行后将State序列化后并持久保存。

03

HITL基础应用：原理解析与注意点

尽管LangGraph处理HITL的核心机制貌似很简单。不过为了更好的掌控和应用它，有几个注意点需要特别了解：

Interrupt的本质是Exception（异常）

Interrupt的本质是什么？为什么它可以中断？原因很简单，因为它就是丢出了一个异常（Exception），异常信息就是中断时送出的数据。所以在发起Interrupt调用时不要做自定义异常捕获，否则可能无法中断。

断点续跑是从中断所在的“节点”开始

需要注意，“断点续跑”只是从中断的node重新开始，并不是从Interrupt函数调用处开始！所以不要在这个节点的Interrupt之前做改变状态（State）的动作！如果可能，尽量让人工节点只负责处理中断。

要有唯一的ID标识一次工作流运行过程

“断点续跑”依赖于首次Agent调用时的thread_id，所以如果需要处理HITL，就需要提供该信息。因为Checkpointer需要借助它做检查点，而恢复运行时则需要提供相同的thread_id来让Checkpointer找到对应的检查点。

在理解了这几个注意点后，最后来总结与回顾整个处理过程：

以一个本地SDK模式下直接调用Agent的客户端为例：

客户端调用invoke启动Agent工作流，指定thread_id和输入信息
工组流运行到人工节点的interrupt调用，发生中断，并携带了中断数据
中断发生。客户端收到Agent的返回状态，从中发现有中断，则提示用户
用户输入反馈后，调用invoke恢复工作流，指定thread_id和resume信息
再次进入人工节点，此时由于有resume信息，interrupt函数不会触发中断，直接返回resume信息；流程得以继续运行。至此，一次中断过程处理结束

在实际生产中，一次的复杂流程运行可能会发生多次中断与人类参与，你需要更谨慎的设计Agent工作流以及客户端对多次中断的处理（借助于循环或者递归的方法。

我们用一个例子演示上面的基础应用：一个借助AI润色文本的Agent，在每次润色以后会请求人工交互以获得修改意见；同时在最后输出前会再次确认成果。

交互过程如下：

04

HITL下的工具调用：两种管控模式

工具（Tools）使用是Agent最普遍的模式。当Agent准备调用外部工具或执行关键操作时，引入人工确认可以避免错误或高风险行为（特别是在MCP后大量共享工具的出现）。尽管在大的方法上和普通的审批没有质的不同，但在细节上有一些更灵活的控制要求。一个最常见的问题是

在哪里拦截工具调用的意图？如何更方便的管控工具是否需要审核？

工具拦截，也就是工具的人工审批环节，应该设置在何处？我们以典型ReAct Agent为例，详细阐述两种人类管控模式。

【集中看守模式】

这种模式下，所有的工具调用会经过一个审批节点：

这种模式中，前置的规划节点（一般是LLM Call）输出工具调用的需求（Tool_Calls）。人工审批节点进行判断，如果发现工具调用具有高风险，则通过 interrupt发起中断，提交给人类审核，否则直接通过。大致逻辑如下：

def human_approval_node(state: State):

        .....从历史消息或者状态获得工具调用消息:tool_calls.....
        tool_calls_info = []
        
        # 获取工具调用信息（这里暂时只取第一个演示）
        tc = tool_calls[0]
        tool_id = tc.get("id", "未知工具ID")
        tool_name = tc.get("name", "未知工具")
        tool_args = tc.get("args", {})
        tool_calls_info.append(f"{tool_name}({tool_args})")
            
        # 非高风险工具自动批准
        if tool_name not in HIGH_RISK_TOOLS:
            return {"human_approved": True}
        
        tool_calls_str = "\n - ".join(tool_calls_info)
        
        # 高风险工具：中断并等待人工审批
        value = interrupt({
            "tool_calls": tool_calls_str,
            "message": "请输入 'ok' 批准工具使用，或输入 'reject' 拒绝"
        })
...

很显然，通过设置这里的HIGH_RISK_TOOLS列表（高风险工具），就可以灵活调整哪些工具需要审核，哪些工具则可以自由放行。

这种模式有一个细节问题：当工具被拒绝时，你不能简单的将请求路由回原节点。由于一些LLM要求在出现Tool_calls的AI消息后，必须有对应的工具结果（LangGraph中ToolMessage类型的消息），否则会导致API错误。因此，这里你可以人为的修改State，添加一条表明工具被拒绝的ToolMessage。

【自我管理模式】

这种模式下，工具的内部逻辑自行决定是否需要人类审批：

比如一个数据库访问的工具，在执行SQL之前，自行判断并针对所有非只读的请求发起中断，要求人工审批。以下是一个需要审批的搜索工具内部：

async def tavily_search(query: str, search_depth: Optional[str] = "basic"):
    ...
    # 中断执行，等待人工审核
    response = interrupt({
        "tool": "tavily_search",
        "args": {
            "query": query,
            "search_depth": search_depth
        },
        "message": f"准备使用Tavily搜索：\n- 查询内容: {query}\n- 搜索深度: {search_depth}\n\n是否允许继续？\n输入 'yes' 接受，'no' 拒绝，或 'edit' 修改查询关键词",
    }) 

    # 处理人工响应
    if response["type"] == "accept":
        pass
    elif response["type"] == "edit":
        query = response["args"]["query"]
    else:
        returnf"该工具被拒绝使用，请尝试其他方法或拒绝回答问题。"
    
    ...开始执行真正的工具逻辑...

这里的search工具在开始真正的工具逻辑之前，首先请求人工审核：人工可以选择同意、拒绝，或者修改搜索参数后继续执行。

这种模式下，由于人类审批的请求由工具自我管理。因此，如果需要针对每个工具都增加这种中断代码，就会变得难以维护。一个较好的方法是：

由于对不同工具的审核与反馈过程相对一致。因此可以设计一个Python的装饰器，来给普通的工具函数“偷偷的”加上人工审核功能（LangGraph官方提供了一个类似的Wrapper函数实现，不过装饰器似乎更简洁）：

现在，你在创建工具时无需考虑HITL。只需要在必要的时候加上装饰器，该工具就具备了人工审核的功能（@human_in_the_loop）：

@human_in_the_loop()
def tavily_search(query: str, search_depth: str = "basic"):
    """使用Tavily进行网络搜索"""
    try:
......

我们创建了一个ReAct Agent来验证这个工具的调用（客户端处理仍需自行实现）：

最后，简单总结两种管控模式的差异：

集中看守模式更适合安全管控与审计要求较高；工具数量可控；需要快速接入第三方工具（比如MCP）的企业与场景
自我管理模式更适合分布式并行开发下希望工具能自治；低风险的常规调用；以自研工具为主的企业与场景

总之，通过引入Human-in-the-Loop，无论哪种模式，都可以对Agent运行过程中的工具风险进行充分管控，这对于在企业中推行Agent应用有重要的意义。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业