1. StateGraph 是什么？

StateGraph 是一个 图构建器，它本身不执行，只负责：

• 定义状态结构
• 定义节点逻辑
• 定义节点之间的顺序 / 条件分支 / 路由
• 定义上下文 Context
• 定义输入输出 Schema

⚠️ 注意：StateGraph 不能直接运行，必须先编译：

compiled = graph.compile()

编译后得到 CompiledStateGraph（可执行版本），可使用：

• .invoke()
• .ainvoke()
• .stream()
• .astream()

2. StateGraph 的核心概念

概念	说明
State	所有节点共享的可变状态（TypedDict）
Context	节点运行时可用的只读上下文，例如 user_id、模型句柄等
Node	一个函数，输入 State，返回部分 State
Reducer	合并不同节点输出的函数
Edge	连接节点执行顺序
Conditional Edge	基于某个函数决定下一个节点
Sequence	快速链接一串节点

3. 初始化 StateGraph

graph = StateGraph(
    state_schema=State,
    context_schema=Context,
    input_schema=Input,
    output_schema=Output
)

参数说明

参数	作用
state_schema（必填）	定义 State 的结构与 reducer（TypedDict + Annotated）
context_schema	定义 Context 结构（运行时只读）
input_schema	定义图的输入结构
output_schema	定义最终返回的输出结构

⚠️ config_schema 已废弃，改用 context_schema

4. State 的 Reducer

StateGraph 允许多个节点写同一个 key。
写冲突时，使用 Reducer 合并。

例：

def reducer(a: list, b: int):
    return a + [b]

class State(TypedDict):
    x: Annotated[list, reducer]

节点 A 返回 {"x": 1}
节点 B 返回 {"x": 2}
→ 自动合并成 {"x": [1, 2]}

5. 添加节点 add_node()

节点是图的最小执行单位，接受 State 返回部分 State。

方法签名简化版

add_node(
    node,               # 名称或函数
    action=None,        # 当 node 是字符串时使用
    defer=False,        # 延后执行（收尾逻辑）
    metadata=None,      # 节点元信息
    input_schema=None,  # 节点专用输入schema
    retry_policy=None,
    cache_policy=None,
    destinations=None
)

常用用法

1. 最简单方式

def my_node(state):
    return {"x": state["x"] + 1}

g.add_node(my_node)

2. 自定义节点名称

g.add_node("calc", my_node)

3. 使用 defer（最后执行）

g.add_node("cleanup", cleanup_fn, defer=True)

4. 添加节点后必须连接

g.add_edge(START, "calc")

6. 添加边 add_edge()

add_edge(start_key, end_key)

表示：当 start_key 完成后执行 end_key

单起点

graph.add_edge("A", "B")

多起点（等待全部完成）

graph.add_edge(["A", "B"], "C")

表示：等 A 和 B 都执行完 → C 才执行。

7. 条件边 add_conditional_edges()

用于“if-else”、“switch-case”。

add_conditional_edges(
    source="A",
    path=path_fn,
    path_map={"yes": "Node1", "no": "Node2"}
)

示例

def router(state):
    return "go" if state["x"] > 10 else "stop"

graph.add_conditional_edges(
    "Check",
    router,
    path_map={
        "go": "NextStep",
        "stop": "__end__"
    }
)

8. 快速添加序列 add_sequence()

等价于：

A → B → C

graph.add_sequence([nodeA, nodeB, nodeC])

也可命名：

graph.add_sequence([
    ("start", nodeA),
    ("compute", nodeB),
    ("finish", nodeC)
])

9. 编译 compile()

编译后才能运行：

compiled = graph.compile()

支持参数

参数	作用
checkpointer	自动保存中间状态（可暂停/恢复）
cache	节点级缓存
interrupt_before	某节点前暂停
interrupt_after	某节点后暂停
debug	打印调试信息
name	给编译后的 graph 命名

10. 编译后使用

invoke（同步）

compiled.invoke({"x": 1}, context={"r": 3})

ainvoke（异步）

await compiled.ainvoke(...)

stream（流式执行）

for step in compiled.stream(...):
    print(step)

11. 总结

方法	用途	示例
add_node	添加节点	g.add_node(“A”, fn)
add_edge	添加顺序边	g.add_edge(“A”, “B”)
add_conditional_edges	条件跳转	if A → B 或 C
add_sequence	快速构建链式流程	g.add_sequence([A, B, C])
compile	图编译为可执行图	compiled = g.compile()

12. 一个最佳范例

from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime

def reducer(lst, v):
    return lst + [v]

class State(TypedDict):
    x: Annotated[list, reducer]
    score: int

class Context(TypedDict):
    r: float

graph = StateGraph(State, context_schema=Context)

def step1(state, runtime: Runtime[Context]):
    x_last = state["x"][-1]
    r = runtime.context["r"]
    return {"x": x_last * r}

def judge(state):
    return "OK" if state["x"][-1] > 5 else "FAIL"

def ok(state):
    return {"score": 100}

def fail(state):
    return {"score": 0}

graph.add_node("step1", step1)
graph.add_node("OK", ok)
graph.add_node("FAIL", fail)

graph.set_entry_point("step1")
graph.add_conditional_edges(
    "step1",
    judge,
    {                 
        "OK": "OK",
        "FAIL": "FAIL",
    }
)

compiled = graph.compile()
print(compiled.invoke({"x": [2], "score": 0}, context={"r": 3}))