Agent 工程体系 · 第 44/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

LangGraph 完整基础：State、Node、Edge、Command、Checkpoint 和中断

LangGraph 是一个面向有状态、长时间运行 Agent 的低层编排框架。它不规定提示词、模型调用方式或 Agent 架构，而是提供一套运行时机制，把确定性代码、LLM 调用、工具调用、人工审批和可恢复执行组织成一个图。官方文档将它定位为 Agent orchestration runtime，核心能力包括持久化执行、人机协同、流式输出和状态管理。(docs.langchain.com)

要理解 LangGraph，不能把它简单看成“给 LLM 加几个函数”。更准确的抽象是：

$$G = (V, E, S, \delta)$$

其中：

• $V$ 是节点集合，每个节点代表一次可执行的计算；
• $E$ 是边集合，决定节点之间的转移关系；
• $S$ 是共享状态；
• $\delta$ 是状态转移函数，描述节点如何读取状态并产生状态更新。

一次执行不是直接调用某个函数，而是从 START 开始，按照边选择待执行节点；节点读取当前状态，返回对状态的部分更新；运行时合并这些更新，形成新的状态，再根据边继续调度节点，直到到达 END、发生错误或触发中断。

---

一、先建立执行模型：图不是调用链，而是状态转移系统

在普通程序中，控制流通常由函数调用栈隐式表示：

main -> classify -> search -> answer

在 LangGraph 中，控制流被显式建模为图：

START -> classify -> search -> answer -> END
                  \
                   -> direct_answer -> END

每一个节点只负责一件事情：

输入：当前 State
输出：State 的部分更新，或者 Command

可以形式化为：

$$N_i : S \rightarrow \Delta S$$

这里的 $\Delta S$ 不是完整状态，而是一个“局部更新”。例如当前状态是：

{
    "question": "杭州今天的天气怎么样？",
    "route": None,
    "answer": None
}

分类节点可以只返回：

{
    "route": "weather"
}

运行时将这个更新应用到原状态，得到：

{
    "question": "杭州今天的天气怎么样？",
    "route": "weather",
    "answer": None
}

因此，节点不应该通过返回完整字典来模拟数据库更新。它只需要返回自己负责修改的字段。这样可以减少节点之间的耦合，也使并行节点的合并规则更清晰。

LangGraph 的 StateGraph 是构建器，不是可直接执行的图；必须先调用 .compile() 生成可执行图，之后才能使用 invoke()、stream()、ainvoke() 等方法。(reference.langchain.com)

---

二、State：图中所有节点共享的状态契约

2.1 State 的职责

State 是节点之间传递信息的共享数据结构。它至少承担三类职责：

1. 保存用户输入和中间结果；
2. 保存路由决策、重试次数、审批状态等控制数据；
3. 作为 Checkpoint 的持久化对象。

一个简单的状态可以用 TypedDict 定义：

from typing import Literal
from typing_extensions import TypedDict


class AgentState(TypedDict):
    question: str
    route: Literal["weather", "general"] | None
    answer: str | None

这里的类型声明主要用于开发期检查、图可视化和代码可读性。它不等价于运行时的完整数据校验。若需要严格验证输入，可以在节点入口使用 Pydantic 等工具，但不要把“类型标注存在”误认为“所有外部输入都已校验”。

2.2 节点读取完整 State，写入部分 State

def classify(state: AgentState):
    question = state["question"]

    if "天气" in question:
        return {"route": "weather"}

    return {"route": "general"}

节点拿到的是当前状态，但返回的只是局部更新。状态变化过程可以表示为：

初始状态：
{
  question: "杭州今天的天气怎么样？",
  route: null,
  answer: null
}

classify 返回：
{
  route: "weather"
}

合并后：
{
  question: "杭州今天的天气怎么样？",
  route: "weather",
  answer: null
}

这也是 LangGraph 与传统“节点返回下一个函数”的差别：节点的输出首先是状态更新，控制流由 Edge 或 Command 决定。

2.3 默认更新语义：覆盖

如果某个 State 字段没有定义 reducer，新的值通常会覆盖旧值：

class State(TypedDict):
    answer: str

节点一：

def draft(state: State):
    return {"answer": "初稿"}

节点二：

def revise(state: State):
    return {"answer": "修订稿"}

执行后：

answer = "修订稿"

这适合 answer、status、route 这类单值字段，但不适合多个并行节点同时向一个列表追加结果。

2.4 Reducer：定义多个更新如何合并

Reducer 是一个二元函数：

$$r(old, update) \rightarrow new$$

例如使用 operator.add 将多个列表更新追加起来：

import operator
from typing import Annotated
from typing_extensions import TypedDict


class ParallelState(TypedDict):
    results: Annotated[list[str], operator.add]

两个并行节点分别返回：

{"results": ["A"]}
{"results": ["B"]}

Reducer 会计算：

operator.add(["A"], ["B"])

得到：

["A", "B"]

如果没有 reducer，两个节点同时更新同一个字段时，系统无法按你的业务意图安全地合并，可能产生冲突或覆盖。因此，Reducer 不是“列表字段的语法糖”，而是并发状态合并协议。

一个常见反例是：

class BadState(TypedDict):
    results: list[str]

然后让两个并行节点都返回：

{"results": ["A"]}
{"results": ["B"]}

工程师如果期待结果是 ["A", "B"]，但没有声明 reducer，那么这个期待没有被 State 契约表达出来。正确做法是明确声明追加语义，或者让每个分支写入不同的键。

2.5 MessagesState 与自定义 State

对于对话型 Agent，可以使用内置的 MessagesState：

from langgraph.graph import MessagesState


def reply(state: MessagesState):
    return {
        "messages": [
            {
                "role": "ai",
                "content": "你好，我已经收到你的问题。",
            }
        ]
    }

但实际项目通常还需要额外字段，例如：

from langgraph.graph import MessagesState


class ChatState(MessagesState):
    user_id: str
    intent: str | None
    approved: bool | None

messages 用于对话上下文，intent 用于路由，approved 用于人机协同。不要把所有数据都塞进消息列表；控制状态应该使用独立字段，否则路由节点需要从自然语言消息中重新推断程序状态。

---

三、Node：执行计算并产生状态更新

3.1 Node 的基本契约

一个节点通常是一个可调用对象：

def node(state: State) -> dict:
    ...

它可以：

• 读取状态；
• 调用模型；
• 调用工具；
• 访问数据库或外部服务；
• 返回状态更新；
• 返回 Command；
• 调用 interrupt() 暂停执行。

例如：

class State(TypedDict):
    question: str
    answer: str | None


def answer_node(state: State):
    question = state["question"]

    # 这里可以替换为真实的模型或检索调用
    answer = f"收到问题：{question}"

    return {"answer": answer}

节点的关键边界是：它负责计算，不负责自行调用下一个节点。下一个节点由 Edge 或 Command 决定。

3.2 节点中的副作用

数据库写入、发送邮件、扣款、调用外部 API 都是副作用。它们与纯计算不同，因为节点可能由于：

• 重试；
• 中断后重新执行；
• 从 Checkpoint 恢复；
• Time Travel 重放；

而被再次调用。

例如：

def charge_card(state: State):
    payment_api.charge(state["amount"])
    return {"status": "paid"}

如果该节点执行成功，但在写入 Checkpoint 前进程崩溃，恢复时可能再次扣款。LangGraph 的持久化执行不能自动把任意外部 API 变成 exactly-once。生产系统需要使用幂等键、业务事务或外部操作日志：

def charge_card(state: State):
    request_id = state["payment_request_id"]

    payment_api.charge(
        amount=state["amount"],
        idempotency_key=request_id,
    )

    return {"status": "paid"}

这里的 payment_request_id 应该在状态中稳定保存。恢复或重放时，外部服务通过幂等键识别重复请求。

3.3 节点错误与恢复

节点异常时，图不会自动推断业务含义。例如以下错误：

def search_node(state: State):
    return {"documents": search(state["question"])}

可能失败于：

• 网络超时；
• 鉴权失败；
• 返回数据格式变化；
• 业务无结果；
• 服务限流。

这些情况应区分处理：

def search_node(state: State):
    try:
        documents = search(state["question"])
    except TimeoutError:
        return {
            "status": "retryable_error",
            "error": "search timeout",
        }
    except PermissionError:
        return {
            "status": "fatal_error",
            "error": "search permission denied",
        }

    return {
        "documents": documents,
        "status": "ok",
    }

然后由条件边将 retryable_error 路由到重试节点，将 fatal_error 路由到结束或人工处理节点。不要让所有异常都被统一包装为“模型失败”，否则运行时无法建立正确的恢复路径。

---

四、Edge：把状态变化连接成控制流

4.1 普通 Edge

普通边表示固定转移：

graph.add_edge("classify", "answer")

它的语义是：

classify 执行完成
    -> answer 一定会被调度

入口和出口也是边：

from langgraph.graph import START, END

graph.add_edge(START, "classify")
graph.add_edge("answer", END)

START 和 END 是虚拟节点。START 表示图的入口，END 表示终止状态。

4.2 条件 Edge

条件边需要一个路由函数：

def route(state: AgentState):
    return state["route"]


builder.add_conditional_edges(
    "classify",
    route,
    {
        "weather": "weather_node",
        "general": "general_node",
    },
)

完整路径是：

START
  -> classify
      -> weather_node
          -> END

      -> general_node
          -> END

路由函数本身通常只负责返回目标节点名，不负责修改状态：

def route(state: AgentState) -> str:
    if state["route"] == "weather":
        return "weather_node"

    return "general_node"

如果既要更新状态，又要改变控制流，使用 Command 更直接；如果只需要路由，条件边的职责更清晰。官方文档也将 Command 定位为“同时进行状态更新和路由”的机制。(docs.langchain.com)

4.3 并行与 Superstep

当一个节点指向多个下游节点时，这些节点可以在同一个后续执行阶段被调度：

builder.add_edge("plan", "search_web")
builder.add_edge("plan", "search_db")

可以把它理解为：

第 0 步：plan
第 1 步：search_web || search_db
第 2 步：aggregate

如果两个并行节点都写入 results，就必须定义 reducer：

class State(TypedDict):
    query: str
    results: Annotated[list[str], operator.add]

否则，两个并行更新之间没有明确的合并规则。

并行执行也带来两个边界：

1. 外部调用完成顺序不应被当作业务顺序；
2. reducer 必须满足业务可接受的合并语义。

例如，operator.add 对列表是追加，但追加顺序在并发场景下不一定代表搜索结果的优先级。如果业务要求稳定排序，应该让结果携带来源和排序键，最后由聚合节点显式排序：

{
    "results": [
        {"source": "web", "score": 0.8, "text": "..."},
        {"source": "db", "score": 0.9, "text": "..."},
    ]
}

4.4 Send：动态 fan-out

当并行任务数量在运行前未知时，可以返回 Send：

from langgraph.types import Send


def fan_out(state):
    return [
        Send("process_item", {"item": item})
        for item in state["items"]
    ]

这里的 process_item 不是接收完整共享状态，而是接收每个 Send 携带的局部输入。它适合 map-reduce：

split
  -> process_item(item1)
  -> process_item(item2)
  -> process_item(item3)
  -> reduce

普通 Edge 表示预先声明的图结构；Send 表示运行时才确定的动态任务集合。官方 Graph API 将它用于 map-reduce 和动态 fan-out。(docs.langchain.com)

---

五、Command：把状态更新和控制流放在同一个返回值中

5.1 Command 的三个主要用途

当前 Python API 中，Command 可以携带：

• update：状态更新；
• goto：转移到指定节点；
• graph：在子图中导航到父图；
• resume：为中断提供恢复值。

其中，节点通常返回 update、goto 和 graph；调用图恢复中断时，输入 Command(resume=...)。(reference.langchain.com)

5.2 节点返回 Command

from typing import Literal
from langgraph.types import Command


def classify(
    state: AgentState,
) -> Command[Literal["weather_node", "general_node"]]:
    if "天气" in state["question"]:
        return Command(
            update={"route": "weather"},
            goto="weather_node",
        )

    return Command(
        update={"route": "general"},
        goto="general_node",
    )

状态和控制流在同一个节点中完成：

classify
  ├─ update route = weather
  └─ goto weather_node

返回类型中的 Literal 不只是装饰。它告诉类型检查器和图渲染工具，该节点可能跳转到哪些节点。官方文档要求使用 Command 做动态路由时声明可达节点。(docs.langchain.com)

5.3 Command 与静态边不能混用表达同一条转移

以下代码容易产生意外：

builder.add_edge("classify", "default_node")


def classify(state):
    return Command(goto="special_node")

这时 default_node 这条静态边仍然存在，special_node 是额外的动态边，两个节点都可能执行。Command 不会删除已经定义的静态边。(docs.langchain.com)

因此，一个节点应当选择一种主要路由方式：

只路由：
节点 -> conditional edge

更新并路由：
节点 -> Command

不要同时为同一个节点配置两套互相竞争的下一步逻辑。

5.4 继续对话时不要误用 Command(update=...)

这是一个重要边界。

错误写法：

graph.invoke(
    Command(update={
        "messages": [
            {"role": "user", "content": "继续刚才的话题"}
        ]
    }),
    config,
)

Command 作为图输入时，运行时会按“从已有 Checkpoint 恢复”的语义处理，而不是像普通新输入一样从 START 开始。图已经完成时，这种写法可能表现为“没有继续执行”。

继续同一线程的新一轮对话应传入普通字典：

graph.invoke(
    {
        "messages": [
            {"role": "user", "content": "继续刚才的话题"}
        ]
    },
    config,
)

只有恢复 interrupt() 时，才使用：

graph.invoke(Command(resume="yes"), config)

官方 Graph API 明确区分了这两种输入语义。(docs.langchain.com)

---

六、Checkpoint：把一次运行变成可恢复的执行历史

6.1 Checkpoint 记录什么

Checkpoint 是某个图执行阶段的持久化快照。它至少需要回答：

• 当前 State 是什么；
• 下一步准备执行哪些节点；
• 这个状态属于哪个 Thread；
• 该快照由哪次执行产生；
• 是否存在等待恢复的任务或中断。

编译图时传入 Checkpointer：

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

graph = builder.compile(
    checkpointer=checkpointer,
)

调用时必须传入 thread_id：

config = {
    "configurable": {
        "thread_id": "conversation-001",
    }
}

result = graph.invoke(
    {"question": "杭州今天的天气怎么样？"},
    config,
)

thread_id 是 Checkpoint 的索引。复用同一个 thread_id，表示继续同一个执行线程；使用新的 thread_id，则表示独立线程。(docs.langchain.com)

6.2 Thread 不是用户，也不是一次 HTTP 请求

Thread 是一条可持续的图执行上下文。它可以对应：

• 一段对话；
• 一个审批流程；
• 一个订单处理流程；
• 一个长期运行的 Agent 任务。

但不要默认把用户 ID 直接当作 thread_id。同一用户可能同时拥有多个会话或多个业务任务：

user_id = user-001

thread_id = chat-2026-09-01-001
thread_id = order-2026-09-01-002

Thread 决定状态隔离边界。用户身份、权限和长期偏好属于业务数据，不能仅依赖 Thread ID 表达。

6.3 Checkpointer 与 Store 的区别

Checkpointer 保存线程内的图状态快照，适用于短期记忆、恢复、中断和 Time Travel；Store 保存应用定义的跨线程数据，适用于用户偏好、长期事实和共享知识。两者是不同层次的持久化机制。(docs.langchain.com)

Checkpointer：
thread-001 -> 当前对话状态、下一节点、历史快照

Store：
user-001 -> 语言偏好、会员等级、长期事实

将用户长期画像塞进每个 Thread 的 State，会导致状态膨胀，也无法自然地跨线程复用。

6.4 查看当前状态和历史

snapshot = graph.get_state(config)

print(snapshot.values)
print(snapshot.next)
print(snapshot.config)

其中：

• values 是当前状态值；
• next 是下一步待执行节点；
• config 包含定位该快照所需的信息。

查看历史：

history = list(graph.get_state_history(config))

for snapshot in history:
    print(
        snapshot.metadata.get("step"),
        snapshot.values,
        snapshot.next,
    )

历史通常按最新快照在前的顺序返回。Checkpoint 使图拥有类似执行日志和调试时间线的能力。(docs.langchain.com)

6.5 故障恢复不是从函数中间恢复

假设图有两个节点：

node_a -> node_b

执行过程：

node_a 成功
写入 checkpoint A
node_b 执行失败

恢复时，运行时可以从最近成功状态继续，而不是重新执行整个图。对于同一个 superstep 中已经成功的其他节点，LangGraph 还会保存 pending writes，使恢复时不必重复运行这些节点。(docs.langchain.com)

但这不等于任意 Python 语句都能从崩溃行号继续。节点内部执行到一半时发生进程崩溃，节点通常会从该节点边界重新开始。因此，节点内部的副作用仍然必须幂等。

6.6 内存 Checkpointer 的边界

InMemorySaver 适合测试和示例：

checkpointer = InMemorySaver()

它把数据放在进程内存中，进程重启后 Checkpoint 丢失。生产环境应使用持久化实现，例如 PostgreSQL 或本地文件型实现，并根据部署场景处理连接、迁移、备份和保留策略。官方文档明确指出，内存型 Checkpointer 不会跨进程重启持久化。(docs.langchain.com)

---

七、中断：把控制权交给外部系统或人

7.1 interrupt 的语义

interrupt(value) 用于在节点执行过程中暂停图，并把 value 交给调用方。它适合：

• 人工审批；
• 修改模型生成的工具参数；
• 等待用户补充信息；
• 进行高风险操作确认。

from langgraph.types import interrupt


def approval_node(state):
    approved = interrupt({
        "type": "approval",
        "message": "是否允许发送邮件？",
        "recipient": state["recipient"],
    })

    return {"approved": approved}

中断需要两个前置条件：

1. 编译图时启用 Checkpointer；
2. 调用图时提供稳定的 thread_id。

否则运行时无法保存暂停位置，也无法在未来找到对应的执行上下文。(docs.langchain.com)

7.2 完整的审批示例

下面示例使用 invoke() 展示最小闭环：

from typing import Literal
from typing_extensions import TypedDict

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


class ApprovalState(TypedDict):
    action: str
    approved: bool | None
    result: str | None


def review(state: ApprovalState):
    approved = interrupt({
        "type": "approval",
        "action": state["action"],
        "message": "是否批准执行该操作？",
    })

    if approved:
        return Command(
            update={"approved": True},
            goto="execute",
        )

    return Command(
        update={
            "approved": False,
            "result": "操作被拒绝",
        },
        goto=END,
    )


def execute(state: ApprovalState):
    # 真实系统中，这里应使用幂等键执行外部副作用
    return {
        "result": f"已执行：{state['action']}",
    }


builder = StateGraph(ApprovalState)
builder.add_node("review", review)
builder.add_node("execute", execute)

builder.add_edge(START, "review")
builder.add_edge("execute", END)

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

config = {
    "configurable": {
        "thread_id": "approval-001",
    }
}

# 第一次执行：在 review 中断
first = graph.invoke(
    {
        "action": "发送付款通知邮件",
        "approved": None,
        "result": None,
    },
    config,
)

print(first["__interrupt__"])

# 恢复执行：interrupt() 的返回值变成 True
second = graph.invoke(
    Command(resume=True),
    config,
)

print(second)

执行路径如下：

第一次 invoke
  -> review
      -> interrupt()
          -> 保存 checkpoint
          -> 返回 __interrupt__
          -> 停止

第二次 invoke(Command(resume=True))
  -> 从同一 thread 恢复
  -> 重新执行 review 节点
  -> interrupt() 返回 True
  -> Command(goto="execute")
  -> execute
  -> END

中断恢复时，节点不是从 interrupt() 调用的下一行继续，而是从该节点开头重新执行。官方文档特别强调了这一点。(docs.langchain.com)

因此下面的写法存在重复副作用风险：

def unsafe_node(state):
    send_email("审查开始")  # 中断恢复时可能再次发送
    approved = interrupt("是否继续？")
    return {"approved": approved}

更安全的方式是把中断放在副作用之前：

def safer_node(state):
    approved = interrupt("是否继续发送邮件？")

    if not approved:
        return {"status": "cancelled"}

    send_email(
        state["recipient"],
        idempotency_key=state["request_id"],
    )

    return {"status": "sent"}

如果中断前必须执行副作用，该副作用必须具备幂等性。中断实现依赖特殊异常机制，不应使用宽泛的 try/except 包住 interrupt()，否则可能把运行时用于暂停的异常捕获掉。(docs.langchain.com)

7.3 中断值与恢复值

中断时：

approved = interrupt({
    "question": "是否批准？"
})

恢复时：

graph.invoke(
    Command(resume=True),
    config,
)

此时：

approved == True

恢复值可以是字符串、布尔值、对象或数组，但应保持可序列化，并在服务端进行校验：

def review(state):
    decision = interrupt({
        "type": "approval",
        "allowed_values": ["approve", "reject"],
    })

    if decision not in {"approve", "reject"}:
        return Command(goto="review")

    return Command(
        update={"decision": decision},
        goto="execute" if decision == "approve" else END,
    )

更复杂的校验通常使用循环：

def collect_age(state):
    while True:
        age = interrupt("请输入年龄")

        if isinstance(age, int) and 0 <= age <= 150:
            return {"age": age}

        # 未通过校验，下一轮再次中断

7.4 多个并行中断

如果多个并行节点同时中断，恢复值不能只依赖位置猜测，应使用中断 ID 与恢复值的映射，把回答绑定到正确的中断任务。官方中断文档将这种情况作为并行人工协同的重要模式。(docs.langchain.com)

---

八、静态断点与动态中断不是一回事

LangGraph 还支持在节点执行前或执行后设置静态断点，例如：

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["execute"],
)

这表示每次运行到 execute 前都暂停，适合：

• 调试；
• 查看某个节点的输入；
• 在开发阶段观察状态。

而动态中断是节点代码主动调用：

approved = interrupt("是否批准？")

它可以基于运行时数据决定是否暂停：

def maybe_review(state):
    if state["risk_score"] >= 0.8:
        approved = interrupt("高风险操作，请审批")
        return {"approved": approved}

    return {"approved": True}

两者的区别是：

静态断点：
由图配置决定，在固定节点边界暂停

动态中断：
由节点逻辑决定，可在节点内部任意位置暂停

审批流程通常使用动态中断；调试流程可以使用静态断点。

---

九、Time Travel：重放与分叉，而不是撤销现实世界的副作用

9.1 Replay

给定历史 Checkpoint，可以从某个历史点重新执行后续节点：

history = list(graph.get_state_history(config))

for snapshot in history:
    print(snapshot.config, snapshot.next)

找到目标快照后，将其配置作为恢复位置：

old_config = history[-1].config
result = graph.invoke(None, old_config)

重放的含义是：

Checkpoint 之前的节点：不再执行
Checkpoint 之后的节点：重新执行

因此，后续的 LLM 调用、网络请求和中断可能再次发生，结果也可能不同。Time Travel 适合调试和探索替代路径，不应被误解为数据库事务回滚。(docs.langchain.com)

9.2 Fork

如果想基于历史状态创建另一条路径，可以先修改状态：

fork_config = graph.update_state(
    old_config,
    {
        "approved": True,
    },
)

result = graph.invoke(None, fork_config)

原始 Checkpoint 不会被修改；update_state() 会创建新的状态版本。若字段定义了 reducer，更新也会经过 reducer，而不是简单覆盖。(docs.langchain.com)

9.3 为什么不能用 Time Travel 撤销扣款

假设：

Checkpoint A
  -> charge_card 成功，真实扣款
Checkpoint B
  -> 后续节点失败

从 A 重放，并不会让银行系统自动撤销第一次扣款。它只会再次执行后续图逻辑。现实世界的副作用必须由业务系统提供补偿操作：

扣款成功 -> 退款
订单创建 -> 取消订单
邮件发送 -> 标记重复请求，不再发送

所以 Checkpoint 保证的是图状态的可观察性和可恢复性，不是外部世界的自动事务一致性。

---

十、一个可运行的端到端图

下面示例包含 State、Node、普通 Edge、条件路由、Checkpoint 和中断：

from typing import Literal
from typing_extensions import TypedDict

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


class State(TypedDict):
    question: str
    route: Literal["known", "unknown"] | None
    answer: str | None
    approved: bool | None


def classify(state: State) -> Command[Literal["known_answer", "ask_human"]]:
    if "LangGraph" in state["question"]:
        return Command(
            update={"route": "known"},
            goto="known_answer",
        )

    return Command(
        update={"route": "unknown"},
        goto="ask_human",
    )


def known_answer(state: State):
    return {
        "answer": "LangGraph 使用图、状态和节点组织 Agent 执行。",
    }


def ask_human(state: State):
    approved = interrupt({
        "type": "clarification",
        "message": "这个问题需要人工补充上下文，是否继续？",
    })

    if approved:
        return {
            "approved": True,
            "answer": "已获得人工确认，可以继续处理。",
        }

    return {
        "approved": False,
        "answer": "人工未确认，停止处理。",
    }


builder = StateGraph(State)

builder.add_node("classify", classify)
builder.add_node("known_answer", known_answer)
builder.add_node("ask_human", ask_human)

builder.add_edge(START, "classify")
builder.add_edge("known_answer", END)
builder.add_edge("ask_human", END)

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

config = {
    "configurable": {
        "thread_id": "demo-thread-001",
    }
}

result = graph.invoke(
    {
        "question": "LangGraph 中的 State 是什么？",
        "route": None,
        "answer": None,
        "approved": None,
    },
    config,
)

print(result)

执行过程：

初始：
question = "LangGraph 中的 State 是什么？"

classify：
route = "known"
goto = "known_answer"

known_answer：
answer = "LangGraph 使用图、状态和节点组织 Agent 执行。"

结束：
next = ()

如果输入改成：

{
    "question": "请处理一个需要人工确认的问题",
    "route": None,
    "answer": None,
    "approved": None,
}

执行会在 ask_human 中断：

classify
  -> route = unknown
  -> ask_human
      -> interrupt
      -> 保存 Checkpoint
      -> 返回中断信息

之后使用同一个 config 恢复：

result = graph.invoke(
    Command(resume=True),
    config,
)

如果换成新的 Thread：

new_config = {
    "configurable": {
        "thread_id": "another-thread",
    }
}

运行时找不到原来的中断位置，会把它当作新的执行线程，而不是恢复旧流程。

---

十一、常见误解与诊断路径

11.1 “State 就是全局变量”

不是。State 是由图运行时管理、可被 reducer 合并、可写入 Checkpoint 的状态契约。节点不应通过模块级变量保存流程进度：

current_approval = {}

这种写法无法可靠支持多线程、进程重启、水平扩展和 Time Travel。流程进度应放入 State，跨线程长期数据则放入 Store。

11.2 “节点返回完整 State 更安全”

通常不是。完整返回会让节点覆盖其他节点负责的字段，尤其在并行执行时更危险。节点应尽量只返回自己改变的字段：

return {"answer": answer}

而不是：

return {
    "question": state["question"],
    "route": state["route"],
    "answer": answer,
    "approved": state["approved"],
}

后者增加了覆盖旧值和制造并发冲突的机会。

11.3 “Checkpoint 可以保证 exactly-once”

不能。Checkpoint 主要解决状态持久化、恢复和历史分叉；外部副作用是否 exactly-once，取决于外部系统的幂等协议、事务设计和补偿机制。

诊断一个重复副作用问题时，应依次检查：

1. 节点是否在恢复或重放后重新执行？
2. 副作用是否发生在 interrupt() 之前？
3. 是否配置了幂等键？
4. 外部服务是否真正支持幂等？
5. 是否需要业务补偿操作？

11.4 “中断会从暂停行继续”

不会。恢复时节点从头重新执行。因此：

def node(state):
    prepare_file()          # 可能再次执行
    value = interrupt("输入")
    submit_file(value)

恢复后 prepare_file() 可能再次执行。应将可重复代码设计为幂等，或把不可重复副作用移到中断之后。

11.5 “图已经 compile，就可以直接 invoke”

StateGraph 本身是构建器。必须：

compiled = builder.compile()
compiled.invoke(...)

不能把未编译的 builder 当作运行时对象。(reference.langchain.com)

11.6 “InMemorySaver 适合生产”

它适合单进程测试，不适合需要进程重启后恢复的生产任务。生产环境还要考虑：

• Checkpoint 数据库的高可用；
• Thread ID 的长度和格式；
• 历史保留与清理；
• 敏感状态加密；
• 大消息和大文档是否应该放在对象存储；
• Checkpoint 写入失败时的告警和重试。

长期对话会不断累积 Checkpoint，可能增加存储和延迟，需要设计保留策略。官方文档也提示了 Checkpoint 无界增长问题。(docs.langchain.com)

---

十二、把几个概念放在同一条因果链上

一个完整的 LangGraph 执行可以归纳为以下过程：

输入
  |
  v
START
  |
  v
Node 读取 State
  |
  +-- 返回部分 State 更新
  |       |
  |       v
  |   Reducer 合并
  |       |
  |       v
  |   Checkpoint
  |
  +-- 返回 Command(update, goto)
  |       |
  |       v
  |   更新状态并动态转移
  |
  +-- 调用 interrupt()
          |
          v
      保存 Checkpoint
          |
          v
      等待 Command(resume=...)
          |
          v
      从中断节点开头重新执行

因此：

• State 描述“系统当前知道什么”；
• Node 描述“系统执行什么计算”；
• Edge 描述“默认如何转移”；
• Command 描述“节点如何同时改变状态和控制流”；
• Checkpoint 描述“如何保存和恢复执行上下文”；
• interrupt 描述“如何将控制权交给外部输入，再恢复图执行”。

如果只掌握节点和边，能够构建普通工作流；如果加入 State 和 Reducer，才能正确处理共享状态与并发；如果加入 Checkpoint 和中断，才能构建可暂停、可恢复、可审查的 Agent；如果进一步使用历史快照和分叉，才能把执行过程变成可诊断、可回放的状态机。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：OpenAI Agents SDK 会话工程：历史、Session、流式、取消和恢复
• 下一篇：LangGraph 持久化执行：Thread、Checkpoint、Interrupt、Time Travel 和恢复
• 延伸：Agent 状态机设计：节点、事件、守卫、转移和可恢复执行

官方资料

• LangGraph Documentation
• LangGraph Repository

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。