StateGraph 核心模型

StateGraph 是 LangGraph 最核心的抽象，它定义了一个有状态的状态机，用于编排复杂的 Agent 工作流。

为什么需要 StateGraph

传统的 LLM Chain 是线性的（A → B → C），无法处理：

• 条件分支：根据 LLM 输出决定下一步
• 循环：工具调用需要反复执行直到完成
• 状态持久化：跨步骤共享上下文
• 并行执行：多个节点同时运行

StateGraph 通过图结构解决了这些问题。

基本概念

State（状态）

State 是整个工作流的"上下文对象"。每个节点读取当前状态，并返回局部更新。

from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    question: str
    documents: List[str]
    answer: str
    messages: Annotated[list, operator.add]  # reducer

State 的两种定义方式

1. TypedDict（推荐）：

from typing import TypedDict

class State(TypedDict):
    foo: str
    bar: int

2. dataclass：

from dataclasses import dataclass, field
from typing import list

@dataclass
class State:
    foo: str = ""
    bar: int = 0

Reducer（归约器）

Reducer 定义了当多个节点写入同一个 state 字段时如何合并。最常见的 reducer 是 operator.add，用于将消息列表追加到一起：

from typing import Annotated
import operator

class State(TypedDict):
    # 使用 operator.add 作为 reducer
    # 多个节点返回的 messages 会被自动合并
    messages: Annotated[list, operator.add]

如果为字段指定了 reducer，节点必须返回该字段的值；如果字段没有 reducer，节点的返回值会直接覆盖之前的值。

Node（节点）

节点就是普通的 Python 函数，接收 state 返回更新：

def retrieve_node(state: AgentState):
    docs = ["文档1", "文档2"]
    return {"documents": docs}

Edge（边）

边定义了执行顺序：

# 普通边：无条件执行
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", END)

# 条件边：根据条件路由
graph.add_conditional_edges(
    "llm_node",
    should_continue,
    ["tool_node", END]
)

完整示例

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, List

class State(TypedDict):
    input: str
    result: str

def process(state: State):
    return {"result": f"处理: {state['input']}"}

# 构建图
builder = StateGraph(State)
builder.add_node("process", process)
builder.add_edge(START, "process")
builder.add_edge("process", END)

# 编译
graph = builder.compile()

# 执行
result = graph.invoke({"input": "hello", "result": ""})
print(result)  # {'input': 'hello', 'result': '处理: hello'}

执行流程

1. 编译（compile）：检查图的拓扑结构是否合法
2. 调用（invoke）：传入初始 state
3. 遍历：从 START 开始，按边定义顺序执行节点
4. 合并：每个节点的返回值通过 reducer 合并到 state
5. 结束：到达 END 后返回最终 state

MessagesState

LangGraph 内置了一个 MessagesState，专门用于聊天场景：

from langgraph.graph import MessagesState

class MyState(MessagesState):
    # messages 字段已经定义好了
    extra_field: str

MessagesState 默认包含一个 messages: Annotated[list, add_messages] 字段，其中 add_messages 是一个内置 reducer，能自动处理消息的去重和追加。

最佳实践

1. State 保持扁平：不要嵌套太深，尽量用 TypedDict
2. 善用 Reducer：如果需要多个节点写入同一字段，确保定义了 reducer
3. 节点函数保持纯：不修改输入的 state，返回新的更新字典
4. 小粒度节点：每个节点只做一件事，方便复用和调试

参考

• 官方文档：Graph API overview
• API 参考：StateGraph