用持久化与人工审批补齐 Agent 执行链

Agent 能生成草稿，不等于能放心替用户发送邮件或提交申请。作者给出的方案很克制：用 LangGraph 编排流程，用 Postgres 保存执行状态，再把人工审批放在真正执行动作之前。目标是让任务在进程重启后能够恢复，并让人决定外部动作是否发生。

这套模式值得看的地方，是它把可靠性拆成了三个问题：状态能否找回、动作有没有授权、失败重试会不会重复执行。前两个有了框架支持，第三个仍需要业务代码兜底。

为什么要把草稿、审批、执行拆开

作者把流程分成三个节点：draft 生成草稿，approval 等待批准，act 执行外部动作。人工关卡（human gate）负责暂停流程，检查点（checkpoint）负责记录状态，两者解决的问题不同。

下面这段代码展示了审批怎样控制执行入口：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command

class State(TypedDict):
    task: str
    draft: str
    approved: bool

def draft_node(state: State) -> dict:
    # call your LLM here; validate its output with Pydantic before trusting it
    return {"draft": f"Proposed action for: {state['task']}"}

def approval_node(state: State) -> dict:
    # pauses the run and hands the draft back to the caller
    decision = interrupt({"draft": state["draft"], "question": "Approve?"})
    return {"approved": decision == "approve"}

def act_node(state: State) -> dict:
    if state["approved"]:
        ...  # the only place with external side effects
    return {}

builder = StateGraph(State)
builder.add_node("draft", draft_node)
builder.add_node("approval", approval_node)
builder.add_node("act", act_node)
builder.add_edge(START, "draft")
builder.add_edge("draft", "approval")
builder.add_edge("approval", "act")
builder.add_edge("act", END)

关键是 decision == "approve" 和 if state["approved"]：只有明确批准，才会进入实际动作。act 节点仍然会被调度，但未批准时不会执行其中的外部操作。

作者强调，带外部副作用的业务动作只放在审批后的节点里。比如发送邮件、提交申请，都不能混在生成草稿的过程中。审批前的步骤也要设计成可以安全重跑，不能仅靠节点名称保证安全。

另外，示例里的草稿只是字符串拼接，执行动作也还是占位符，不能把它当成完整业务实现。

Postgres 保存的是任务进度，不是执行承诺

内存状态会跟着进程消失。作者用 PostgresSaver 编译图，让每一步的状态按 thread_id 保存：

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/agents"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates the checkpoint tables once
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "job-1234"}}
    result = graph.invoke({"task": "apply to job 1234"}, config)
    # the run stops at interrupt(); the result carries the interrupt payload

这里要盯住 thread_id。它应该对应一个业务对象，例如职位申请、工单或订单，而不是某次进程启动。这样恢复任务时，查业务对象就能找到对应的执行记录。checkpointer.setup() 用于首次创建检查点表。

流程执行到 interrupt() 后暂停，并把待审批内容交给调用方。批准可以来自 Web UI、聊天消息或 CLI，再通过 graph.invoke(Command(resume="approve"), config) 恢复同一个线程。

检查点能保存进度，但不能保证外部动作只发生一次。

作者也明确提醒：节点执行到一半崩溃，恢复后仍可能重跑副作用。因此，即使审批已经通过，发送或提交逻辑仍要尽量做到幂等，也就是重复调用不会产生重复业务结果。原文没有提供这部分的具体实现。

人工审批之前，还要拦住哪些问题

作者在实际运行中补了几条规则。

首先，LLM 输出进入状态前，要用 Pydantic 校验。它负责挡住结构不符合预期的结果，但结构正确不代表事实正确。

所以作者还增加了 verifier 步骤：把生成内容与来源事实核对，拒绝缺乏依据的内容，再送给人审批。这样审核者可以把精力放在判断动作是否合适，而不是逐项寻找模型编造的细节。

最后，把 token 用量记录到 thread_id 下。这样看到的是某个业务对象消耗了多少，而不只是整个服务的一笔总账。

这些都是作者转述的运行经验。原文没有给出 verifier 的评分方法、阈值、误判率，也没有展示成本或可靠性的对照数据。

我的判断

我认可这里的职责划分：模型生成、人来授权、数据库保存进度、业务代码处理重复执行。每一层都能回答一个明确的问题，出了故障也更容易定位。

但「加上持久化和审批，就能用于生产」仍然说得偏满。示例没有交代审批者身份校验、并发审批、超时处理，以及外部动作成功但状态尚未保存时怎样恢复。这些都需要应用自己补齐。

如果流程始终只有固定的几个步骤，普通数据库状态机配合任务队列也可以承载相同思路。LangGraph 的便利在于把图编排、检查点和中断放到一套接口里；原文没有证明它比替代方案更快或更省开发成本。

「无需新增基础设施」也有前提：团队已经运行 Postgres，并愿意让它承担这部分状态存储。

能拿来做什么

• 给 AI 自动化增加审批页时，可以展示 interrupt() 返回的草稿，并让后端根据业务对象找到对应 thread_id，再恢复任务。
• 做工单、申请或订单自动化时，把生成内容与实际提交拆成独立步骤，在提交入口处理重复执行，避免把批准误当成防重保障。
• 在日常 AI 编程中，可以要求生成代码分别实现 Pydantic 结构校验、来源事实核验和按线程记录 token，再逐项检查实现是否完整。