LangGraph 02：StateGraph 基础

发表于 更新于 分类于
用 TypedDict、dataclass、Pydantic 定义 State，理解 Annotated 与 operator.add，以及自定义 reducer 与图可视化。

前言

《LangGraph 01：生态与环境》已经跑通「单节点问候」图。
真正写业务图时，先要定清楚：状态长什么样、节点返回的局部更新如何合并进全局状态。
本文不调用 LLM，只把 State 三种写法、Annotated + reducer、自定义合并函数 与 图可视化 讲清楚。
运行环境仍用上一篇的 uv 工程即可。

依赖

沿用系列工程依赖；本篇无需新增包。
若尚未初始化，可按 《LangGraph 01：生态与环境》 安装 langgraph 与 rich。

1
uv add langgraph rich

Pydantic 状态示例需要 pydantic（多数环境随生态间接安装；缺失时再执行 uv add pydantic）。

实现

TypedDict 状态

TypedDict 是最常见的图状态定义：轻量、字段清晰，节点通常返回「只含变更字段」的 dict。
下面用 logs 配合 operator.add 做列表追加，cur_id 无 reducer，后写覆盖前写。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from operator import add
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


class OverAllState(TypedDict):
    logs: Annotated[list[str], add]
    cur_id: str


def node_1(state: OverAllState) -> dict:
    return {
        "logs": ["node_1 执行完成"],
        "cur_id": f"{state['cur_id']}, node_1",
    }


def node_2(state: OverAllState) -> dict:
    return {
        "logs": ["node_2 执行完成"],
        "cur_id": f"{state['cur_id']}, node_2",
    }


builder = StateGraph(OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()
result = graph.invoke({"cur_id": "start", "logs": []})
rprint(result)

logs 会变成 ['node_1 执行完成', 'node_2 执行完成']（若初始带了条目会一并保留）。
cur_id 最终类似 start, node_1, node_2。

dataclass 状态

需要默认值或属性访问时，可用 @dataclass 作为 state_schema。
invoke 可传入 dataclass 实例；节点仍可返回 dict 或同类型实例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from dataclasses import dataclass
from operator import add
from typing import Annotated

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


@dataclass
class OverAllState:
    logs: Annotated[list[str], add]
    cur_id: str


def node_1(state: OverAllState) -> dict:
    return {
        "logs": ["node_1 执行完成"],
        "cur_id": f"{state.cur_id}, node_1",
    }


def node_2(state: OverAllState) -> dict:
    return {
        "logs": ["node_2 执行完成"],
        "cur_id": f"{state.cur_id}, node_2",
    }


builder = StateGraph(OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()
result = graph.invoke(OverAllState(logs=[], cur_id="start"))
rprint(result)

访问字段时用 state.cur_id，与 TypedDict 的 state["cur_id"] 不同。
合并语义与上一节相同，仍由 Annotated[..., add] 决定。

Pydantic 状态

需要运行时校验、嵌套模型时，可用 BaseModel 作状态。
性能通常弱于 TypedDict / dataclass；简单图不必强行上 Pydantic。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from operator import add
from typing import Annotated

from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel
from rich import print as rprint


class OverAllState(BaseModel):
    logs: Annotated[list[str], add] = []
    cur_id: str


def node_1(state: OverAllState) -> dict:
    return {
        "logs": ["node_1 执行完成"],
        "cur_id": f"{state.cur_id}, node_1",
    }


def node_2(state: OverAllState) -> dict:
    # 只更新部分字段也可以
    return {"cur_id": f"{state.cur_id}, node_2"}


builder = StateGraph(OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()
result = graph.invoke({"cur_id": "start"})
rprint(result)

节点未返回的带 reducer 字段会按规则保留；node_2 未写 logs 时，不会清空已有日志。

Annotated 与 add

没有 reducer 时，同名字段以后写覆盖先写。
列表、计数等「要累积」的通道，用 Annotated[类型, reducer] 声明合并方式。

operator.add 对 list 做拼接、对 int 做相加，是最常用的内置选择。
并行扇出时，若多个节点同时写同一通道，必须有合适的 reducer，否则可能触发非法更新错误。

自定义 reducer

reducer 签名可理解为：left 是通道当前值，right 是本节点（或本步）提交的更新，返回合并后的新值。
下面用纯函数演示合并逻辑，再挂到状态字段上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from rich import print as rprint


def my_reducer(left: list[str], right: list[str]) -> list[str]:
    # left: 合并前已有值；right: 本节点返回值
    return left + right


class OverAllState(TypedDict):
    logs: Annotated[list[str], my_reducer]
    cur_id: str


def node_1(state: OverAllState) -> dict:
    return {"logs": ["node_1"], "cur_id": "n1"}


def node_2(state: OverAllState) -> dict:
    return {"logs": ["node_2"], "cur_id": "n2"}


builder = StateGraph(OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)

graph = builder.compile()
rprint(graph.invoke({"logs": ["start"], "cur_id": "s"}))

自定义 reducer 适合去重、限长、按 id 合并等 operator.add 表达不了的规则。
对话消息场景应优先用内置 add_messages（见 《LangGraph 03：消息与多 Schema》），不要自己用 add 硬拼 BaseMessage。

可视化

编译后可通过 get_graph() 导出结构，便于对照边是否连对。

1
2
3
4
5
# ASCII（终端可读）
graph.get_graph().print_ascii()

# Mermaid 文本（可粘贴到支持 Mermaid 的编辑器）
print(graph.get_graph().draw_mermaid())

Notebook 里还可用 draw_mermaid_png() 生成图片；脚本环境有时依赖额外渲染包，失败时用 Mermaid 文本即可。

直接查看图

1
2
3
from IPython.display import display

display(graph)

验证

依次运行 TypedDict 与自定义 reducer 两个脚本。
确认 logs 为追加而非覆盖，且 cur_id 为最后一次写入。
再用 print_ascii / draw_mermaid 核对 START → node_1 → node_2 → END。

总结

  1. State 可用 TypedDict、dataclass、Pydantic；日常优先 TypedDict。
  2. 无 reducer 则覆盖；要累积就写 Annotated[..., reducer]。
  3. 自定义 reducer 的 (left, right) -> merged 语义要稳定、可幂等思考。
  4. 下一篇进入消息列表与多 Schema： 《LangGraph 03：消息与多 Schema》。