文章目录
第 3 章：State 与 Schema — 让 Agent 拥有结构化记忆
3.1 本章目标
学完本章后，你将能够：
- 使用 Pydantic
BaseModel定义结构化的 Agent 状态（State） - 理解 State 在 LangGraph 工作流中的流转方式
- 掌握
previous参数的用法和实现状态链式传递 - 在多步骤流程中正确传递和更新状态
- 构建旅行规划助手 v0.2：从用户输入中提取结构化信息
3.2 核心概念（图文讲解）
3.2.1 生动类比：手术台上的病历
想象你在医院做手术。手术台上有一本病历，记录了你的所有关键信息：姓名、年龄、过敏史、血压、心率、手术部位等。每个参与手术的人（麻醉师、主刀医生、护士）都会：
- 读取病历上的信息
- 执行自己的专业操作
- 更新病历（记录麻醉剂量、手术进展、生命体征等）
- 把病历传给下一个人
LangGraph 的 State（状态） 就是这本「病历」——它是一个贯穿整个工作流的数据结构，每个 Task 都可以读取和更新它。在 Functional API 中，这个「病历」通过函数参数和返回值在 Task 之间传递。
3.2.2 状态流转图
3.2.3 关键术语解释
| 术语 | 含义 | 在病历中的类比 |
|---|---|---|
| State | 贯穿工作流的数据结构，使用 Pydantic BaseModel 定义 | 手术台上的病历 |
| Schema | State 的结构定义，规定有哪些字段、每个字段的类型 | 病历的格式模板 |
previous 参数 | 在 task 中接收上一个 task 执行的返回值 | 上一个医生传递的病历 |
| 状态流转 | State 在 Task 之间传递和更新的过程 | 病历在不同医生之间传递 |
| 不可变性 | 每个 task 返回新的 State 实例，而不是修改原有实例 | 每次更新盖新章，不涂改原内容 |
3.3 实战：旅行规划助手 v0.2
3.3.1 场景描述
在 v0.1 中，用户输入是一段自由文本，LLM 直接生成回复。但真实场景中，我们需要从用户输入中提取出结构化的信息（目的地、天数、预算、偏好），然后基于这些结构化信息生成行程，最后格式化为统一美观的输出。v0.2 将整个流程拆分为三个 Task，使用 Pydantic State 在它们之间传递数据。
3.3.2 完整代码
# ============================================================
# 旅行规划助手 v0.2 — 第 3 章完整可运行代码
# 保存为 ch03_state_schema.py 后运行：python ch03_state_schema.py
# 运行前请确保已设置：export OPENAI_API_KEY="your-key"
# 依赖：langgraph>=1.2.9, langchain-openai>=0.3.0, pydantic>=2.0
# ============================================================
import os # 导入操作系统模块，用于读取环境变量
from pydantic import BaseModel, Field # 从 Pydantic 导入基类和字段描述工具
from typing import Optional # Optional 类型用于声明可选字段
from langgraph.func import entrypoint, task # 导入 LangGraph Functional API 核心装饰器
from langchain_openai import ChatOpenAI # 导入 OpenAI 聊天模型封装
# ============================================================
# 第一步：定义 State（状态）—— 使用 Pydantic BaseModel
# State 是贯穿整个工作流的数据结构，每个 Task 都可以读取和更新它
# ============================================================
class TravelPlanState(BaseModel):
"""旅行规划状态：工作流中所有 Task 共享的数据结构。
每个字段都带有类型注解和描述，既是文档也是运行时验证。
"""
# 用户原始输入（整个工作流的起点）
raw_input: str = Field(
default="", # 默认值：空字符串
description="用户的原始自然语言输入", # 字段说明，便于理解和调试
)
# 从用户输入中提取的结构化信息
destination: str = Field(
default="", # 默认值：空字符串
description="旅行目的地，如：东京、巴黎、纽约",
)
days: int = Field(
default=0, # 默认值：0 天
description="旅行天数，如：3、5、7",
)
budget: str = Field(
default="", # 默认值：空字符串
description="预算范围，如：经济型、中等、豪华",
)
preferences: str = Field(
default="", # 默认值：空字符串
description="用户的偏好，如：美食、历史文化、自然风光、购物",
)
# 中间产物：LLM 生成的行程内容
itinerary: str = Field(
default="", # 默认值：空字符串
description="LLM 生成的详细行程内容",
)
# 最终输出：格式化后的完整结果
final_output: str = Field(
default="", # 默认值：空字符串
description="格式化后的最终输出内容",
)
# ============================================================
# 第二步：初始化 LLM（全局复用）
# ============================================================
llm = ChatOpenAI(
model="gpt-4o-mini", # 使用 gpt-4o-mini 模型
temperature=0.3, # 降低温度，让信息提取更准确/确定
)
# ============================================================
# 第三步：定义 Task 1 — 提取结构化需求
# ============================================================
@task # 标记为 LangGraph 任务
def extract_requirements(state: TravelPlanState) -> TravelPlanState:
"""从用户的原始输入中提取结构化信息。
读取 state.raw_input，调用 LLM 解析，填充 destination/days/budget/preferences 字段。
"""
# 从 state 中读取用户的原始输入
user_input = state.raw_input # 获取 state 中的 raw_input 字段
# 构建提示词：要求 LLM 以 JSON 格式返回结构化信息
prompt = f"""请从以下用户输入中提取旅行相关的结构化信息：
用户输入：{user_input}
请以 JSON 格式返回，包含以下字段：
- destination: 目的地（中文名称）
- days: 旅行天数（整数）
- budget: 预算（如：经济型、中等、豪华）
- preferences: 偏好（如：美食、历史文化、自然风光）
只返回 JSON，不要包含其他内容。"""
# 调用 LLM 进行信息提取
response = llm.invoke(prompt) # response 是 AIMessage 对象
# 解析 LLM 返回的 JSON 内容
import json # 导入 json 模块用于解析 LLM 返回的 JSON 字符串
try:
# 尝试从 LLM 回复中提取 JSON（LLM 可能用 ```json 包裹）
content = response.content # 获取 LLM 返回的文本
if "```json" in content:
# 如果 LLM 使用 Markdown 代码块包裹 JSON，提取中间部分
content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
# 如果 LLM 使用普通代码块包裹 JSON，提取中间部分
content = content.split("```")[1].split("```")[0].strip()
data = json.loads(content) # 将 JSON 字符串解析为 Python 字典
except (json.JSONDecodeError, IndexError):
# 如果 JSON 解析失败，使用默认值（避免整个流程崩溃）
data = {}
# 创建一个新的 state 副本，并更新提取到的字段
# 使用 model_copy() 创建副本，然后 update 更新字段——保持不可变性
updated_state = state.model_copy(
update={
"destination": data.get("destination", "未知目的地"), # 提取目的地，默认值兜底
"days": data.get("days", 1), # 提取天数，默认 1 天兜底
"budget": data.get("budget", "中等"), # 提取预算，默认中等兜底
"preferences": data.get("preferences", "综合体验"), # 提取偏好，默认值兜底
}
)
return updated_state # 返回更新后的 state（新实例，不修改原实例）
# ============================================================
# 第四步：定义 Task 2 — 生成行程
# ============================================================
@task # 标记为 LangGraph 任务
def generate_itinerary(state: TravelPlanState) -> TravelPlanState:
"""根据结构化信息生成详细的旅行行程。
读取 state 中的 destination/days/budget/preferences，调用 LLM 生成行程。
"""
# 构建包含结构化信息的提示词
prompt = f"""你是一个专业的旅行规划师。请根据以下信息生成一份详细的旅行行程：
目的地：{state.destination}
旅行天数：{state.days} 天
预算：{state.budget}
偏好：{state.preferences}
请按天组织行程，每天包含：
- 上午活动（景点参观、体验活动）
- 午餐推荐（当地特色餐厅）
- 下午活动（景点参观、购物）
- 晚餐推荐（当地特色餐厅）
- 晚间活动（可选：夜市、夜景、演出）
最后给出总预算估算。"""
# 调用 LLM 生成行程
response = llm.invoke(prompt) # 发送提示词，等待 LLM 回复
# 更新 state：将 LLM 生成的行程填入 itinerary 字段
updated_state = state.model_copy(
update={"itinerary": response.content} # 将 LLM 输出存入 itinerary
)
return updated_state # 返回更新后的 state
# ============================================================
# 第五步：定义 Task 3 — 格式化输出
# ============================================================
@task # 标记为 LangGraph 任务
def format_output(state: TravelPlanState) -> TravelPlanState:
"""将行程和结构化信息组合成美观的最终输出格式。
不涉及 LLM 调用，纯 Python 字符串格式化。
"""
# 构建 Markdown 格式的最终输出
final = f"""# 🌍 旅行规划报告
## 基本信息
- **目的地**：{state.destination}
- **旅行天数**：{state.days} 天
- **预算**：{state.budget}
- **偏好**：{state.preferences}
## 详细行程
{state.itinerary}
---
*由 LangGraph 旅行规划助手 v0.2 生成*
"""
# 更新 state：将格式化后的内容填入 final_output 字段
updated_state = state.model_copy(
update={"final_output": final} # 将格式化文本存入 final_output
)
return updated_state # 返回更新后的 state
# ============================================================
# 第六步：定义 Entrypoint — 编排整个工作流
# ============================================================
@entrypoint() # 标记为工作流入口
def travel_planner_v2(raw_input: str) -> TravelPlanState:
"""旅行规划助手 v0.2 入口函数。
编排三个 Task：提取需求 → 生成行程 → 格式化输出。
使用 previous 参数进行状态链式传递。
"""
# 第一步：创建初始 State，将用户输入存入 raw_input 字段
initial_state = TravelPlanState(raw_input=raw_input)
# 第二步：调用 extract_requirements，传入初始 state
# .result() 等待 Task 完成，返回更新后的 state
state_v1 = extract_requirements(initial_state).result()
# 第三步：调用 generate_itinerary，将上一步的结果作为输入
# 这就是「状态链式传递」——每个 Task 的输出成为下一个 Task 的输入
state_v2 = generate_itinerary(state_v1).result()
# 第四步：调用 format_output，将上一步的结果作为输入
state_v3 = format_output(state_v2).result()
# 返回最终 state（包含完整的所有字段）
return state_v3
# ============================================================
# 程序入口：运行 Agent
# ============================================================
if __name__ == "__main__":
# 检查 API Key 是否已配置
if not os.environ.get("OPENAI_API_KEY"):
print("错误：未设置 OPENAI_API_KEY 环境变量！")
print("请先运行：export OPENAI_API_KEY='your-api-key-here'")
exit(1)
# 模拟用户输入
user_query = "我想去日本东京玩5天，预算中等，喜欢美食和历史文化。"
print("=" * 60)
print("旅行规划助手 v0.2 正在处理...")
print("=" * 60)
# 调用入口函数，传入用户输入
final_state = travel_planner_v2.invoke(user_query)
# 打印最终输出
print(final_state.final_output) # 输出格式化后的完整报告
print("=" * 60)
# 额外展示：打印 State 中所有字段的值（用于调试和理解）
print("\n[调试信息] State 各字段内容：")
print(f" destination: {final_state.destination}")
print(f" days: {final_state.days}")
print(f" budget: {final_state.budget}")
print(f" preferences: {final_state.preferences}")
print(f" itinerary 长度: {len(final_state.itinerary)} 字符")
print(f" final_output 长度: {len(final_state.final_output)} 字符")
3.3.3 运行结果展示
运行 python ch03_state_schema.py 后，你将看到类似以下输出：
============================================================
旅行规划助手 v0.2 正在处理...
============================================================
# 🌍 旅行规划报告
## 基本信息
- **目的地**：东京
- **旅行天数**：5 天
- **预算**：中等
- **偏好**：美食、历史文化
## 详细行程
**第 1 天：浅草文化与下町风情**
- 上午：参观浅草寺与雷门，感受江户时代的文化氛围
- 午餐：浅草今半，品尝百年历史的寿喜烧
- 下午：漫步合羽桥道具街，探索日本厨具文化
- 晚餐：在吾妻桥附近的居酒屋体验地道日式下酒菜
- 晚间：隅田川散步，欣赏东京晴空塔夜景
**第 2 天：皇居周边与银座美食**
- 上午：参观皇居东御苑，了解日本皇室历史
...
---
*由 LangGraph 旅行规划助手 v0.2 生成*
============================================================
[调试信息] State 各字段内容：
destination: 东京
days: 5
budget: 中等
preferences: 美食、历史文化
itinerary 长度: 1247 字符
final_output 长度: 1370 字符
3.3.4 代码逐段解析
第一部分：Pydantic State 定义
class TravelPlanState(BaseModel):
raw_input: str = Field(default="", description="用户的原始自然语言输入")
destination: str = Field(default="", description="旅行目的地")
days: int = Field(default=0, description="旅行天数")
...
使用 Pydantic BaseModel 定义 State 有三大好处：(1) 自动类型验证——如果传入了错误类型，Pydantic 会抛出清晰的错误；(2) IDE 自动补全——所有字段都有完整的类型提示；(3) 文档即代码——Field(description="...") 既是注释，也可被其他工具读取。
第二部分：状态更新模式
updated_state = state.model_copy(
update={"destination": data.get("destination", "未知目的地")}
)
model_copy(update={...}) 是 Pydantic v2 的推荐方式——它会创建一个新的 State 实例，只更新指定的字段，其他字段保持不变。这是「不可变更新」模式，避免了直接修改原对象可能带来的副作用。
第三部分：状态链式传递
state_v1 = extract_requirements(initial_state).result()
state_v2 = generate_itinerary(state_v1).result()
state_v3 = format_output(state_v2).result()
每个 Task 接收上一个 Task 的返回结果作为输入，这就是「状态链式传递」。数据像流水线一样依次经过每个 Task，每个 Task 只负责更新自己关心的字段。
第四部分：previous 参数模式（可选写法）
在 LangGraph Functional API 中，你也可以使用 previous 参数让 Task 自动接收上一个 Task 的输出：
@task
def generate_itinerary(previous: TravelPlanState) -> TravelPlanState:
# previous 参数自动接收上一个 Task 的返回值
# 不需要手动传递，LangGraph 运行时自动处理
...
不过在本教程中，我们采用显式传递的方式，因为这样代码更清晰、更容易理解和调试。
3.4 API 速查
| API | 类型 | 说明 | 签名/导入路径 |
|---|---|---|---|
BaseModel | 类 | Pydantic 数据模型基类 | from pydantic import BaseModel |
Field() | 函数 | 定义字段的元数据（默认值、描述等） | from pydantic import Field |
model_copy(update={...}) | 方法 | 创建 State 副本并更新指定字段 | new_state = state.model_copy(update={"field": value}) |
model_dump() | 方法 | 将 State 序列化为字典 | data = state.model_dump() |
.result() | 方法 | 等待 Task 完成并返回结果 | result = my_task(state).result() |
previous | 参数约定 | 在 Task 中声明以接收上一个 Task 的返回值 | def my_task(previous: MyState) -> MyState: |
3.5 常见错误与避坑指南
错误 1：直接修改 State 而不是创建新实例
错误代码：
@task
def my_bad_task(state: TravelPlanState) -> TravelPlanState:
# 错误：直接修改原 State 实例的属性
state.destination = "东京" # 直接修改原对象
state.days = 5
return state # 返回的是被修改过的原对象
正确代码：
@task
def my_good_task(state: TravelPlanState) -> TravelPlanState:
# 正确：使用 model_copy() 创建新实例，保持不可变性
updated_state = state.model_copy(
update={
"destination": "东京", # 在新实例上更新字段
"days": 5,
}
)
return updated_state # 返回全新的实例
原因：直接修改原实例可能导致意外的副作用（例如其他 Task 也引用了同一个 State 对象）。不可变更新是 LangGraph 设计的最佳实践，也是函数式编程的核心原则。
错误 2：State 字段类型不匹配
错误代码：
class TravelPlanState(BaseModel):
days: int = Field(default=0, description="旅行天数")
# 在某个 Task 中
updated_state = state.model_copy(
update={"days": "5天"} # 错误：days 是 int 类型，但传入了字符串 "5天"
)
正确代码：
class TravelPlanState(BaseModel):
days: int = Field(default=0, description="旅行天数")
# 在某个 Task 中
updated_state = state.model_copy(
update={"days": 5} # 正确：传入 int 类型，与字段声明一致
)
原因：Pydantic 会进行运行时类型验证。如果传入的类型不匹配，在某些情况下 Pydantic 会尝试自动转换（如 "5" 转为 5），但 "5天" 无法转换，会抛出 ValidationError。
错误 3：忘记给 State 字段设置默认值
错误代码：
class TravelPlanState(BaseModel):
destination: str # 错误：没有默认值，创建实例时必须传入
days: int
budget: str
正确代码：
class TravelPlanState(BaseModel):
destination: str = Field(default="") # 正确：有默认值，可以逐步填充
days: int = Field(default=0)
budget: str = Field(default="")
原因：State 在工作流中是被逐步填充的。如果字段没有默认值，创建初始 State 时就必须传入所有字段的值，这违背了「逐步构建」的设计理念。
错误 4：在 LLM 提示词中未处理 JSON 解析失败
错误代码：
@task
def extract_requirements(state: TravelPlanState) -> TravelPlanState:
response = llm.invoke(prompt)
data = json.loads(response.content) # 错误：LLM 可能返回非 JSON 格式
return state.model_copy(update={"destination": data["destination"]})
正确代码：
@task
def extract_requirements(state: TravelPlanState) -> TravelPlanState:
response = llm.invoke(prompt)
try:
content = response.content
# 处理 LLM 可能用 Markdown 代码块包裹的情况
if "```" in content:
content = content.split("```")[1].split("```")[0].strip()
data = json.loads(content)
except (json.JSONDecodeError, IndexError, KeyError):
data = {} # 解析失败时使用空字典，后续用 .get() 安全取值
return state.model_copy(
update={
"destination": data.get("destination", "未知"),
"days": data.get("days", 1),
}
)
原因：LLM 的输出不是 100% 可靠的。它可能返回非 JSON 文本、用 Markdown 代码块包裹 JSON、或者 JSON 格式不完整。始终添加 try-except 保护并用 .get() 提供默认值。
3.6 最佳实践总结
-
State 字段都设默认值：State 在工作流中被逐步填充，每个字段都应该有合理的默认值。这样创建初始 State 时只需要传入关键字段，其他字段在后续 Task 中逐步完善。
-
使用 model_copy() 而非直接修改：保持 State 的不可变性——每个 Task 返回新实例，而不是修改原实例。这避免了副作用，让代码的行为更可预测。
-
LLM 解析加容错：任何从 LLM 输出中解析结构化数据的操作都必须添加 try-except 保护。LLM 的输出格式不稳定，提供默认值兜底可以避免整个流程崩溃。
-
State 字段命名清晰：使用有意义的字段名（如
raw_input、itinerary、final_output），让每个字段的用途一目了然。配合Field(description="...")提供额外文档。 -
Task 职责单一：每个 Task 只做一件事——提取需求、生成行程、格式化输出。不要在一个 Task 中混入多种职责，这样代码更易测试、更易维护、更易复用。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/weixin_42439274/article/details/162849834
