先看一条会中途失败的流程
设想一个可复现的场景：16:00，用户提交“LangGraph1 持久化”，系统开始调研。它要拆分问题、检索官方资料、整理证据、生成草稿，最后等待人工审核。理想路径很直观：
查看 Mermaid 源码
flowchart LR
A[输入主题] --> B[拆分问题]
B --> C[检索资料]
C --> D[整理证据]
D --> E[生成草稿]
E --> F{人工审核}
F -->|退回意见| E
F -->|通过| G[发布文章]审核退回会回到生成草稿，只有通过才进入发布。
真正让系统变复杂的不是步骤数量，而是时间线：16:03，第三个来源超时；16:05，草稿生成；16:06，页面进入“等待审核”；第二天，审核者才回复，浏览器还因为网络抖动重试了一次发布请求。此时我们必须知道已经完成了什么、当前停在哪里，以及下一次应该从哪个事实继续。
函数链把关键事实藏起来了
最初的实现可能只有几行普通 Python：
questions = split_questions(topic)
sources = collect_sources(questions)
evidence = organize_evidence(sources)
draft = write_draft(topic, evidence)
decision = ask_human(draft)
if decision == "通过":
publish(draft)
else:
draft = revise(draft, decision)短流程用它没有问题，但一旦拆成多个 HTTP 请求，几个关键事实就没有地方保存。比如第一次请求结束时，内存里可能是：
topic = "LangGraph 持久化"
sources = ["官方文档 / persistence", "官方文档 / interrupts"]
draft = "第 1 版草稿"
decision = 等待审核如果服务在这时重启，这些局部变量和调用栈都会消失。第二天审核接口只收到 通过，却无法判断它对应哪个主题、哪一版草稿，也不知道检索是否已经完成。
“重启后找不到草稿”和“审核时不能一直挂住请求”看起来是两个问题，根源却相同：流程中的事实还停留在一次函数调用里。只要服务重启，或一次 HTTP 请求结束，主题、草稿和“等待审核”就没有一个可以继续读取的位置。
所以，拆分工作单元之前要先回答一个更基础的问题：这些单元共享哪些数据，谁可以修改它们，暂停或重试时又要保留什么。LangGraph 用 State 把这份运行状态写成明确的数据契约。
State 不是随手传递的字典
State 是一次运行的数据契约。它要回答三个具体问题：字段由哪个节点写入，后续哪个节点会读取，运行暂停或重试时是否需要保留。它不是一个用来暂存所有对象的垃圾桶：
from typing_extensions import TypedDict
class ResearchState(TypedDict, total=False):
topic: str
questions: list[str]
sources: list[str]
evidence: list[str]
draft: str
review: str这份 State 可以用表格读懂：
| 字段 | 谁写入 | 什么时候需要 |
|---|---|---|
topic | 入口 | 全流程 |
questions | 拆分节点 | 检索节点 |
sources | 检索节点 | 证据整理 |
draft | 生成节点 | 审核和发布 |
review | 审核节点 | 路由到修改或发布 |
表格只说明字段归属，真正运行时还要分清三件事：节点读到的 State、节点返回的局部更新，以及运行时合并更新后的 State。以一次输入为例：
查看 Mermaid 源码
flowchart TB
s0["State<br/>topic"]
n1["split_questions<br/>读取：topic<br/>返回：questions"]
s1["State<br/>topic + questions"]
n2["collect_sources<br/>读取：questions<br/>返回：sources"]
s2["State<br/>topic + questions + sources"]
n3["write_draft<br/>读取：topic、questions、sources<br/>返回：draft"]
s3["State<br/>topic + questions + sources + draft"]
s0 -->|传入 State| n1
n1 -->|合并局部更新| s1
s1 -->|传入 State| n2
n2 -->|合并局部更新| s2
s2 -->|传入 State| n3
n3 -->|合并局部更新| s3
classDef state fill:#eff6ff,stroke:#2563eb,color:#0f172a
classDef node fill:#fff7ed,stroke:#ea580c,color:#0f172a
class s0,s1,s2,s3 state
class n1,n2,n3 node图中的 State 方框显示当前已有字段，Node 方框显示它实际读取的字段和返回的局部更新；箭头上的“合并局部更新”表示运行时把返回值写回 State，未被修改的字段继续保留。Node 不需要返回完整 State。
如果 collect_sources 超时，最后一份成功状态仍然包含 questions。系统可以据此只重试检索节点；如果把所有数据都藏在函数局部变量里，重试就只能从入口重新拆分问题、重新调用已经成功的步骤。列表是否追加、去重和限长仍然是业务规则，State 类型本身不会自动替你解决。
不过，State 只规定了流程中有哪些事实，并不决定谁在什么时候修改它。要把失败限制在“检索资料”这一步，还需要为计算划出边界，并声明这些边界之间的执行顺序。
Node、Edge 和 compile 各自负责什么
Node 是一个有明确输入和输出的工作单元。读者应该能回答“它读取哪些 State 字段，成功后写回哪些字段，失败时副作用有没有发生”。例如 split_questions 只读取 topic，只写回 questions；collect_sources 只读取 questions，只写回 sources：
def split_questions(state: ResearchState) -> dict[str, list[str]]:
topic = state["topic"]
return {
"questions": [
f"{topic} 的核心抽象是什么？",
f"{topic} 如何处理失败和恢复？",
]
}
def collect_sources(state: ResearchState) -> dict[str, list[str]]:
# 教学替身。生产实现要有来源白名单、超时和有限重试。
return {"sources": [f"官方资料：{question}" for question in state["questions"]]}Edge 只表达控制流。固定边说明必经路径，条件边说明根据状态选择哪条路；路由函数不应该顺便写数据库或发送消息：
from typing import Literal
def route_review(state: ResearchState) -> Literal["revise", "publish"]:
return "publish" if state.get("review") == "通过" else "revise"StateGraph 是声明结构的构建器，compile() 才得到可执行图。构建阶段只是在登记“有哪些节点、它们如何连接”；运行阶段才会按入口 State 调用节点并合并更新。compile() 成功只证明图结构可执行，不代表来源可信、发布幂等或数据已经脱敏：
from langgraph.graph import END, START, StateGraph
builder = StateGraph(ResearchState)
builder.add_node("split_questions", split_questions)
builder.add_node("collect_sources", collect_sources)
builder.add_edge(START, "split_questions")
builder.add_edge("split_questions", "collect_sources")
builder.add_edge("collect_sources", END)
graph = builder.compile()这里先停在两个节点，是有意的。假设检索节点调用了三个来源，前两个成功、第三个超时：如果检索是独立节点，系统可以记录前两个结果并重试第三个；如果“拆题 → 检索 → 生成 → 发布”是一个大节点，失败时只能看到整个函数报错，不知道哪些调用已经成功，更无法安全重做。
现在，State 负责保存事实，Node 负责产生局部更新，Edge 负责决定下一步，compile() 把这些声明变成可执行对象。概念已经连起来，但还需要真正运行一次，确认节点返回的更新会按预期合并进 State。
用一个最小实验确认状态真的在流动
环境按 2026-08-24 的 LangGraph 1.x 文档组织，运行时请将依赖锁定到项目实际验证的版本：
python -m venv .venv
source .venv/bin/activate
pip install "langgraph>=1,<2" "typing-extensions>=4.12"下面的实验不调用模型，只为了确认入口、节点和返回 State 的关系：
from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph
class QuestionState(TypedDict):
question: str
answer: str
def normalize(state: QuestionState) -> dict[str, str]:
return {"question": state["question"].strip()}
def answer(state: QuestionState) -> dict[str, str]:
return {"answer": f"收到问题：{state['question']}"}
builder = StateGraph(QuestionState)
builder.add_node("normalize", normalize)
builder.add_node("answer", answer)
builder.add_edge(START, "normalize")
builder.add_edge("normalize", "answer")
builder.add_edge("answer", END)
graph = builder.compile()
result = graph.invoke({"question": " LangGraph 是什么？", "answer": ""})
print(result)预期输出包含 {"question": "LangGraph 是什么？", "answer": "收到问题：LangGraph 是什么？"}。执行轨迹是：入口带空格，normalize 写回清理后的问题，answer 读取新值并写入答案，最后到达 END。如果只看到最终字符串而无法解释这三步，说明还没有建立 State 的心智模型。
可以再做一次失败验证：让 collect_sources 抛出 TimeoutError。症状是图在检索节点停止，原因是外部请求失败；当前版本的处理是保留已拆分的问题并把错误交给上层，验证方式是日志中能定位到 collect_sources，而不是得到一个伪造的草稿。
这套模型还没有解决什么
目前的图仍然依赖当前进程。进程结束后 State 会消失，人工审核也没有暂停点；外部写入如果失败，重跑节点可能产生重复发布。invoke 能验证一次调用，却不能替代持久化、恢复和幂等设计。
参考资料
- LangGraph Graph API（官方文档，访问日期：2026-08-24）
- LangGraph GitHub（官方仓库，访问日期：2026-08-24）
Footnotes
-
LangGraph 官方文档：用于编排带状态、分支和暂停恢复能力的长流程。 ↩