Skip to content

LangGraph State Schema 详细解读

📚 概述

本文档详细解读 LangGraph 中的 State Schema(状态模式)。状态是 LangGraph 的核心概念,它定义了图中数据的结构和类型。理解状态模式是掌握 LangGraph 的关键一步。

在本教程中,我们将深入探讨:

  • 什么是 State Schema
  • 三种定义状态的方式(TypedDict、Dataclass、Pydantic)
  • 每种方式的优缺点和适用场景
  • 实战代码详解

🎯 核心概念

什么是 State Schema?

在 LangGraph 中,State Schema 定义了:

  1. 数据结构:图中存储哪些数据
  2. 数据类型:每个字段的类型是什么
  3. 通信协议:节点之间如何传递和更新数据

可以把 State 想象成一个共享的数据容器,所有节点都可以读取和更新它。

核心特性

1. Channel(通道)概念

State 中的每个键(key)就是一个通道(channel)

State = {
    "name": "Alice",    # name 通道
    "mood": "happy"     # mood 通道
}
  • 每个节点可以读取任何通道的值
  • 每个节点可以更新一个或多个通道
  • 通道之间相互独立

2. 节点更新机制

python
def node_1(state):
    # 读取 name 通道
    current_name = state["name"]

    # 更新 name 通道(返回字典)
    return {"name": current_name + " is happy"}

关键点:

  • 输入:节点接收完整的 State
  • 输出:节点返回要更新的字段(部分更新)
  • 合并:LangGraph 自动将返回值合并到 State 中

🔧 三种定义方式

LangGraph 支持三种定义 State Schema 的方式,每种都有独特的优势。

方式 1: TypedDict ⭐

这是最常用、最简单的方式。

基础用法

python
from typing_extensions import TypedDict

class TypedDictState(TypedDict):
    foo: str
    bar: str

特点:

  • ✅ 语法简洁
  • ✅ IDE 支持好(自动补全、类型提示)
  • ✅ 性能最优(无运行时验证)
  • ❌ 不做运行时类型检查

进阶用法:Literal 类型

python
from typing import Literal

class TypedDictState(TypedDict):
    name: str
    mood: Literal["happy", "sad"]  # 限定只能是这两个值之一

Literal 详解:

Literal 是 Python 的类型提示工具,用于限制值的范围:

python
# ✅ 正确
state = {"name": "Alice", "mood": "happy"}

# ❌ IDE 会警告(但运行时不报错)
state = {"name": "Alice", "mood": "angry"}  # angry 不在允许的值中

在 LangGraph 中使用

python
from langgraph.graph import StateGraph, START, END

# 1. 定义状态
class TypedDictState(TypedDict):
    name: str
    mood: Literal["happy", "sad"]

# 2. 定义节点
def node_1(state):
    return {"name": state["name"] + " is ... "}

# 3. 创建图
builder = StateGraph(TypedDictState)  # 传入状态类
builder.add_node("node_1", node_1)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)

# 4. 编译和运行
graph = builder.compile()
result = graph.invoke({"name": "Lance"})  # 用字典初始化

关键点:

  • 使用 StateGraph(TypedDictState) 指定状态模式
  • 用字典调用 graph.invoke({"name": "Lance"})
  • 节点通过 state["key"] 访问字段

方式 2: Dataclass

Python 3.7+ 提供的数据类装饰器。

基础用法

python
from dataclasses import dataclass

@dataclass
class DataclassState:
    name: str
    mood: Literal["happy", "sad"]

与 TypedDict 的区别

特性TypedDictDataclass
访问方式state["name"]state.name
初始化字典类实例
性能更快稍慢
功能基础更多(默认值、方法等)

在 LangGraph 中使用

python
# 定义节点 - 注意访问方式改变
def node_1(state):
    # ✅ 使用点号访问(不是 state["name"])
    return {"name": state.name + " is ... "}

# 创建图
builder = StateGraph(DataclassState)
builder.add_node("node_1", node_1)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)

graph = builder.compile()

# 用 dataclass 实例初始化
result = graph.invoke(DataclassState(name="Lance", mood="sad"))

重要细节:

虽然 state 是 dataclass,但节点仍然返回字典

python
def node_1(state):
    # 输入:dataclass
    name = state.name  # 点号访问

    # 输出:字典!
    return {"name": name + " updated"}  # ✅ 正确
    # return DataclassState(...)  # ❌ 不需要

为什么可以这样?

LangGraph 内部将 state 的每个字段存储为独立的 channel。节点返回的字典只需要有匹配的键,就能更新对应的 channel。


方式 3: Pydantic BaseModel ⭐⭐

最强大的方式,提供运行时验证。

什么是 Pydantic?

Pydantic 是 Python 的数据验证库,核心类是 BaseModel

python
from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

# ✅ 自动类型转换
user = User(name="Alice", age="25")  # "25" 自动转为 25
print(user.age)  # 25 (int)

# ❌ 验证失败会抛出异常
user = User(name="Alice", age="invalid")  # ValidationError

TypedDict vs Pydantic 对比

python
# TypedDict - 不做运行时检查
class DataclassState:
    name: str
    mood: Literal["happy", "sad"]

# ❌ 这段代码可以运行(虽然 IDE 会警告)
state = DataclassState(name="Lance", mood="angry")  # 没有报错
python
# Pydantic - 运行时验证
from pydantic import BaseModel, field_validator

class PydanticState(BaseModel):
    name: str
    mood: str

    @field_validator('mood')
    @classmethod
    def validate_mood(cls, value):
        if value not in ["happy", "sad"]:
            raise ValueError("Mood must be 'happy' or 'sad'")
        return value

# ✅ 正确的值
state = PydanticState(name="Lance", mood="happy")  # OK

# ❌ 错误的值会抛出异常
state = PydanticState(name="Lance", mood="angry")  # ValidationError!

自定义验证器详解

python
from pydantic import BaseModel, field_validator, ValidationError

class PydanticState(BaseModel):
    name: str
    mood: str

    @field_validator('mood')  # 装饰器:指定要验证的字段
    @classmethod  # 必须是类方法
    def validate_mood(cls, value):
        # value 是传入的值
        if value not in ["happy", "sad"]:
            raise ValueError("Each mood must be either 'happy' or 'sad'")
        return value  # 返回验证后的值

# 测试验证
try:
    state = PydanticState(name="John", mood="mad")
except ValidationError as e:
    print("Validation Error:", e)
    # 输出:Validation Error: 1 validation error for PydanticState
    #      mood
    #        Input should be 'happy' or 'sad'

验证器的工作流程:

传入值 → field_validator → 检查逻辑 → 返回值或抛出异常
   ↓                                        ↓
"mad"                                   ValueError

                                    ValidationError

在 LangGraph 中使用

python
# 创建图(与之前完全相同)
builder = StateGraph(PydanticState)
builder.add_node("node_1", node_1)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", END)

graph = builder.compile()

# 用 Pydantic 实例初始化
result = graph.invoke(PydanticState(name="Lance", mood="sad"))

🎭 实战案例:情绪决策系统

让我们构建一个完整的系统,展示三种状态定义方式的使用。

系统架构

用户输入姓名

[node_1] 生成描述

 (随机决策)

    / \
   /   \
  ↓     ↓
[node_2] [node_3]
(happy)  (sad)
  ↓      ↓
  \      /
   \    /
    ↓  ↓
   返回结果

完整代码实现

python
import random
from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义状态
class TypedDictState(TypedDict):
    name: str
    mood: Literal["happy", "sad"]

# 2. 定义节点
def node_1(state):
    print("---Node 1---")
    return {"name": state["name"] + " is ... "}

def node_2(state):
    print("---Node 2---")
    return {"mood": "happy"}

def node_3(state):
    print("---Node 3---")
    return {"mood": "sad"}

# 3. 条件函数
def decide_mood(state) -> Literal["node_2", "node_3"]:
    """50/50 随机选择"""
    if random.random() < 0.5:
        return "node_2"  # 50% 概率选择 happy
    return "node_3"      # 50% 概率选择 sad

# 4. 构建图
builder = StateGraph(TypedDictState)

# 添加节点
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 添加边
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)  # 条件边
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

# 5. 编译和运行
graph = builder.compile()
result = graph.invoke({"name": "Lance"})

print(result)
# 输出:
# ---Node 1---
# ---Node 2---  (或 ---Node 3---)
# {'name': 'Lance is ... ', 'mood': 'happy'}  (或 'sad')

代码详解

1. 条件边的工作原理

python
builder.add_conditional_edges("node_1", decide_mood)
#                             ^^^^^^^^^  ^^^^^^^^^^^
#                             源节点      条件函数

执行流程:

1. 执行 node_1
2. 调用 decide_mood(state)
3. decide_mood 返回 "node_2" 或 "node_3"
4. 路由到对应的节点
5. 执行 node_2 或 node_3
6. 到达 END

2. Literal 在条件函数中的作用

python
def decide_mood(state) -> Literal["node_2", "node_3"]:
    #                    ^^^^^^^^^^^^^^^^^^^^^^^^^^
    #                    返回类型提示:只能是这两个字符串之一
    return "node_2"  # ✅ 正确
    return "node_5"  # ❌ IDE 会警告

这不仅是类型提示,还帮助 LangGraph 理解可能的路由目标。

3. 状态更新的执行顺序

python
初始状态: {"name": "Lance", "mood": None}

node_1: 返回 {"name": "Lance is ... "}
合并后: {"name": "Lance is ... ", "mood": None}

node_2: 返回 {"mood": "happy"}
合并后: {"name": "Lance is ... ", "mood": "happy"}  ← 最终结果

🎓 核心知识点总结

LangGraph 特有概念

1. State Schema 的作用

作用说明示例
定义数据结构指定有哪些字段name: str, mood: str
类型提示IDE 自动补全输入 state. 自动提示字段
验证(可选)Pydantic 运行时检查阻止无效数据
节点通信统一的数据接口所有节点使用相同结构

2. Channel(通道)模型

python
State = {
    "name": ...,  # name 通道
    "mood": ...   # mood 通道
}

# 节点可以更新部分通道
def node(state):
    return {"name": "new_value"}  # 只更新 name 通道

关键特性:

  • 每个字段是独立的通道
  • 节点可以只更新部分通道
  • 未返回的通道保持不变

3. 三种状态方式对比

特性TypedDictDataclassPydantic
语法复杂度⭐ 简单⭐⭐ 中等⭐⭐ 中等
访问方式state["key"]state.keystate.key
运行时验证❌ 无❌ 无✅ 有
性能最快稍慢(有验证)
适用场景简单类型需要默认值/方法需要严格验证
初始化方式字典类实例类实例

选择建议:

python
# 简单项目 → TypedDict
class State(TypedDict):
    message: str

# 需要默认值 → Dataclass
@dataclass
class State:
    message: str = "default"

# 需要验证 → Pydantic
class State(BaseModel):
    message: str

    @field_validator('message')
    @classmethod
    def validate_message(cls, v):
        if len(v) < 1:
            raise ValueError("Message cannot be empty")
        return v

Python 特有知识点

1. TypedDict 详解

python
from typing_extensions import TypedDict

# 基础用法
class State(TypedDict):
    name: str
    age: int

# 可选字段
from typing import Optional

class State(TypedDict):
    name: str
    age: Optional[int]  # 可以是 int 或 None

# total=False 使所有字段可选
class State(TypedDict, total=False):
    name: str  # 可选
    age: int   # 可选

2. Literal 类型详解

python
from typing import Literal

# 限制字符串值
def set_mood(mood: Literal["happy", "sad"]):
    ...

set_mood("happy")  # ✅
set_mood("angry")  # ❌ IDE 警告

# 限制多种类型
def process(value: Literal[1, 2, "auto"]):
    ...

process(1)      # ✅
process("auto") # ✅
process(3)      # ❌ IDE 警告

3. Pydantic Validator 详解

python
from pydantic import BaseModel, field_validator, validator

class State(BaseModel):
    name: str
    age: int

    # 方式 1: field_validator (推荐)
    @field_validator('age')
    @classmethod
    def validate_age(cls, v):
        if v < 0:
            raise ValueError('Age must be positive')
        return v

    # 方式 2: 多字段验证
    @field_validator('name', 'age')
    @classmethod
    def validate_both(cls, v, info):
        # info.field_name 可以知道是哪个字段
        if info.field_name == 'name' and len(v) < 1:
            raise ValueError('Name cannot be empty')
        return v

4. Dataclass 进阶用法

python
from dataclasses import dataclass, field

@dataclass
class State:
    name: str
    mood: str = "neutral"  # 默认值

    # 列表字段需要用 field(default_factory)
    tags: list = field(default_factory=list)

    # 计算属性
    @property
    def display_name(self):
        return f"{self.name} ({self.mood})"

state = State(name="Alice")
print(state.mood)          # "neutral"
print(state.display_name)  # "Alice (neutral)"

💡 最佳实践

1. 何时使用哪种状态定义?

✅ TypedDict - 大多数情况

python
# 推荐用于:
# - 简单的数据结构
# - 性能敏感的应用
# - 不需要运行时验证的场景

class State(TypedDict):
    messages: list
    user_id: str

✅ Dataclass - 需要默认值或方法

python
# 推荐用于:
# - 需要默认值
# - 需要计算属性
# - 面向对象设计

@dataclass
class State:
    messages: list = field(default_factory=list)
    retries: int = 3

    def should_retry(self):
        return self.retries > 0

✅ Pydantic - 需要严格验证

python
# 推荐用于:
# - 外部输入(API、用户输入)
# - 关键数据(配置、凭证)
# - 需要复杂验证逻辑

class State(BaseModel):
    api_key: str
    max_retries: int

    @field_validator('api_key')
    @classmethod
    def validate_key(cls, v):
        if not v.startswith('sk-'):
            raise ValueError('Invalid API key format')
        return v

    @field_validator('max_retries')
    @classmethod
    def validate_retries(cls, v):
        if v < 1 or v > 10:
            raise ValueError('Retries must be between 1 and 10')
        return v

2. 状态设计原则

原则 1: 最小化状态

python
# ❌ 不好 - 包含冗余数据
class State(TypedDict):
    user_input: str
    user_input_lower: str  # 可以从 user_input 计算
    user_input_length: int  # 可以从 user_input 计算

# ✅ 好 - 只存储必需数据
class State(TypedDict):
    user_input: str

def node(state):
    # 需要时再计算
    lower = state["user_input"].lower()
    length = len(state["user_input"])

原则 2: 使用明确的类型

python
# ❌ 不好 - 类型模糊
class State(TypedDict):
    data: dict  # 什么 dict?

# ✅ 好 - 类型明确
from typing import List, Dict

class Message(TypedDict):
    role: str
    content: str

class State(TypedDict):
    messages: List[Message]
    user_info: Dict[str, str]

原则 3: 分组相关字段

python
# ❌ 不好 - 平铺所有字段
class State(TypedDict):
    user_name: str
    user_age: int
    user_email: str
    config_api_key: str
    config_timeout: int

# ✅ 好 - 使用嵌套结构
class UserInfo(TypedDict):
    name: str
    age: int
    email: str

class Config(TypedDict):
    api_key: str
    timeout: int

class State(TypedDict):
    user: UserInfo
    config: Config

3. 常见陷阱

陷阱 1: 修改可变对象

python
# ❌ 错误 - 直接修改 state
def node(state):
    state["messages"].append("new message")  # 危险!
    return state

# ✅ 正确 - 返回新对象
def node(state):
    new_messages = state["messages"] + ["new message"]
    return {"messages": new_messages}

陷阱 2: 忘记初始化

python
# ❌ 错误 - 字段未初始化
class State(TypedDict):
    messages: list

graph.invoke({})  # 报错!messages 未定义

# ✅ 正确 - 提供初始值
graph.invoke({"messages": []})

# 或使用 Dataclass 提供默认值
@dataclass
class State:
    messages: list = field(default_factory=list)

graph.invoke(State())  # OK

陷阱 3: Pydantic 验证性能

python
# ❌ 性能问题 - 复杂验证
class State(BaseModel):
    data: list

    @field_validator('data')
    @classmethod
    def validate_data(cls, v):
        # 每次更新都运行!
        for item in v:
            complex_validation(item)  # 慢!
        return v

# ✅ 优化 - 只在必要时验证
class State(TypedDict):
    data: list

def validate_once(state):
    # 只在特定节点验证一次
    for item in state["data"]:
        complex_validation(item)
    return state

🚀 进阶技巧

1. 混合使用多种状态

python
# 全局状态用 TypedDict(性能)
class OverallState(TypedDict):
    messages: list
    user_id: str

# 特定节点用 Pydantic(验证)
class ValidatedInput(BaseModel):
    query: str

    @field_validator('query')
    @classmethod
    def validate_query(cls, v):
        if len(v) < 3:
            raise ValueError("Query too short")
        return v

def validation_node(state: OverallState):
    # 验证输入
    validated = ValidatedInput(query=state["messages"][-1])
    return {"messages": state["messages"]}

2. 动态状态字段

python
# 使用 Dict 存储动态数据
class State(TypedDict):
    metadata: Dict[str, Any]  # 动态字段

def node(state):
    # 可以动态添加任意键
    metadata = state.get("metadata", {})
    metadata["timestamp"] = time.time()
    metadata["custom_field"] = "value"
    return {"metadata": metadata}

3. 状态版本管理

python
# 添加版本字段跟踪状态变化
class State(TypedDict):
    data: str
    version: int

def node_1(state):
    return {
        "data": state["data"] + " updated",
        "version": state["version"] + 1
    }

# 可以根据 version 做条件判断
def should_continue(state):
    return "continue" if state["version"] < 5 else "stop"

📊 三种方式对比总结

功能对比表

功能TypedDictDataclassPydantic
类型提示
IDE 支持
运行时验证
默认值
自定义方法
访问方式["key"].key.key
初始化字典实例实例
性能最快中等
学习曲线

使用场景总结

python
# 🎯 选择决策树

# 1. 需要运行时验证?
#    是 → Pydantic
#    否 → 继续

# 2. 需要默认值或方法?
#    是 → Dataclass
#    否 → 继续

# 3. 追求最佳性能?
#    是 → TypedDict

实际项目建议

小型项目(< 10 个节点):

python
# 全用 TypedDict
class State(TypedDict):
    messages: list
    config: dict

中型项目(10-50 个节点):

python
# 主状态用 TypedDict,配置用 Pydantic
class State(TypedDict):
    messages: list
    config: AppConfig  # Pydantic model

class AppConfig(BaseModel):
    api_key: str
    timeout: int

大型项目(> 50 个节点):

python
# 分层设计
class BaseState(TypedDict):
    # 核心字段
    messages: list

class ValidatedState(BaseModel):
    # 需验证的字段
    user_input: str

@dataclass
class CachedState:
    # 需默认值的字段
    cache: dict = field(default_factory=dict)

🔍 常见问题

Q1: TypedDict 的类型提示不起作用怎么办?

原因: TypedDict 只提供静态类型检查,不做运行时验证。

解决方案:

  1. 使用 IDE(VSCode、PyCharm)进行静态检查
  2. 使用 mypy 进行类型检查:
    bash
    pip install mypy
    mypy your_script.py
  3. 如需运行时验证,改用 Pydantic

Q2: 为什么 Dataclass 节点还要返回字典?

原因: LangGraph 将状态按字段(channel)存储,节点返回的字典会自动映射到对应的 channel。

示例:

python
@dataclass
class State:
    name: str
    age: int

def node(state):
    # state 是 dataclass,访问用点号
    updated_name = state.name + " Smith"

    # 返回字典,LangGraph 会自动更新对应 channel
    return {"name": updated_name}  # ✅ 正确

Q3: Pydantic 验证失败会导致图崩溃吗?

是的。 Pydantic 验证失败会抛出 ValidationError,导致图执行中断。

解决方案:

python
def node(state):
    try:
        validated = MyModel(**state)
        return {"data": validated.data}
    except ValidationError as e:
        # 处理错误
        return {"error": str(e), "data": None}

Q4: 可以在运行时改变状态结构吗?

不推荐。 State Schema 应该在定义图时确定。

如果需要动态字段:

python
# 使用 Dict 存储动态数据
class State(TypedDict):
    static_field: str
    dynamic_data: Dict[str, Any]  # 动态字段放这里

📖 扩展阅读


🎯 实践练习

练习 1: 实现聊天机器人状态

需求:

  • 存储对话历史(消息列表)
  • 存储用户信息(姓名、年龄)
  • 存储配置(API key、模型名称)

挑战: 分别用 TypedDict、Dataclass、Pydantic 实现,对比优缺点。

练习 2: 状态验证

需求:

  • 创建一个 Pydantic 状态
  • 添加验证器确保:
    • 年龄在 0-150 之间
    • 邮箱格式正确
    • 密码长度 >= 8

练习 3: 嵌套状态

需求:

  • 创建嵌套的状态结构(用户 → 地址 → 城市/街道)
  • 实现节点更新嵌套字段

总结:State Schema 是 LangGraph 的基础。理解三种定义方式的优缺点,能帮助你设计出更健壮、高效的 AI 应用。选择合适的状态定义方式,是构建可维护 LangGraph 应用的第一步!

基于 MIT 许可证发布。内容版权归作者所有。