浅谈 HITL 与 Checkpointer

发表于最近更新于最近

作者 Neo

18~23 分钟 阅读

前言

本地跑通了 Multi-Agent 之后，我一直在想，企业级 AI 应用的两个关键技术还没真正用上：一个是 HITL（Human-in-the-Loop，人工介入），一个是 Checkpointer（状态持久化与回滚）。正好最近在研究跨境物流报关场景——这个领域因为涉及海关监管，人工审核是硬性要求，不是可选项。于是我用 LangGraph 搭了一套带 HITL 和状态回滚的智能物流报关 Agent，下面把踩坑和心得做一次总结。

1. 为什么报关场景必须上 HITL

报关场景和普通问答不一样，它有一条绝对红线：涉及商品申报、电池、危险品时，AI 不能直接给结论，必须调用风控工具，触发高风险时必须转人工审核。

这意味着 Agent 不能是一个“一跑到底”的自动化流水线，而必须是一个可暂停、可干预、可恢复的状态机。

LangGraph 天然支持这种模式，核心就是两个能力：

interrupt()：在任意节点挂起执行，等待外部输入
checkpointer：把当前状态持久化，外部输入到达后从断点恢复

2. 状态图设计：三个节点一条闭环

我设计的 Agent 状态图结构如下：

agent 节点：LLM 主推理，负责生成 tool_call 或最终回答
tools 节点：执行风控工具，返回判定结果
human_approval_node：人工审批挂起与唤醒

流转逻辑是：

text

START → agent → tools → (判断是否需要人工) → human_approval → agent → END

其中关键的一条条件边在 tools 之后：

python

builder.add_conditional_edges(
    "tools",
    self._check_risk_interrupt,
    {
        "human_approval": "human_approval_node",
        "agent": "agent"
    }
)

如果工具返回的 status == "INTERRUPT_REQUIRED"，就转到人工审批节点；否则继续回到 agent 循环。

3. 挂起与唤醒的工程细节

human_approval_node 里最核心的是 interrupt() 调用：

python

approval_payload = interrupt({
    "type": "RISK_APPROVAL_REQUIRED",
    "message": "商品触发高风险风控规则，必须进行人工审核！",
    "risk_details": risk_info,
    "action_required": ["APPROVE", "REJECT"],
})

这里有两个踩坑点：

3.1 上下文注入问题

直接在节点函数里调用 interrupt() 有时会报“找不到当前 checkpoint 上下文”。原因是 interrupt() 依赖 LangGraph 内部的 Runnable 配置上下文，而节点函数是异步的，上下文可能丢失。

解法是手动注入：

python

token = var_child_runnable_config.set(config) if config else None
try:
    approval_payload = interrupt({...})
finally:
    if token:
        var_child_runnable_config.reset(token)

3.2 唤醒后的人工反馈如何传回 LLM

审批完成后，系统会拿到 APPROVE 或 REJECT 的结果。我的做法是把审批结果包装成一条 HumanMessage 塞回消息流：

python

system_feedback = (
    f"【HITL 人工审核结果】:\n审核操作: {status}\n审核意见: {reviewer_note}"
)
return {
    "messages": [HumanMessage(content=system_feedback)],
    "need_human_approval": False,
}

这样 agent 节点再次执行时，就能带着人工审核意见继续推理。

4. Checkpointer 的持久化与回滚

LangGraph 的 checkpoint 机制可以把任意节点的中间状态存入 Redis。我的用法是：

python

self.checkpointer = AsyncRedisSaver.from_conn_string(self.redis_url)
self.app = builder.compile(checkpointer=self.checkpointer)

当 interrupt() 被触发时，当前状态自动写入 Redis；外部调用 invoke 或 astream 时带上相同的 thread_id，LangGraph 就会从 Redis 恢复断点继续执行。

4.1 状态回滚怎么做

“回滚”并不是 LangGraph 的原始概念，但可以这样理解：恢复到挂起时的消息列表，把最后一条工具结果删掉，重新跑一遍工具节点和审批节点。

实现上，我是在回滚接口里用 aget_state 拿到 checkpoint 里的消息历史，截取到挂起前的状态，然后重新用 ainvoke 触发恢复。这个过程中，消息列表的完整性至关重要。

5. 踩坑最深的：reasoning_content 丢失

回滚后再次审批，报错：

text

Error code: 400 - The `reasoning_content` in the thinking mode must be passed back to the API.

这个问题的根源在于：DeepSeek 在 thinking 模式下，assistant 消息里的 reasoning_content 字段必须原样传回。但 LangGraph 的 checkpoint 序列化并不会保留这个字段——它只存 content、tool_calls 等标准字段。

5.1 第一轮尝试：重写 LangChain 序列化方法

我首先尝试继承 ChatOpenAI 重写 _convert_message_to_dict，把 reasoning_content 注入请求体。结果发现新版 langchain-openai 根本没有这个方法了，重写了个寂寞。

5.2 第二轮尝试：从 Redis 恢复

我写了一个 Redis 缓存层，在 LLM 返回时把 reasoning_content 单独存一份，回滚后再根据消息 ID 捞回来。但问题是，如果消息 ID 对不上或者缓存过期，还是会丢。

5.3 最终方案：绕过 LangChain，手动构造请求

我彻底放弃依赖 LangChain 的消息序列化，改为：

手动把 LangChain 消息列表转成 DeepSeek API 需要的 dict 格式
对每条 AIMessage 强制填充 reasoning_content（有就用真实的，没有就用保底占位符）
用原生 openai.AsyncOpenAI 客户端直接调用

python

api_messages = prepare_deepseek_payload(cleaned_messages)

response_raw = await self.raw_client.chat.completions.create(
    model=os.getenv("DEEPSEEK_MODEL"),
    messages=api_messages,
    tools=openai_tools,
    extra_body={"thinking": {"type": "enabled"}},
    temperature=0.1,
)

同时还要把 LangChain 的工具对象转换成 OpenAI 原生格式：

python

openai_tools = [convert_to_openai_tool(tool) for tool in ALL_TOOLS]

这一步做完，回滚流程彻底打通。

6. 实时推送：WebSocket 通知审核人员

当工具节点判定需要人工介入时，不能只写数据库等人来查，而是要通过 WebSocket 主动推送给 B 端审核人员：

python

ticket_payload = {
    "type": "NEW_RISK_TICKET",
    "session_id": session_id,
    "risk_details": interrupt_info,
}
await manager.broadcast_to_b(ticket_payload)

这样审核人员能实时看到新工单，而不是轮询刷新。

7. 实战总结与思考

HITL 和 Checkpointer 看起来是两个独立的技术点，但真正落地时是深度耦合的：没有 Checkpointer，HITL 的挂起状态就无法持久化；没有 HITL，Checkpointer 的回滚能力也失去了业务价值。

踩过的坑也很有代表性：

interrupt() 的上下文注入问题，反映出异步环境下 Runnable 配置传递的脆弱性
reasoning_content 丢失问题，本质上是框架序列化与模型 API 之间的协议差异，框架不一定能帮你兜底，关键时候还是得理解底层协议

在算力有限、模型 API 各不相同的现实环境下，工程能力的重要性甚至超过模型能力。这也是我坚持做这套实战练习的意义所在。

万法归宗

许可协议: CC BY 4.0