StateGraph 状态图

前面几篇你认识了 State、Node、Edge、ConditionalEdge——这些是"零件"。本篇讲把它们装起来的容器：StateGraph。它是 LangGraph 的核心类，理解它就理解了"图"这一抽象。

一、StateGraph 是什么

StateGraph(State) 是状态图的容器类，你往里加节点、加边，最后 compile() 得到可执行版本。

python

from langgraph.graph import StateGraph, START, END

graph = StateGraph(MyState)
graph.add_node("a", fn_a)
graph.add_edge(START, "a")
graph.add_edge("a", END)

app = graph.compile()   # 编译产物是 CompiledGraph
result = app.invoke(initial_state)

它对应 LangGraph4j 的 StateGraph.of(State.class)，概念完全一致。

二、三大方法

StateGraph 提供三大核心方法：

1. add_node

python

graph.add_node(name: str, action: Callable)

注册一个节点。name 是字符串标识（后面连边用），action 是节点函数。

python

graph.add_node("retrieve", retrieve_fn)
graph.add_node("generate", generate_fn)

2. add_edge

python

graph.add_edge(source: str, target: str)

连一条普通边。source/target 是节点名或 START/END。

python

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

3. add_conditional_edges

python

graph.add_conditional_edges(
    source: str,
    router: Callable,
    path_map: dict | list | None = None,
)

加一条条件边。详见条件边 ConditionalEdge。

python

graph.add_conditional_edges(
    "decide",
    route_fn,
    {"yes": "branch_a", "no": "branch_b"},
)

这三个方法 + compile() 几乎覆盖了所有构图需求。更高级的能力（子图、人机交互）后续模块讲。

三、compile() 参数

compile() 把"声明式图"变成"可执行版本"，会校验结构并生成 CompiledGraph。

python

app = graph.compile(
    checkpointer=...,           # 检查点器（持久化 + 记忆）
    interrupt_before=[...],     # 在指定节点之前暂停
    interrupt_after=[...],      # 在指定节点之后暂停
)

参数详解：

checkpointer

让图"有记忆"。每次节点执行后保存状态快照，配合 thread_id 可以中断、恢复、回放。详见编译与运行。

python

from langgraph.checkpoint.memory import MemorySaver
app = graph.compile(checkpointer=MemorySaver())

interrupt_before / interrupt_after

人机交互（HITL）用。图执行到指定节点前/后暂停，等你审核后用 Command 恢复。

python

app = graph.compile(
    interrupt_before=["approve"],   # 执行 approve 前暂停
)

LangGraph 0.2+ 还推荐用 interrupt() 函数在节点内部主动暂停，更灵活。见进阶特性。

编译会校验什么

所有节点都被加边连到了图里（孤立节点会警告）。
从 START 能到达所有节点（不可达节点会警告）。
所有节点都能到达 END（"死端"会警告）。
条件边的 path_map 引用的节点都存在。

校验不通过会抛 ValueError，把错误修了再编译。

四、为什么需要 compile

你可能会问：为什么不能直接 graph.invoke(...) 跑声明式图？多一步编译有啥用？

mermaid

graph LR
    A[声明式图<br/>StateGraph] -->|compile| B[可执行版本<br/>CompiledGraph]
    B -->|invoke/stream| C[运行结果]

编译做的事：

校验：检查图结构合法（无孤立节点、无死循环等）。
生成执行计划：把声明式的边和节点编译成可调度结构。
注入运行时能力：把 checkpointer、interrupt 等运行时配置打包进可执行版本。
缓存优化：编译后的版本可复用，多次 invoke 不用重编译。

CompiledGraph 是不可变的（配置好后不再变），可以并发调用。

五、完整示例：多节点 + 条件边

下面是一个"翻译流程"图：判断语言 → 翻译 → 校对，校对不通过回到翻译：

python

# stategraph_demo.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    text: str
    target_lang: str
    translated: str
    quality_ok: bool
    attempts: Annotated[int, add]


def detect_lang(state: State) -> dict:
    """假装检测目标语言（实际可让 LLM 检测）。"""
    text = state["text"]
    target = state["target_lang"] or "en"
    return {"target_lang": target}


def translate(state: State) -> dict:
    """假装翻译。"""
    text = state["text"]
    target = state["target_lang"]
    return {"translated": f"[{target}] {text}"}


def review(state: State) -> dict:
    """假装质量检查：尝试次数 < 2 时算不通过。"""
    ok = state["attempts"] >= 2
    return {"quality_ok": ok}


def route_review(state: State) -> str:
    """校对通过则结束，否则回到翻译。"""
    if state["quality_ok"]:
        return "end"
    return "translate"


g = StateGraph(State)
g.add_node("detect", detect_lang)
g.add_node("translate", translate)
g.add_node("review", review)

g.add_edge(START, "detect")
g.add_edge("detect", "translate")
g.add_edge("translate", "review")
g.add_conditional_edges(
    "review",
    route_review,
    {"end": END, "translate": "translate"},
)
app = g.compile()

# 用 stream 看每一步
for event in app.stream({
    "text": "你好世界",
    "target_lang": "en",
    "translated": "",
    "quality_ok": False,
    "attempts": 0,
}):
    print(event)

输出：

text

{'detect': {'target_lang': 'en'}}
{'translate': {'translated': '[en] 你好世界'}}
{'review': {'quality_ok': False}}
{'translate': {'translated': '[en] 你好世界'}}
{'review': {'quality_ok': True}}

图结构：

mermaid

graph LR
    S([START]) --> D[detect 检测语言]
    D --> T[translate 翻译]
    T --> R[review 校对]
    R --> C{通过?}
    C -->|否| T
    C -->|是| E([END])

这是个典型"翻译-校对-重试"循环，用条件边实现。

六、可视化图

python

print(app.get_graph().draw_mermaid())

输出可以直接贴到 mermaid 渲染器。建议每个图写完都打印一次检查结构。

如果想画得更花，可以：

python

# 需要 pip install grandalf（不是 graphviz）
app.get_graph().draw_mermaid_png(save_path="graph.png")

七、子图：把图当节点用

StateGraph 编译后的 CompiledGraph 可以作为另一个图的节点，实现模块化：

python

sub_app = sub_graph.compile()

def big_node(state: BigState) -> dict:
    # 调子图
    sub_result = sub_app.invoke({"some_field": state["x"]})
    return {"y": sub_result["out"]}

big_graph.add_node("sub", big_node)

更地道的做法是直接把编译好的子图当节点传入 add_node（LangGraph 会自动适配状态）。子图详见进阶特性模块。

八、与 LangGraph4j 对比

维度	Python	Java
容器类	`StateGraph(State)`	`StateGraph.of(State.class)`
加节点	`add_node("n", fn)`	`addNode("n", fn)`
加边	`add_edge("a", "b")`	`addEdge("a", "b")`
条件边	`add_conditional_edges(...)`	`addConditionalEdges(...)`
编译	`graph.compile(checkpointer=...)`	`graph.compile(checkpointer, ...)`
编译产物	`CompiledGraph`	`CompiledGraph`
子图	直接 add_node	直接 addNode
入口出口	`START` / `END` 常量	`START` / `END` 常量

API 几乎一一对应，迁移成本主要在节点写法（Python 函数 vs Java 类）。

九、常见踩坑

不调 compile 直接 invoke：graph.invoke(...) 会报错，StateGraph 本身没有 invoke 方法。必须先 app = graph.compile()。
编译前调用 add_node：节点和边必须在 compile() 之前加完。编译后再 add 不会生效（除非重新编译）。
重复 compile：每次 compile 都生成新的 CompiledGraph。如果想配 checkpointer 又想 HITL，一次性传齐参数：
python
```
app = graph.compile(checkpointer=MemorySaver(), interrupt_before=["approve"])
```
孤立节点：注册了节点但没加边——编译会警告。要么删要么连边。
不可达节点：从 START 走不到的节点——警告。检查入口边。
死端节点：节点没出边也没连 END——警告。
同名节点重复注册：add_node("x", fn1) 后又 add_node("x", fn2)——后者会覆盖前者，但容易出 bug。改名或检查逻辑。
checkpointer 配了但 invoke 没传 thread_id：状态不会持久化。invoke 时必须传 config={"configurable": {"thread_id": "xxx"}}。详见编译与运行。
interrupt_before 节点名拼错：编译时不一定报错，运行时跳过不存在的节点会失效。
图太复杂难维护：节点超过 10 个考虑拆子图。一个函数文件超过 300 行该拆模块。

十、小结

StateGraph(State) 是图容器，三大方法：add_node / add_edge / add_conditional_edges。
compile() 校验结构 + 生成可执行版本 CompiledGraph，可注入 checkpointer / interrupt_before / interrupt_after。
编译后不可变，可复用、可并发。
用 app.get_graph().draw_mermaid() 可视化检查。
子图可作为节点复用，实现模块化。
与 LangGraph4j 概念一一对应。

下一篇讲 Reducer 归约器——决定字段如何被多个节点合并进化。

StateGraph 状态图 ​

一、StateGraph 是什么 ​

二、三大方法 ​

1. add_node ​

2. add_edge ​

3. add_conditional_edges ​

三、compile() 参数 ​

checkpointer ​

interrupt_before / interrupt_after ​

编译会校验什么 ​

四、为什么需要 compile ​

五、完整示例：多节点 + 条件边 ​

六、可视化图 ​

七、子图：把图当节点用 ​

八、与 LangGraph4j 对比 ​

九、常见踩坑 ​

十、小结 ​

StateGraph 状态图

一、StateGraph 是什么

二、三大方法

1. add_node

2. add_edge

3. add_conditional_edges

三、compile() 参数

checkpointer

interrupt_before / interrupt_after

编译会校验什么

四、为什么需要 compile

五、完整示例：多节点 + 条件边

六、可视化图

七、子图：把图当节点用

八、与 LangGraph4j 对比

九、常见踩坑

十、小结