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

深入了解企业级Agent系统的关键挑战和解决方案。

核心内容：
1. HITL在企业级场景中的必要性与常见模式
2. LangGraph框架下HITL的核心机制与应用
3. HITL工具调用模式与远程协作策略

在LLM Agent的自动化流程中，Human-in-the-Loop（HITL，人类参与闭环）已经成为了一个绕不开的设计模式。尤其是面对那些要求苛刻的企业级场景，HITL能让人类在流程中对Agent的运行进行适时监督与接入，从而大大提升系统的准确性、可信度和用户体验。不过话说回来，从技术实现的角度看，HITL也常常让人头疼：流程的中断、恢复、人类反馈、持久化、前后台协作……这些问题一个比一个棘手。本文会以LangGraph为框架，针对企业级环境下的HITL关键挑战做一次深入解读。整个文章将涵盖以下几个部分：

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

本篇作为上篇，会先探讨前四个部分的内容。所有代码会在整个系列完结后一并提供。

01

HITL的必要性与常见模式

尽管我们大多数时候都乐于看到高度自动化的AI带来的效率提升，但现实往往没那么理想：

• LLM的能力远非绝对可靠，甚至会犯下严重的错误（比如幻觉、推理错误）、表现出不确定性（答案信心不足）、或者不合规。
• 很多时候，为了可靠性，我们必须牺牲一部分效率。尤其是在企业级应用场景里，一次关键错误可能带来灾难性后果。

HITL在企业中的应用其实随处可见。例如，智能客服面对复杂问题时，人工介入修正答案；企业OA系统里，发送重要邮件或公告前需要人工审批；又或者，Agent在调用破坏性工具之前，需要经过人工审核。随着Agent在B端逐步落地，HITL会成为很多部署方案的标配。

用一张图可以很直观地展示HITL的几种典型模式：

这些模式在人工智能的参与方式和时机上各有不同，大致可以分成三类：

• 审批确认
  ：在Agent执行某些关键动作之后，或达到某个指标时，需要等待人类审核，并给予批准、拒绝或下一步行动指示。

• 信息注入
  ：流程跑到了某个环节，可能某些条件需要人类补充更多信息。比如任务调整、信息补充、对中间结果的反馈校正等。

• 安全管控
  ：本质上和审批类似，主要是针对Agent的工具使用进行管控，尤其是那些破坏性较大的工具，需要审核其参数，必要时直接终止操作。

很多时候这些模式并不会孤立存在，一个复杂的企业级流程往往是多种HITL模式的混合体。这自然就会带来流程上的复杂性，也是很多技术难题的根源。

02

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

那么，实现带有人类参与的Agent系统的关键到底在哪里？通俗来说，你首先需要解决“流程中断与恢复”的问题，而这背后依赖的是一套“状态持久化”机制。简单点说，你需要一种能力，可以把流程“挂起”在某个特定节点（或步骤），等待人类参与和反馈，然后从中断点完整地

：这就要求系统能够记录中断时的所有上下文，并且确保恢复后的状态完全一致。显然，用sleep等待或轮询这种阻塞式方案是行不通的。

LangGraph给出的解决方案，是Interrupt（中断）、Command Resume（命令恢复）、Checkpoint（检查点）这三大机制。

：它的作用就是暂停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的作用会暂时挂起整个工作流，并将review_data返回给人类处理。一旦人类给出反馈，并希望流程继续，这个节点就会收到反馈信息（这里的decision），从而恢复运行。

：要恢复工作流，你需要获得人类的反馈，并将这个反馈注入工作流的状态（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，变成之前中断调用（interrupt）的返回值（也就是上面代码中的decision）。

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

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

这里创建了一个基于Postgres的checkpointer，并将其交给Agent。首次通过setup创建数据库对象后，每个节点运行结束时，状态都会被序列化并持久保存。

03

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

虽然LangGraph处理HITL的核心机制看起来很简单，但为了能更好地掌控和运用它，有几个关键点需要特别注意：

：Interrupt到底是个什么东西？为什么它能中断流程？原因很简单——它本质上就是一个抛出的异常（Exception），异常信息就是中断时送出的数据。所以，在调用interrupt时，千万不要自己去捕获自定义异常，否则很可能无法正常中断。

：需要清楚一点，“断点续跑”只是从中断的节点重新开始执行，并不是从interrupt函数调用的那一行代码处恢复！因此，在这个节点中，interrupt调用之前，不要做任何修改状态（State）的动作。最好让这个人工节点只负责处理中断逻辑。

：断点续跑依赖于首次Agent调用时的thread_id。所以，凡是需要处理HITL的场景，都必须提供这个信息。Checkpointer需要它来创建检查点，而恢复运行时也需要提供相同的thread_id，才能找到对应的检查点。

理解了这些注意点后，我们来梳理一下整个处理过程。以一个本地SDK模式下直接调用Agent的客户端为例：

1. 客户端调用invoke启动Agent工作流，并指定thread_id和输入信息。
2. 工作流运行到人工节点的interrupt调用，发生中断，并携带了中断数据。
3. 客户端收到Agent返回的状态，发现有中断，于是提示用户。
4. 用户输入反馈后，客户端再次调用invoke恢复工作流，并指定thread_id和resume信息。
5. 再次进入人工节点时，由于已经有了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 ta vily_search(query: str, search_depth: Optional[str] = "basic"):
    ...
    # 中断执行，等待人工审核
    response = interrupt({
        "tool": "ta vily_search",
        "args": {
            "query": query,
            "search_depth": search_depth
        },
        "message": f"准备使用Ta vily搜索：n- 查询内容: {query}n- 搜索深度: {search_depth}nn是否允许继续？n输入 'yes' 接受，'no' 拒绝，或 'edit' 修改查询关键词",
    }) 

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

这里的search工具在真正开始执行逻辑之前，会先请求人工审核：人工可以选择同意、拒绝，或者修改搜索参数后继续执行。这种模式下，由于人类审批的请求由工具自我管理，如果每个工具都要增加这种中断代码，维护起来就会很麻烦。一个更好的做法是：由于不同工具的审核与反馈过程相对一致，可以设计一个Python装饰器，给普通的工具函数“偷偷”加上人工审核功能（LangGraph官方其实提供了一个类似的Wrapper函数，但装饰器用起来更简洁）。

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

@human_in_the_loop()
def ta vily_search(query: str, search_depth: str = "basic"):
    """使用Ta vily进行网络搜索"""
    try:
......

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

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

• 集中看守模式
  更适合安全管控与审计要求较高、工具数量可控、需要快速接入第三方工具（比如MCP）的企业与场景。

• 自我管理模式
  则更适合分布式并行开发环境下，希望工具能够自治、低风险的常规调用、以及以自研工具为主的企业与场景。

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