StateGraph与节点

要把「预处理 → 业务逻辑 → 格式化输出」拆成三个可测步骤，线性脚本很快变成一团 if/else。StateGraph 把每一步写成节点，共享字段写在 State 里，控制流用边显式连接——图结构即文档。

1. 定位

维度	内容
角色	图构建入口：定义 State schema、注册节点、连边
输入 → 输出	`invoke(initial_state)` → 最终 `State` 快照
核心 API	`StateGraph`、`add_node`、`add_edge`、`compile`
依赖 LangChain	节点内可用 `BaseMessage`；本篇示例用纯 Python mock

2. 图拓扑

节点表

节点名	职责	读 State	写 State
`parse`	解析原始输入	`raw`	`parsed`
`process`	业务计算	`parsed`	`result`
`format`	格式化输出	`result`	`output`

边表

源	目标	类型
START	parse	固定
parse	process	固定
process	format	固定
format	END	固定

3. invoke 生命周期

1. 传入 initial_state（字段符合 State TypedDict）
2. 调度 START 的后继 → parse
3. parse 返回 {"parsed": ...}，合并进 state
4. 沿固定边依次执行 process → format
5. 到达 END，返回完整 state

Superstep：每个节点执行完毕、state 合并后，调度器再选下一批节点。本篇为纯线性图，每轮 superstep 仅跑一个节点。

4. 原理

4.1 StateGraph 与 State schema

StateGraph(State) 接受 TypedDict（或 Pydantic）类，约束节点可读写字段。节点函数签名 (state: State) -> dict，返回值是部分更新，不是全量 state。

4.2 节点即纯函数（推荐）

节点应尽量无副作用；副作用（IO、LLM）集中在一处，便于单测 mock。

4.3 compile 产物

builder.compile() 返回 CompiledGraph，提供 invoke / stream / get_state 等方法。

5. 最小可运行示例

from typing import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    raw: str
    parsed: int
    result: int
    output: str


def parse_node(state: State) -> dict:
    return {"parsed": int(state["raw"])}


def process_node(state: State) -> dict:
    return {"result": state["parsed"] * 2}


def format_node(state: State) -> dict:
    return {"output": f"结果={state['result']}"}


builder = StateGraph(State)
builder.add_node("parse", parse_node)      # 拓扑 §2 parse
builder.add_node("process", process_node)
builder.add_node("format", format_node)
builder.add_edge(START, "parse")
builder.add_edge("parse", "process")
builder.add_edge("process", "format")
builder.add_edge("format", END)

graph = builder.compile()
out = graph.invoke({"raw": "21", "parsed": 0, "result": 0, "output": ""})
print(out["output"])  # 结果=42

6. 执行追踪

输入：{"raw": "21", "parsed": 0, "result": 0, "output": ""}

步骤	节点	关键 state
0	初始	`raw="21"`, 其余默认
1	parse 后	`parsed=21`
2	process 后	`result=42`
3	format 后	`output="结果=42"`

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
`StateGraph(state_schema)`	必填	约束字段与 reducer	TypedDict	字段尽量少
`add_node(name, fn)`	name 唯一	注册可调度单元	动词命名	一节点一职责
`add_edge(src, dst)`	—	固定后继	START/END 常量	先画边表再写代码
`compile()`	无参可编译	生成可执行图	开发先无 checkpointer	持久化另配
`invoke(input)`	dict	同步跑完全图	小图单测	生产考虑 async
`compile(debug=True)`	默认 False	打印调度细节	本地调试	生产关闭

7. 易踩坑

节点返回全量 state：只需返回变更字段；多余字段可能被当作覆盖写入。
节点名与字符串边不一致：add_edge("pars", ...) 拼写错误要到 invoke 才报错。
State 缺字段且无默认：invoke 输入应包含 schema 全部键，或给 Optional 默认值。

小结

StateGraph = State schema + 节点 + 边；compile → invoke 是固定套路。
节点返回 partial update；多节点写同一字段需 reducer（见状态专篇）。
写代码前先列节点表 + 边表，避免逻辑藏在函数嵌套里。