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 合并。
例：
defreducer(a:list,b:int):returna+[b]classState(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,# 节点专用输入schemaretry_policy=None,cache_policy=None,destinations=None)常用用法
1. 最简单方式
defmy_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"})示例
defrouter(state):return"go"ifstate["x"]>10else"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（异步）
awaitcompiled.ainvoke(...)stream（流式执行）
forstepincompiled.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. 一个最佳范例
fromtyping_extensionsimportTypedDict,Annotatedfromlanggraph.graphimportStateGraphfromlanggraph.runtimeimportRuntimedefreducer(lst,v):returnlst+[v]classState(TypedDict):x:Annotated[list,reducer]score:intclassContext(TypedDict):r:floatgraph=StateGraph(State,context_schema=Context)defstep1(state,runtime:Runtime[Context]):x_last=state["x"][-1]r=runtime.context["r"]return{"x":x_last*r}defjudge(state):return"OK"ifstate["x"][-1]>5else"FAIL"defok(state):return{"score":100}deffail(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}))
)