LangGraph（二）：审核要等到明天，状态如何回来

先看一个“服务没报错但流程丢了”的现场

设想一次实际请求：10:00，资料调研助手生成第 1 版草稿，接口返回“等待审核”；10:02，服务滚动发布并重启；第二天 09:00，审核页面提交“通过”，却得到“找不到运行”或重新生成了一份草稿。浏览器连接早已关闭，但业务流程并没有结束。

这不是前端状态管理的小问题。graph.invoke() 默认只在当前调用里持有 State；HTTP 请求结束后，进程内变量和浏览器内存都不能作为可靠存储。要跨请求继续，系统至少要保存三件彼此关联的事实：

缺少任意一项都会出错：只有 State，不知道下一步；只有位置，没有草稿；只有一个前端按钮，则无法证明它属于哪条运行。恢复不是把页面重新打开，而是把这三项事实重新拼成同一条执行记录。

恢复时，三个事实必须由同一个 thread_id 关联起来：

flowchart LR
    state[当前 State<br/>草稿 / 来源 / 审核意见] --> checkpoint[Checkpoint]
    position[执行位置<br/>暂停在 human_review] --> checkpoint
    identity[thread_id<br/>research-001] --> checkpoint
    checkpoint --> resume[恢复同一条运行]

Checkpoint 保存的是运行事实，不是最终答案

Checkpoint¹ 是某个线程在图执行过程中的持久化快照。它应该让系统回答：已经完成哪些节点，当前 State 是什么，恢复后从哪里继续。thread_id 则是这条运行的稳定身份；同一个输入换一个 ID，不会自动变成同一条工作。

下面只保留持久化相关的新增代码；builder 表示已经注册好节点和边的 StateGraph，ResearchState 是流程状态类型。InMemorySaver 只适合实验，进程退出后数据会消失：

from langgraph.checkpoint.memory import InMemorySaver


checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {
    "configurable": {
        "thread_id": "research-2026-08-24-001",
    }
}

result = graph.invoke(
    {"question": "LangGraph 是什么？", "answer": ""},
    config=config,
)

后续调用必须传入同一个 thread_id，运行时才有机会读取原来的检查点。例如第一次请求用 research-001 在 human_review 暂停，第二次请求仍用这个 ID，才能把审核结果交回原来的草稿；如果第二次生成 research-002，LangGraph 会把它当成新运行，重新执行入口节点。最容易观察这个差异的实验是：只改一位 ID，再比较两次调用返回的 State 和节点轨迹。

生产环境还要定义持久化实现、加密、保留期、清理、并发冲突和 schema 迁移；“接上 Checkpointer”不是完整的恢复方案。

Checkpointer 和 Store 解决的不是同一类问题

假设多个调研线程共享一份“组织已经批准的来源列表”。它跨越多次运行，不属于任何一个具体 thread。相反，某次运行的草稿、当前节点和等待审核的 payload 只服务于该 thread：

把三者统称为“记忆”会掩盖生命周期和权限差异。Store 中的长期数据必须有 namespace、版本和访问控制；未经审核的模型推断、临时 prompt 和凭证不应自动升级为长期记忆。Checkpointer 也不是订单表或审计账本，它记录的是图运行事实。

用 interrupt 把人工等待变成可恢复边界

最容易写错的人工审核实现是在节点里调用 input()，或者让 HTTP 请求一直挂着等待回复。前者绑定进程，后者绑定连接；两种实现都无法承受数小时的等待。

interrupt² 把“需要外部决定”显式变成暂停点。首次执行走到 human_review 时，LangGraph 保存草稿和暂停信息，并把审核 payload 返回给上层；审核接口稍后再用原来的 thread 传入 Command(resume=...)。它不是把 Python 函数挂起，而是把“等待”变成可持久化的运行状态：

from langgraph.types import Command, interrupt


def human_review(state: ResearchState):
    decision = interrupt({
        "kind": "review",
        "draft": state["draft"],
        "message": "请回复‘通过’或填写退回意见",
    })

    if decision == "通过":
        return {"review": "通过"}
    return {"review": str(decision)}


# 首次调用：图在 human_review 处暂停。
# 审核完成后：
graph.invoke(
    Command(resume="通过"),
    config={"configurable": {"thread_id": "research-2026-08-24-001"}},
)

恢复后，resume 的值会作为 interrupt(...) 的返回值交给 human_review，然后条件边读取 review，选择修改或发布。前端拿到的 payload 应是审核需要的标题、草稿摘要和可操作项，不应该把内部 State、检索原文或系统提示词原样下发。

暂停点周围要能安全重做

恢复一个带 interrupt 的节点时，暂停点之前的代码可能再次执行。例如下面的顺序有风险：

send_email("草稿已生成")  # 第一次运行已经发送
decision = interrupt({"draft": state["draft"]})

如果审核恢复导致节点从头执行，邮件可能再次发送。可以用一个假邮件函数验证：第一次运行打印一条 send_email，恢复后如果又打印一条，说明副作用放错了位置。读取数据、生成确定性摘要通常可以重做；发送邮件、创建订单、发布文章则必须移到暂停点之后，或使用幂等键和“是否已经成功”的查询保护。

更稳妥的流程是：

flowchart LR
    A[生成草稿] --> B[保存草稿版本]
    B --> C{interrupt 等待审核}
    C -->|退回| A
    C -->|通过| D[幂等发布]

暂停之前的写入要么可重复，要么有稳定幂等键；发布节点应在审核通过之后才执行。

如果副作用必须发生在暂停之前，就要保存外部请求 ID，并在恢复时先查询“是否已经成功”，不能直接再次创建。Checkpoint 能让图回到一个位置，却不会替外部系统补上幂等语义。

三条容易被忽略的失败路径

这些验证分别覆盖了三种不同风险：状态找不到是恢复问题，重复发布是写入问题，越权读取是安全问题。页面显示“已暂停”只能证明 UI 收到了一个状态事件，不能证明这三个边界已经成立。

流式输出能观察，不负责恢复

长流程可以用 graph.stream(..., stream_mode="updates") 把节点更新交给页面，帮助用户区分“正在检索”和“已经失败”。但流式连接本身不是持久化机制：浏览器断线后，服务端是否继续运行、事件是否可重放、前端如何补齐，都要单独设计。

可以把两种数据分开：事件流展示变化，GET /api/runs/{id} 返回当前快照。页面刷新先取快照，再从快照序号之后订阅事件；这样错过某条事件也不会只能猜测当前状态。事件流和 UI 消费都应建立在这份快照之上。

现在我们能回答“从哪里继续”，却还不能回答“失败后哪些工作可以重做”。搜索超时、模型输出不符合约定和发布接口重复写入，需要不同的错误策略；恢复能力也必须和幂等、预算一起设计。

参考资料

• LangGraph Persistence（官方文档，访问日期：2026-08-24）
• LangGraph Interrupts（官方文档，访问日期：2026-08-24）
• LangGraph Streaming（官方文档，访问日期：2026-08-24）