记住一句话：节点之间从不直接传参，它们只通过 State 通信。Node A 不会把返回值『交给』Node B，而是把更新写进 State，框架合并后，Node B 读到的是合并后的新 State。这条约束消除了显式的数据传递分支，是 LangGraph 可持久化、可回放、可并行的根本原因。

State：唯一的共享数据结构

State 是整张图共享的『内存』。它通常用 TypedDict 定义（也支持 Pydantic BaseModel 与 dataclass）。图运行时维护一个 State 实例，每个节点读取它、返回一个『只包含要更新字段』的字典，框架按字段把更新合并进去——没有声明 reducer 的字段，默认行为是 覆盖（overwrite）。

# 安装：langgraph 自身不强依赖任何 LLM 厂商
# pip install -U langgraph

from typing import TypedDict

# State 用 TypedDict 声明字段与类型
# 注意：这里没有任何 reducer，所以字段默认是『覆盖』语义
class MyState(TypedDict):
    question: str      # 用户输入
    answer: str        # 节点写入的答案
    step_count: int    # 计数器

# State 只是一个『形状声明』，运行时它就是一个普通 dict
# 节点拿到的 state 形如：{"question": "...", "answer": "...", "step_count": 0}

Node：读 State，返回部分更新

Node 就是一个普通可调用对象（函数或 Runnable），签名是 (state) -> dict。它接收当前完整 State，返回一个 只含要更新字段的 dict（partial state）。返回 {} 表示『我不改任何状态』，返回 None 也合法。关键纪律：节点应当是纯函数式的——不要原地修改传入的 state，而要返回新的更新 dict，让框架去合并。

from typing import TypedDict

class MyState(TypedDict):
    question: str
    answer: str
    step_count: int

# 一个纯函数节点：读 state.question，写回 answer 和 step_count
# 注意它只返回『要更新的字段』，没碰 question
def answer_node(state: MyState) -> dict:
    q = state["question"]
    result = f"你问的是：{q}，答案是 42"
    # 返回 partial state —— 框架会把这两个字段 merge 进主 State
    return {
        "answer": result,
        "step_count": state.get("step_count", 0) + 1,
    }

# 反例（不要这样做）：原地修改 state
# def bad_node(state):
#     state["answer"] = "..."   # 绕过了框架的合并 / 持久化机制
#     return state              # 也违背 partial-return 约定
✓推荐做法
  • 节点只返回『改了的字段』，让 diff 一目了然
  • 把节点写成纯函数：相同输入 → 相同输出，便于测试与回放
  • 副作用（调 LLM、查库）可以有，但状态变更一律走 return
✗不推荐
  • 不要原地 mutate 传入的 state 再 return state
  • 不要在节点里手动拼下一个节点该收到什么——那是 Edge 的职责
  • 不要返回未在 State 里声明的字段（默认会被忽略或报错）
⚠常见误区
  • 返回整个 state 看似『没错』，但会让 update stream、reducer、checkpoint diff 失真
  • 在节点里持有可变全局变量，会破坏 checkpointer 的可回放性

把任意一个节点单独拎出来，喂一个构造的 state dict，能独立得到确定的返回 dict —— 满足即合格。

Edge 与 START / END：控制流从哪来、到哪去

Node 负责『算什么』，Edge 负责『接下来执行谁』。Edge 分两种：普通边（静态：A 跑完无条件去 B）和 条件边（运行时调用一个路由函数，根据当前 State 决定去哪个节点，本系列在 conditional-edges-routing 章详解）。START 和 END 是两个特殊的虚拟节点：START 标记图的入口（从它连出的边决定第一个真正执行的节点），END 标记终止（走到它图就停）。它们不执行任何逻辑，只是控制流的锚点。

你想表达的意图用什么API 写法
图从哪个节点开始从 START 连普通边graph.add_edge(START, "node_a")
A 跑完总是去 B普通边graph.add_edge("node_a", "node_b")
A 跑完根据状态分流条件边graph.add_conditional_edges("node_a", route_fn)
执行到此结束连到 ENDgraph.add_edge("node_b", END)
口诀START 进、普通边直走、条件边分叉、END 收尾 —— 四件事覆盖所有控制流。

把它们拼起来：START → node → END 最小图

现在用 StateGraph 把上面四个抽象组装成一张能跑的完整图：声明 State → add_node 注册节点 → add_edge 连接 START、节点、END → compile() 得到可执行图 → invoke() 喂初始 State 拿最终 State。这是 LangGraph 一切复杂图的骨架。

# pip install -U langgraph
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1) 定义 State
class MyState(TypedDict):
    question: str
    answer: str
    step_count: int

# 2) 定义节点（纯函数，返回 partial state）
def answer_node(state: MyState) -> dict:
    return {
        "answer": f"你问的是：{state['question']}，答案是 42",
        "step_count": state.get("step_count", 0) + 1,
    }

# 3) 构图：注册节点 + 连边
builder = StateGraph(MyState)
builder.add_node("answer", answer_node)   # 节点名 -> 函数
builder.add_edge(START, "answer")          # 入口：START -> answer
builder.add_edge("answer", END)            # 出口：answer -> END

# 4) 编译成可执行图
graph = builder.compile()

# 5) invoke：传入初始 State（只需给必要字段），拿回合并后的最终 State
final_state = graph.invoke({"question": "生命的意义", "step_count": 0})
print(final_state)
# {'question': '生命的意义', 'answer': '你问的是：生命的意义，答案是 42', 'step_count': 1}

执行模型：超步 super-step 与 partial state 合并

LangGraph 的执行不是『一个节点接一个节点的函数调用栈』，而是借鉴 Google Pregel 的 超步（super-step） 模型。每个超步分三拍：(1) 选出当前『被激活』的节点（上一步的边指向它们）；(2) 这批节点 并行 执行，各自返回 partial state；(3) 框架把这批返回值按字段、按 reducer 合并回主 State，形成下一个超步的输入。如此循环，直到没有节点被激活或走到 END。

  1. 超步 0：从 START 连出的边激活第一批节点（最小图里就是 answer）
  2. 节点执行：answer 读当前 State，返回 {"answer": ..., "step_count": ...}
  3. 状态合并：按字段 merge —— 无 reducer 的字段覆盖，有 reducer 的字段按 reducer 累积
  4. 推进边：answer 的出边指向 END，没有新节点被激活，图终止
  5. 返回最终合并后的完整 State

节点返回了整个 state

典型表现
update 模式的 stream 里每步都吐出全部字段，diff 失真；带 reducer 的字段出现重复累积
判断标准
节点返回的 dict 只包含本节点真正修改的字段
解决方向
改成只 return 变更字段，例如 return {"answer": result}，不要 return state

忘记连 START 或 END

典型表现
compile() 或 invoke() 报错，提示图没有入口 / 节点不可达 / 永不终止
判断标准
至少有一条 START -> 某节点 的边，且每条路径最终可达 END
解决方向
add_edge(START, "first_node") 指定入口；用 add_edge("last_node", END) 显式收尾

InvalidUpdateError: 同步并发更新同一 key

典型表现
两个并行节点都写同一个无 reducer 字段时报 InvalidUpdateError
判断标准
被并发写的字段要么只由一个节点写，要么声明了 reducer
解决方向
给该字段加 reducer：Annotated[list, operator.add] 或 add_messages，详见 state-reducers 章

在节点里 mutate state

典型表现
checkpointer 回放 / time-travel 时状态对不上，stream 的 updates 不准
判断标准
节点不修改传入对象，所有变更走 return 的 partial dict
解决方向
把 state["x"] = v 改成 return {"x": v}，保持节点纯函数化

四个抽象到此闭环：State 存数据、Node 算更新、Edge 定走向、START/END 划边界，超步负责并行执行与合并。下一章 quickstart-state-graph 会把这套骨架扩展成一张有分支、有多节点的真实图，让你第一次跑通一个有意义的 StateGraph。