LangGraph 实战第 2 课：给 Agent 装一道人工审批闸门

生命不息，折腾不止。上一篇横向对比完三家框架，这篇我们真刀真枪上手 LangGraph——手把手给 Agent 装一道「人工审批」闸门，让它该停就停，顺便把「状态契约」这四个字彻底讲透。

一、先补上一课的坑：状态契约到底是个啥

上一篇我说要带你们亲手搭一个「带人工审批的多步 Agent」。动手之前，得先把 LangGraph 最容易把人绕晕的核心概念讲清楚，因为它决定你后面写图顺不顺手——状态契约（State）。

一句话：State 就是这张图里所有节点共享的一张「表格」，每个节点进来能读、出去能写，字段的格式是写死的。 节点和节点之间不直接互相传参，而是通过这张共享的表来交换数据。所以它才叫「契约」——上游写什么、下游读什么，全靠这一张表约定好。

用代码表达，State 通常是一个 TypedDict：

from typing import TypedDict

class DocState(TypedDict):
    draft: str          # 生成的草稿
    decision: str       # 人工审批结果：approve / reject
    status: str         # 最终状态：published / rejected

每个键叫一个「通道（channel）」。这里有两个要命的知识点，80% 的人在这里栽跟头：

第一，节点返回的是「增量」，不是整张表。 一个节点只返回它想改的那几个键，没提到的键 LangGraph 自动保留。比如草稿节点只 return {"draft": "..."}，它不碰 decision，decision 就保持原样。

第二，通道默认是「覆盖」，想「累加」得用 reducer。 同一个键，新值默认把旧值盖掉。可如果你想让一个列表越写越长（比如对话记录），就得用 Annotated 挂一个 reducer，告诉 LangGraph「这个键用合并的方式更新」：

from typing import Annotated
from langgraph.graph.message import add_messages

class DocState(TypedDict):
    messages: Annotated[list, add_messages]   # 累加，别覆盖
    draft: str                                 # 覆盖就行

这就是「状态契约」的全部秘密：一张定义好字段和合并规则的表，节点围着它读写。 搞懂它，下面的人工审批就是往这张表里加两个字段的事。

二、环境准备

装两个包就够，LangGraph 自带图引擎，接模型用 langchain-openai：

1	pip install langgraph langchain-openai

模型这一块，LangGraph 本身不带模型，你接哪家都行。我用 OpenAI 兼容协议，如果你也走 GPT/Claude 这类模型，又想要国内直连、免绑卡、人民币按量付费，可以把 base_url 指到中转站（比如 ai.aklibk.com），这样不用折腾海外卡和网络：

import os
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    base_url="https://ai.aklibk.com/v1",      # OpenAI 兼容中转，按需换
    api_key=os.getenv("AKLIBK_API_KEY"),      # 中转站后台生成的 key
)

不换中转也能跑，base_url 不填就走 OpenAI 官方。这行配置和「人工审批」没有强绑定，纯粹是让新手少走一步弯路。

三、搭图：写草稿 → 人工审批 → 发布

目标：Agent 根据一个主题生成推文草稿，然后停下来把草稿交给人看，人点了「批准」才发布，点「否决」就取消。三步走。

第一步，定义状态契约。 一张表装下全程要流转的字段：

from typing import TypedDict, Literal, Annotated
from langgraph.graph.message import add_messages

class DocState(TypedDict):
    messages: Annotated[list, add_messages]  # 全程对话/草稿，累加
    draft: str        # LLM 生成的草稿
    decision: str     # 人的决定：approve / reject
    status: str       # 终点状态：published / rejected

第二步，写节点。 每个节点就是一个普通函数，吃 state、吐增量 dict。关键的「人工审批」节点里放 interrupt()——这就是那道闸门：

from langgraph.types import interrupt

def write_draft(state: DocState) -> dict:
    """根据主题生成草稿。"""
    topic = state["messages"][-1].content
    resp = llm.invoke([{"role": "user",
                        "content": f"围绕「{topic}」写一条 100 字以内的推文草稿。只给正文。"}])
    return {"draft": resp.content}

def human_review(state: DocState) -> dict:
    """闸门：图形执行到这里就停，草稿交给人拍板。"""
    decision = interrupt({
        "question": "草稿已生成，是否发布？",
        "draft": state["draft"],   # 把草稿一起抛给外面的人看
    })
    return {"decision": decision}  # decision 是 resume 时返回的值（"approve"/"reject"）

def publish(state: DocState) -> dict:
    return {"status": "published"}

def cancel(state: DocState) -> dict:
    return {"status": "rejected"}

第三步，连线并编译。 这里必须挂一个 checkpointer（检查点存储器），否则 interrupt() 直接报错——因为「暂停」本质是把状态存进抽屉，没抽屉就没法暂停：

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

builder = StateGraph(DocState)
builder.add_node("write_draft", write_draft)
builder.add_node("human_review", human_review)
builder.add_node("publish", publish)
builder.add_node("cancel", cancel)

builder.add_edge(START, "write_draft")
builder.add_edge("write_draft", "human_review")

def route_after_review(state: DocState) -> Literal["publish", "cancel"]:
    return "publish" if state["decision"] == "approve" else "cancel"

builder.add_conditional_edges("human_review", route_after_review)
builder.add_edge("publish", END)
builder.add_edge("cancel", END)

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

到这里，图就搭好了：写草稿 → 人工审批（卡住）→ 看决定 → 发布或取消。

四、第一次跑：卡在 interrupt，看抽屉里的东西

第一次 invoke 会一路跑到 human_review 里的 interrupt()，然后停住，控制权交回给你。注意 config 里的 thread_id，它是这个会话的身份证，恢复时靠它找回被暂停的那次执行：

from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "doc-20261015"}}

result = graph.invoke(
    {"messages": [HumanMessage(content="帮我把今天 LAN Base 的开源发布写条推文")]},
    config,
)

# 图形暂停后，__interrupt__ 里就是草稿和问题
print(result.get("__interrupt__"))
# 你会看到类似：
# [Interrupt(value={'question': '草稿已生成，是否发布？', 'draft': '...草稿正文...'})]

看懂这行输出，就懂了整个机制：interrupt() 把「问题 + 草稿」抛出来，状态被 checkpointer 存进 thread_id 对应的抽屉，图形就此打住。此时你的 status 还是空，因为 publish/cancel 根本没执行。

五、放行还是否决：Command(resume=…) 恢复

真人看完草稿，给个决定，用 Command(resume=...) 把决定塞回同一个 thread_id：

from langgraph.types import Command

# 批准 → 图形从 interrupt 处继续，decision="approve" → 路由到 publish
final = graph.invoke(Command(resume="approve"), config)
print(final["status"])   # -> "published"

# 否决同理，换一个 thread_id 再来一次
config2 = {"configurable": {"thread_id": "doc-20261015-reject"}}
graph.invoke({"messages": [HumanMessage(content="写条敏感话题推文试试")]}, config2)
final2 = graph.invoke(Command(resume="reject"), config2)
print(final2["status"])  # -> "rejected"

Command(resume=...) 的参数会原封不动变成 interrupt() 的返回值，也就是节点里 decision 拿到的值。批准路由到 publish，否决路由到 cancel，status 各有归宿。整套「人工审批」就是这么几个动作，没有魔法。

六、三个必踩的坑 + 下一步

坑要提前说，别等报错再懵：

坑一：没挂 checkpointer 就上 interrupt()。 编译器会直接抛错（提示 interrupt 需要 checkpointer）。上面例子里的 InMemorySaver() 一定要 compile(checkpointer=...) 传进去。老教程里常见 from langgraph.checkpoint.memory import MemorySaver，新文档统一成了 InMemorySaver，是同一个内存检查点，按你装的版本报错提示 import 就行。

坑二：恢复时换了 thread_id。 暂停好比把一个执行放进抽屉，抽屉编号就是 thread_id。你重新 invoke(Command(resume=...)) 时 thread_id 对不上，等于去一个新抽屉里找旧的执行，根本恢复不了，既拿不到草稿也续不上流程。

坑三：内存 checkpointer 一重启就蒸发。 InMemorySaver 是给演示和调 bug 用的，数据全在内存里，进程一挂就没。真要上线，换成持久化的 SqliteSaver（pip install langgraph-checkpoint-sqlite），暂停的执行落盘，服务重启了照样能恢复。

到这里，「状态契约」和「人工审批」这两个最核心的坎你已经迈过去了。剩下的 LangGraph，无非是往同一张表上多挂几条边、多排几个节点。

生命不息，折腾不止。下一篇「LangGraph 实战第 3 课」：把这个审批 Agent 的 checkpointer 从内存换成皮实的 SqliteSaver，再实测 get_state / update_state 怎么做「时间旅行」——把跑错的步骤倒回去改完再重来，让 Agent 真的能「后悔」。

文章作者: 空缺

文章链接: https://kqnb.me/2026/10/15/langgraph-lesson2-human-approval/