LangGraph State Schema 详细解读
📚 概述
本文档详细解读 LangGraph 中的 State Schema(状态模式)。状态是 LangGraph 的核心概念,它定义了图中数据的结构和类型。理解状态模式是掌握 LangGraph 的关键一步。
在本教程中,我们将深入探讨:
- 什么是 State Schema
- 三种定义状态的方式(TypedDict、Dataclass、Pydantic)
- 每种方式的优缺点和适用场景
- 实战代码详解
🎯 核心概念
什么是 State Schema?
在 LangGraph 中,State Schema 定义了:
- 数据结构:图中存储哪些数据
- 数据类型:每个字段的类型是什么
- 通信协议:节点之间如何传递和更新数据
可以把 State 想象成一个共享的数据容器,所有节点都可以读取和更新它。
核心特性
1. Channel(通道)概念
State 中的每个键(key)就是一个通道(channel):
State = {
"name": "Alice", # name 通道
"mood": "happy" # mood 通道
}
- 每个节点可以读取任何通道的值
- 每个节点可以更新一个或多个通道
- 通道之间相互独立
2. 节点更新机制
def node_1(state):
# 读取 name 通道
current_name = state["name"]
# 更新 name 通道(返回字典)
return {"name": current_name + " is happy"}
关键点:
- 输入:节点接收完整的 State
- 输出:节点返回要更新的字段(部分更新)
- 合并:LangGraph 自动将返回值合并到 State 中
🔧 三种定义方式
LangGraph 支持三种定义 State Schema 的方式,每种都有独特的优势。
方式 1: TypedDict ⭐
这是最常用、最简单的方式。
基础用法
from typing_extensions import TypedDict
class TypedDictState(TypedDict):
foo: str
bar: str
特点:
- ✅ 语法简洁
- ✅ IDE 支持好(自动补全、类型提示)
- ✅ 性能最优(无运行时验证)
- ❌ 不做运行时类型检查
进阶用法:Literal 类型
from typing import Literal
class TypedDictState(TypedDict):
name: str
mood: Literal["happy", "sad"] # 限定只能是这两个值之一
Literal 详解:
Literal
是 Python 的类型提示工具,用于限制值的范围:
# ✅ 正确
state = {"name": "Alice", "mood": "happy"}
# ❌ IDE 会警告(但运行时不报错)
state = {"name": "Alice", "mood": "angry"} # angry 不在允许的值中
在 LangGraph 中使用
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+ 提供的数据类装饰器。
基础用法
from dataclasses import dataclass
@dataclass
class DataclassState:
name: str
mood: Literal["happy", "sad"]
与 TypedDict 的区别
特性 | TypedDict | Dataclass |
---|---|---|
访问方式 | state["name"] | state.name |
初始化 | 字典 | 类实例 |
性能 | 更快 | 稍慢 |
功能 | 基础 | 更多(默认值、方法等) |
在 LangGraph 中使用
# 定义节点 - 注意访问方式改变
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,但节点仍然返回字典:
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
:
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 对比
# TypedDict - 不做运行时检查
class DataclassState:
name: str
mood: Literal["happy", "sad"]
# ❌ 这段代码可以运行(虽然 IDE 会警告)
state = DataclassState(name="Lance", mood="angry") # 没有报错
# 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!
自定义验证器详解
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 中使用
# 创建图(与之前完全相同)
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)
↓ ↓
\ /
\ /
↓ ↓
返回结果
完整代码实现
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. 条件边的工作原理
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 在条件函数中的作用
def decide_mood(state) -> Literal["node_2", "node_3"]:
# ^^^^^^^^^^^^^^^^^^^^^^^^^^
# 返回类型提示:只能是这两个字符串之一
return "node_2" # ✅ 正确
return "node_5" # ❌ IDE 会警告
这不仅是类型提示,还帮助 LangGraph 理解可能的路由目标。
3. 状态更新的执行顺序
初始状态: {"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(通道)模型
State = {
"name": ..., # name 通道
"mood": ... # mood 通道
}
# 节点可以更新部分通道
def node(state):
return {"name": "new_value"} # 只更新 name 通道
关键特性:
- 每个字段是独立的通道
- 节点可以只更新部分通道
- 未返回的通道保持不变
3. 三种状态方式对比
特性 | TypedDict | Dataclass | Pydantic |
---|---|---|---|
语法复杂度 | ⭐ 简单 | ⭐⭐ 中等 | ⭐⭐ 中等 |
访问方式 | state["key"] | state.key | state.key |
运行时验证 | ❌ 无 | ❌ 无 | ✅ 有 |
性能 | 最快 | 快 | 稍慢(有验证) |
适用场景 | 简单类型 | 需要默认值/方法 | 需要严格验证 |
初始化方式 | 字典 | 类实例 | 类实例 |
选择建议:
# 简单项目 → 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 详解
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 类型详解
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 详解
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 进阶用法
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 - 大多数情况
# 推荐用于:
# - 简单的数据结构
# - 性能敏感的应用
# - 不需要运行时验证的场景
class State(TypedDict):
messages: list
user_id: str
✅ Dataclass - 需要默认值或方法
# 推荐用于:
# - 需要默认值
# - 需要计算属性
# - 面向对象设计
@dataclass
class State:
messages: list = field(default_factory=list)
retries: int = 3
def should_retry(self):
return self.retries > 0
✅ Pydantic - 需要严格验证
# 推荐用于:
# - 外部输入(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: 最小化状态
# ❌ 不好 - 包含冗余数据
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: 使用明确的类型
# ❌ 不好 - 类型模糊
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: 分组相关字段
# ❌ 不好 - 平铺所有字段
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: 修改可变对象
# ❌ 错误 - 直接修改 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: 忘记初始化
# ❌ 错误 - 字段未初始化
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 验证性能
# ❌ 性能问题 - 复杂验证
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. 混合使用多种状态
# 全局状态用 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. 动态状态字段
# 使用 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. 状态版本管理
# 添加版本字段跟踪状态变化
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"
📊 三种方式对比总结
功能对比表
功能 | TypedDict | Dataclass | Pydantic |
---|---|---|---|
类型提示 | ✅ | ✅ | ✅ |
IDE 支持 | ✅ | ✅ | ✅ |
运行时验证 | ❌ | ❌ | ✅ |
默认值 | ❌ | ✅ | ✅ |
自定义方法 | ❌ | ✅ | ✅ |
访问方式 | ["key"] | .key | .key |
初始化 | 字典 | 实例 | 实例 |
性能 | 最快 | 快 | 中等 |
学习曲线 | 低 | 中 | 中 |
使用场景总结
# 🎯 选择决策树
# 1. 需要运行时验证?
# 是 → Pydantic
# 否 → 继续
# 2. 需要默认值或方法?
# 是 → Dataclass
# 否 → 继续
# 3. 追求最佳性能?
# 是 → TypedDict
实际项目建议
小型项目(< 10 个节点):
# 全用 TypedDict
class State(TypedDict):
messages: list
config: dict
中型项目(10-50 个节点):
# 主状态用 TypedDict,配置用 Pydantic
class State(TypedDict):
messages: list
config: AppConfig # Pydantic model
class AppConfig(BaseModel):
api_key: str
timeout: int
大型项目(> 50 个节点):
# 分层设计
class BaseState(TypedDict):
# 核心字段
messages: list
class ValidatedState(BaseModel):
# 需验证的字段
user_input: str
@dataclass
class CachedState:
# 需默认值的字段
cache: dict = field(default_factory=dict)
🔍 常见问题
Q1: TypedDict 的类型提示不起作用怎么办?
原因: TypedDict 只提供静态类型检查,不做运行时验证。
解决方案:
- 使用 IDE(VSCode、PyCharm)进行静态检查
- 使用 mypy 进行类型检查:bash
pip install mypy mypy your_script.py
- 如需运行时验证,改用 Pydantic
Q2: 为什么 Dataclass 节点还要返回字典?
原因: LangGraph 将状态按字段(channel)存储,节点返回的字典会自动映射到对应的 channel。
示例:
@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
,导致图执行中断。
解决方案:
def node(state):
try:
validated = MyModel(**state)
return {"data": validated.data}
except ValidationError as e:
# 处理错误
return {"error": str(e), "data": None}
Q4: 可以在运行时改变状态结构吗?
不推荐。 State Schema 应该在定义图时确定。
如果需要动态字段:
# 使用 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 应用的第一步!