写在前面
在构建复杂 AI Agent 时，我们经常面临一个核心挑战：如何让 Agent 的行为既灵活又可控？
传统的线性流程（如 LangChain 的 Chain）只能按预设顺序执行，无法处理分支、循环等复杂逻辑。而 LangGraph 的 StateGraph 正是为解决这一问题而生——它将 Agent 工作流建模为状态机，通过显式的状态管理，让复杂的流程变得清晰、可控、可调试。
本文将深入剖析 LangGraph StateGraph 的设计原理，并提供完整的实战代码，帮助你掌握复杂 Agent 工作流的正确设计姿势。
一、为什么 Agent 需要状态机？
1.1 从线性链到状态机的演进
让我们先理解一个核心概念：线性链 vs 状态机的区别。
┌─────────────────────────────────────────────────────────────────────┐
│ 线性链（Linear Chain） │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │ 输入 │───▶│ 步骤1 │───▶│ 步骤2 │───▶│ 输出 │ │
│ └────────┘ └────────┘ └────────┘ └────────┘ │
│ │
│ 特点： │
│ - 每一步都有明确的下一个步骤 │
│ - 无法根据结果调整流程 │
│ - 适合简单的一次性任务 │
│ │
└─────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────┐
│ 状态机（State Machine） │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌────────┐ │
│ ┌─────│ 状态A │─────┐ │
│ │ └────────┘ │ │
│ ▼ ▼ │
│ ┌────────┐ ┌────────┐ │
│ │ 状态B │──────────▶│ 状态C │ │
│ └────────┘ └────────┘ │
│ │ │ │
│ │ ┌────────┐ │ │
│ └────▶│ 状态D │◀────┘ │
│ └────────┘ │
│ │ │
│ ▼ │
│ ┌────────┐ │
│ │ 结束 │ │
│ └────────┘ │
│ │
│ 特点： │
│ - 每个状态可以转换到多个可能的下个状态 │
│ - 可以根据条件动态选择下一个状态 │
│ - 支持循环和分支，适合复杂的交互式任务 │
│ │
└─────────────────────────────────────────────────────────────────────┘1.2 为什么简单链不够用？
考虑一个实际的客服 Agent 场景：
用户：帮我查一下订单
Agent 传统流程：
1. 理解用户意图 → 2. 调用订单 API → 3. 返回结果 → 结束
Agent 真实场景：
1. 理解用户意图 → 2. 调用订单 API
- 如果 API 超时：重试 3 次？还是转人工？
- 如果订单不存在：是直接说"没有找到"还是问"是哪个订单？"
- 如果用户追问：是继续查物流还是回答其他问题？
3. 返回结果 → 用户可能还有后续问题 → 回到步骤1真实场景充满了分支、循环、异常处理——这些都是线性链无法优雅处理的。
1.3 状态机的核心价值
StateGraph 状态机为 Agent 开发带来了三个核心价值：
二、LangGraph StateGraph 核心概念
2.1 概念全景图
┌─────────────────────────────────────────────────────────────────────┐
│ LangGraph StateGraph 核心概念 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Graph │ │
│ │ 图 │ │
│ └──────┬──────┘ │
│ │ │
│ ┌───────────┼───────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ State │ │ Node │ │ Edge │ │
│ │ 状态 │ │ 节点 │ │ 边 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ State（状态） │
│ ├─ 定义：整个应用的"全局上下文仓库" │
│ ├─ 结构：TypedDict 或 Pydantic 模型 │
│ └─ 更新：通过节点返回增量更新，框架自动合并 │
│ │
│ Node（节点） │
│ ├─ 定义：执行业务逻辑的函数 │
│ ├─ 输入：当前状态（State） │
│ └─ 输出：部分状态更新（dict） │
│ │
│ Edge（边） │
│ ├─ 定义：节点之间的流转关系 │
│ ├─ 类型：静态边（固定跳转）、条件边（动态决策） │
│ └─ 循环：通过条件边实现 │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 状态设计：TypedDict 定义状态结构
状态是 StateGraph 的核心，所有节点共享同一个状态空间。
from typing import TypedDict, Annotated, Optional
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
"""
Agent 状态结构定义
使用 TypedDict 定义状态的好处：
1. 类型检查：在 IDE 中可以获得完整的类型提示
2. 代码文档：状态字段一目了然
3. 易于扩展：可以随时添加新字段
"""
# === 消息历史 ===
# Annotated + add_messages：自动合并新旧消息
messages: Annotated[list, add_messages]
# === 对话上下文 ===
user_id: str # 用户 ID
session_id: str # 会话 ID
current_intent: Optional[str] = None # 当前识别到的意图
# === 任务进度 ===
task_status: str # pending/in_progress/completed/failed
task_result: Optional[str] = None # 任务结果
error_message: Optional[str] = None # 错误信息
# === 业务数据 ===
retrieved_docs: list = [] # 检索到的文档
search_results: list = [] # 搜索结果
# === 元信息 ===
revision_count: int = 0 # 修改/重试次数
created_at: Optional[str] = None # 创建时间状态设计的最佳实践：
精简原则：只存储"影响流程走向"和"后续节点需要"的字段
类型标注：使用完整的类型标注，充分利用 IDE 的类型检查
默认值：为可选字段提供合理的默认值
分区设计：将状态按功能分组（如消息、业务数据、元信息）
2.3 节点设计：每个节点是一个纯函数
节点是状态机中的"执行单元"，负责具体的业务逻辑。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
# 初始化 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# === 节点1：理解用户意图 ===
def understand_intent(state: AgentState) -> dict:
"""
理解用户意图节点
输入：用户最新消息
输出：识别到的意图
业务逻辑：
1. 提取消息内容
2. 调用 LLM 进行意图分类
3. 返回意图标签
"""
messages = state["messages"]
latest_message = messages[-1].content if messages else ""
# 构建意图识别提示
intent_prompt = f"""
请分析以下用户消息，识别其主要意图：
用户消息：{latest_message}
可能的意图：
- query_order: 查询订单
- cancel_order: 取消订单
- track_shipment: 物流追踪
- refund: 退款
- complaint: 投诉
- chitchat: 闲聊
只输出意图标签，不需要其他内容。
"""
response = llm.invoke(intent_prompt)
intent = response.content.strip()
return {"current_intent": intent}
# === 节点2：订单查询 ===
def query_order(state: AgentState) -> dict:
"""
订单查询节点
输入：用户 ID + 意图
输出：订单信息或错误信息
业务逻辑：
1. 模拟调用订单 API
2. 返回查询结果
3. 如果失败，返回错误信息
"""
user_id = state.get("user_id", "")
# 模拟订单查询（实际项目中替换为真实 API 调用）
# order_data = order_api.get_orders(user_id)
order_data = {
"order_id": "ORD123456",
"status": "shipped",
"items": ["商品A x 1", "商品B x 2"],
"total": 299.00
}
result = f"您的订单状态：已发货\n订单号：{order_data['order_id']}\n商品：{', '.join(order_data['items'])}"
return {
"task_status": "completed",
"task_result": result
}
# === 节点3：物流追踪 ===
def track_shipment(state: AgentState) -> dict:
"""
物流追踪节点
"""
# 模拟物流查询
shipment_info = """
物流信息：
- 2024-01-15 10:30:00 包裹已发出
- 2024-01-16 14:20:00 到达中转站
- 2024-01-17 09:15:00 正在派送
"""
return {
"task_status": "completed",
"task_result": shipment_info
}
# === 节点4：生成回复 ===
def generate_response(state: AgentState) -> dict:
"""
生成最终回复节点
整合任务结果，生成对用户的回复
"""
task_result = state.get("task_result", "抱歉，暂时无法处理您的请求。")
response_prompt = f"""
请将以下任务结果转化为自然的对话回复：
任务结果：{task_result}
要求：
- 语言亲切友好
- 如果有后续操作建议，一并告知用户
- 控制在 100 字以内
"""
response = llm.invoke(response_prompt)
return {"messages": [response]}2.4 边设计：条件边、循环边、并行边
边定义了状态机中的流转规则，是状态机的"交通规则"。
from langgraph.graph import StateGraph, START, END
# === 构建工作流图 ===
workflow = StateGraph(AgentState)
# === 1. 注册所有节点 ===
workflow.add_node("understand_intent", understand_intent)
workflow.add_node("query_order", query_order)
workflow.add_node("track_shipment", track_shipment)
workflow.add_node("generate_response", generate_response)
# === 2. 定义静态边 ===
# 从起点到意图理解
workflow.add_edge(START, "understand_intent")
# === 3. 定义条件边（核心！）===
def route_by_intent(state: AgentState) -> str:
"""
意图路由函数
根据识别的意图，决定下一步执行哪个节点
Args:
state: 当前状态
Returns:
下一个节点的名称
"""
intent = state.get("current_intent", "")
# 根据意图路由到不同的处理节点
intent_routes = {
"query_order": "query_order",
"track_shipment": "track_shipment",
"cancel_order": "query_order", # 取消订单也先查询
"refund": "query_order",
"complaint": "generate_response", # 投诉直接转人工
}
return intent_routes.get(intent, "generate_response")
# 添加条件边：从意图理解根据意图类型路由
workflow.add_conditional_edges(
source="understand_intent",
path=route_by_intent,
path_map={
"query_order": "query_order",
"track_shipment": "track_shipment",
"generate_response": "generate_response"
}
)
# === 4. 完成后生成回复 ===
workflow.add_edge("query_order", "generate_response")
workflow.add_edge("track_shipment", "generate_response")
workflow.add_edge("generate_response", END)
# === 5. 编译图 ===
app = workflow.compile()2.5 循环边的实现
循环是状态机的核心能力之一。下面展示如何实现"失败重试"机制：
# === 带有重试机制的工作流 ===
class RetryState(TypedDict):
"""支持重试的状态"""
task_name: str
attempt: int # 当前尝试次数
max_attempts: int = 3 # 最大尝试次数
result: Optional[str] = None
error: Optional[str] = None
status: str = "pending" # pending/running/success/failed
def execute_task(state: RetryState) -> dict:
"""执行任务节点"""
import random
# 模拟任务执行（70% 成功率）
success = random.random() > 0.3
if success:
return {
"result": f"任务 {state['task_name']} 执行成功！",
"status": "success"
}
else:
return {
"error": "任务执行失败，请重试",
"status": "failed",
"attempt": state["attempt"] + 1
}
def check_retry(state: RetryState) -> str:
"""
检查是否需要重试
Returns:
"execute_task" - 继续重试
END - 结束流程
"""
# 如果达到最大尝试次数，结束
if state["attempt"] >= state["max_attempts"]:
return END
# 如果任务失败但还有尝试次数，继续重试
if state["status"] == "failed":
return "execute_task"
return END
# 构建带循环的工作流
retry_workflow = StateGraph(RetryState)
retry_workflow.add_node("execute_task", execute_task)
# 设置入口
retry_workflow.add_edge(START, "execute_task")
# 添加条件边：执行完成后检查是否需要重试
retry_workflow.add_conditional_edges(
source="execute_task",
path=check_retry,
path_map={
"execute_task": "execute_task", # 重试
END: END # 结束
}
)
retry_app = retry_workflow.compile()
# 执行
result = retry_app.invoke({
"task_name": "数据同步",
"attempt": 0,
"max_attempts": 3
})
print(f"最终状态: {result['status']}")
print(f"总尝试次数: {result['attempt']}")三、人工介入（Human-in-the-loop）设计
3.1 什么是 Human-in-the-loop？
在某些关键决策点，我们需要人工介入，而不是完全交给 AI 自动执行。
┌─────────────────────────────────────────────────────────────────────┐
│ Human-in-the-loop 工作流 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ 开始 │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ 自动处理 │──── 正常流程 ────▶ 生成结果 │
│ └────┬────┘ │
│ │ │
│ │ 触发人工介入 │
│ ▼ │
│ ┌─────────┐ │
│ │ 等待审批 │◀─────── 人工处理 ───────┐ │
│ └────┬────┘ │ │
│ │ │ │
│ ▼ │ │
│ ┌─────────┐ │ │
│ │ 审批通过？│ │ │
│ └────┬────┘ │ │
│ │ │ │
│ ┌───┴───┐ │ │
│ │是 否│ │ │
│ ▼ ▼ │ │
│ ┌─────┐ ┌─────┐ │ │
│ │继续 │ │终止 │ │ │
│ └──┬──┘ └──┬──┘ │ │
│ │ │ │ │
│ └────────┴───────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘3.2 人工介入的实现
from enum import Enum
from typing import Union
class ApprovalStatus(Enum):
"""审批状态枚举"""
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
class ApprovalState(TypedDict):
"""包含人工审批的状态"""
request_id: str
request_type: str # refund/complaint/quote
content: str
approval_status: ApprovalStatus
approver_comments: Optional[str]
final_decision: Optional[str]
def auto_process(state: ApprovalState) -> dict:
"""自动处理请求"""
# 模拟自动审核（简单规则）
is_auto_approved = len(state["content"]) < 100
if is_auto_approved:
return {
"approval_status": ApprovalStatus.APPROVED,
"final_decision": "自动审核通过"
}
else:
return {
"approval_status": ApprovalStatus.PENDING,
"final_decision": None
}
def wait_for_human(state: ApprovalState) -> dict:
"""
等待人工审批
在实际应用中，这里会：
1. 发送通知给审批人
2. 暂停工作流执行
3. 等待审批结果
"""
# 模拟人工审批结果
# 实际应用中，应该通过外部系统（如飞书审批、邮件）获取结果
print(f"\n⏳ 等待人工审批...")
print(f" 请求ID: {state['request_id']}")
print(f" 内容: {state['content'][:50]}...")
# 这里模拟审批结果
# 实际项目中，应该从外部系统获取
# 例如：读取数据库、轮询消息队列等
simulated_approval = ApprovalStatus.APPROVED
simulated_comments = "同意退款处理"
return {
"approval_status": simulated_approval,
"approver_comments": simulated_comments,
"final_decision": "人工审批通过" if simulated_approval == ApprovalStatus.APPROVED else "人工审批拒绝"
}
def process_approval(state: ApprovalState) -> dict:
"""处理审批结果"""
if state["approval_status"] == ApprovalStatus.APPROVED:
return {"final_decision": f"{state['final_decision']} - {state['approver_comments']}"}
else:
return {"final_decision": "请求被拒绝"}
def route_approval(state: ApprovalState) -> str:
"""根据审批状态路由"""
if state["approval_status"] == ApprovalStatus.PENDING:
return "wait_for_human"
return "process_approval"
# 构建带人工介入的工作流
approval_workflow = StateGraph(ApprovalState)
approval_workflow.add_node("auto_process", auto_process)
approval_workflow.add_node("wait_for_human", wait_for_human)
approval_workflow.add_node("process_approval", process_approval)
approval_workflow.add_edge(START, "auto_process")
approval_workflow.add_conditional_edges(
source="auto_process",
path=route_approval,
path_map={
"wait_for_human": "wait_for_human",
"process_approval": "process_approval"
}
)
approval_workflow.add_edge("wait_for_human", "process_approval")
approval_workflow.add_edge("process_approval", END)
approval_app = approval_workflow.compile()
# 测试
result = approval_app.invoke({
"request_id": "REQ001",
"request_type": "refund",
"content": "商品损坏，要求全额退款。订单号：12345，购买日期：2024-01-10",
"approval_status": ApprovalStatus.PENDING,
"approver_comments": None,
"final_decision": None
})
print(f"\n✅ 最终决策: {result['final_decision']}")四、持久化：基于 Checkpoint 的状态持久化
4.1 为什么需要持久化？
在实际应用中，我们经常需要：
断点续传：用户中断后，下次继续执行
多会话支持：同一个工作流支持多个并发用户
故障恢复：系统崩溃后恢复执行状态
LangGraph 通过 Checkpoint 机制实现了状态的持久化。
4.2 Checkpoint 的工作原理
┌─────────────────────────────────────────────────────────────────────┐
│ Checkpoint 持久化机制 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 内存状态 ──────────────────────────────────────▶ 持久化存储 │
│ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Step 1 状态 │ ─── checkpoint.save() ───────────────────▶ │ Checkpoint 1 │ │
│ └─────────────┘ └─────────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Step 2 状态 │ ─── checkpoint.save() ───────────────────▶ │ Checkpoint 2 │ │
│ └─────────────┘ └─────────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Step 3 状态 │ ─── checkpoint.save() ───────────────────▶ │ Checkpoint 3 │ │
│ └─────────────┘ └─────────────┘ │
│ │
│ 恢复时 ──────────────────────────────────────▶ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ 最新状态 │ ◀──│ Checkpoint │ ◀── 从指定 checkpoint 恢复 │
│ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘4.3 持久化实现
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
class PersistentState(TypedDict):
"""支持持久化的状态"""
messages: Annotated[list, add_messages]
user_id: str
conversation_turns: int
context: dict
def create_checkpointer(persist_directory: str = "./checkpoints"):
"""
创建检查点持久化器
Args:
persist_directory: 持久化存储目录
Returns:
检查点持久化器
"""
import os
os.makedirs(persist_directory, exist_ok=True)
# 选择持久化后端
# 1. MemorySaver: 内存存储，适合开发调试
# 2. SqliteSaver: SQLite 存储，适合单机部署
# 3. PostgresSaver: PostgreSQL 存储，适合生产环境
# 使用 SQLite 作为持久化后端
db_path = os.path.join(persist_directory, "checkpoints.db")
checkpointer = SqliteSaver.from_conn_string(db_path)
return checkpointer
def build_persistent_workflow(checkpointer):
"""构建支持持久化的工作流"""
def node_a(state: PersistentState) -> dict:
return {"conversation_turns": 1}
def node_b(state: PersistentState) -> dict:
return {"conversation_turns": state["conversation_turns"] + 1}
workflow = StateGraph(PersistentState)
workflow.add_node("node_a", node_a)
workflow.add_node("node_b", node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)
# 编译时传入检查点持久化器
return workflow.compile(checkpointer=checkpointer)
# === 使用示例 ===
# 1. 创建持久化器
checkpointer = create_checkpointer("./checkpoints")
# 2. 构建工作流
app = build_persistent_workflow(checkpointer)
# 3. 配置检查点（为每个会话创建独立的 thread）
config = {
"configurable": {
"thread_id": "user_123_session_001" # 会话唯一标识
}
}
# 4. 首次执行
print("=== 首次执行 ===")
result1 = app.invoke(
{"messages": [], "user_id": "user_123", "conversation_turns": 0, "context": {}},
config=config
)
print(f"Turn: {result1['conversation_turns']}")
# 5. 模拟中断后恢复（不传输入，自动从 checkpoint 恢复状态）
print("\n=== 模拟恢复 ===")
result2 = app.invoke(None, config=config) # 输入为 None，自动从 checkpoint 恢复
print(f"Turn: {result2['conversation_turns']}")
# 6. 查看检查点历史
print("\n=== 检查点历史 ===")
for checkpoint in app.get_state_history(config):
print(f" - Turn {checkpoint['config']['configurable']['turn']}: {checkpoint['values']}")五、复杂工作流示例：多步研究 Agent
5.1 完整架构图
┌─────────────────────────────────────────────────────────────────────┐
│ 多步研究 Agent 完整架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ START │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ understand │ 理解用户研究主题 │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ search │───▶│ search │ 并行搜索多个信息源 │
│ │ (web) │ │ (wiki) │ │
│ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │
│ └────────┬─────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ synthesize │ 综合分析搜索结果 │
│ └──────────────────┬──────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ analyze │ 深度分析 + 结构化整理 │
│ └──────────────────┬──────────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────┐ │
│ │ quality_check │ 质量检查 │
│ └─────────┬─────────┘ │
│ │ │
│ ┌────────┴────────┐ │
│ │ score >= 8? │ │
│ └────────┬────────┘ │
│ ┌───┴───┐ │
│ │是 否│ │
│ ▼ ▼ │
│ ┌─────────┐ ┌─────────────┐ │
│ │ generate │ │ revise │ │
│ │ report │ │ (loop) │ │
│ └────┬────┘ └──────┬──────┘ │
│ │ │ │
│ └──────┬───────┘ │
│ ▼ │
│ ┌─────────┐ │
│ │ output │ 生成最终报告 │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ END │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘5.2 完整代码实现
"""
多步研究 Agent 完整实现
实现：搜索 → 分析 → 总结 → 审核 → 输出
"""
from typing import TypedDict, Annotated, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
import os
# === 配置 ===
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key")
# === 1. 定义状态 ===
class ResearchState(TypedDict):
"""
研究 Agent 状态结构
包含完整的流程状态追踪
"""
# 用户输入
topic: str # 研究主题
# 搜索阶段
web_search_results: list = [] # 网络搜索结果
wiki_search_results: list = [] # 维基百科搜索结果
synthesized_info: str = "" # 综合信息
# 分析阶段
analysis: str = "" # 分析结果
key_points: list = [] # 关键要点
# 报告阶段
draft_report: str = "" # 初稿
final_report: str = "" # 最终报告
# 质量控制
quality_score: float = 0.0 # 质量评分
revision_count: int = 0 # 修改次数
max_revisions: int = 3 # 最大修改次数
# 元信息
messages: Annotated[list, add_messages] # 对话历史
# === 2. 初始化工具 ===
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
web_search = DuckDuckGoSearchRun()
wiki_search = WikipediaQueryRun()
# === 3. 定义节点 ===
def understand_topic(state: ResearchState) -> dict:
"""理解研究主题节点"""
topic = state["topic"]
prompt = f"""
请分析以下研究主题，明确研究目标和范围：
主题：{topic}
请输出：
1. 主题的核心问题
2. 研究的主要方向
3. 预期的结论类型
"""
response = llm.invoke(prompt)
return {"messages": [response]}
def search_web(state: ResearchState) -> dict:
"""网络搜索节点"""
topic = state["topic"]
try:
results = web_search.run(f"{topic} 最新研究")
return {"web_search_results": [results]}
except Exception as e:
return {"web_search_results": [f"搜索失败: {str(e)}"]}
def search_wiki(state: ResearchState) -> dict:
"""维基百科搜索节点"""
topic = state["topic"]
try:
results = wiki_search.run(topic)
return {"wiki_search_results": [results]}
except Exception as e:
return {"wiki_search_results": [f"搜索失败: {str(e)}"]}
def synthesize(state: ResearchState) -> dict:
"""综合搜索结果节点"""
topic = state["topic"]
web_results = "\n".join(state.get("web_search_results", []))
wiki_results = "\n".join(state.get("wiki_search_results", []))
prompt = f"""
请综合以下搜索结果，形成结构化的信息摘要：
主题：{topic}
网络搜索结果：
{web_results[:2000]}
维基百科结果：
{wiki_results[:2000]}
请提取：
1. 核心事实（3-5条）
2. 关键数据点
3. 主要观点
"""
response = llm.invoke(prompt)
return {
"synthesized_info": response.content,
"messages": [response]
}
def deep_analyze(state: ResearchState) -> dict:
"""深度分析节点"""
synthesized = state.get("synthesized_info", "")
prompt = f"""
基于以下综合信息，进行深度分析：
{synthesized}
请进行以下分析：
1. 原因分析（为什么会这样？）
2. 影响分析（带来了什么影响？）
3. 趋势分析（未来会如何发展？）
4. 风险与机遇
每个分析点给出具体的论据支持。
"""
response = llm.invoke(prompt)
# 提取关键要点
points_prompt = f"""
从以下分析中提取 5 个关键要点：
{response.content}
每个要点用一句话概括。
"""
points_response = llm.invoke(points_prompt)
key_points = [p.strip() for p in points_response.content.split('\n') if p.strip()]
return {
"analysis": response.content,
"key_points": key_points,
"messages": [response]
}
def quality_check(state: ResearchState) -> dict:
"""质量检查节点"""
analysis = state.get("analysis", "")
synthesized = state.get("synthesized_info", "")
prompt = f"""
请评估以下研究报告的质量：
综合信息：{synthesized[:500]}
分析内容：{analysis[:500]}
请从以下维度评分（每项 1-10 分）：
1. 信息完整性
2. 分析深度
3. 逻辑清晰度
4. 数据支撑
5. 实用性建议
最终给出平均分（1-10）。
只输出分数，格式如：8.5
"""
response = llm.invoke(prompt)
# 解析分数
try:
score_text = response.content.strip()
score = float(score_text)
# 确保分数在 1-10 范围内
score = max(1.0, min(10.0, score))
except:
score = 6.0 # 默认中等分数
return {"quality_score": score, "messages": [response]}
def generate_report(state: ResearchState) -> dict:
"""生成报告节点"""
topic = state["topic"]
analysis = state.get("analysis", "")
key_points = state.get("key_points", [])
prompt = f"""
请基于以下分析，撰写一份完整的研究报告：
主题：{topic}
关键要点：
{chr(10).join([f"- {p}" for p in key_points[:5]])}
详细分析：
{analysis[:1500]}
报告要求：
1. 结构清晰：引言 → 主体 → 结论
2. 字数：800-1000 字
3. 语言：专业但易懂
4. 包含可操作的建议
"""
response = llm.invoke(prompt)
return {
"draft_report": response.content,
"messages": [response]
}
def revise_report(state: ResearchState) -> dict:
"""修改报告节点"""
current_report = state.get("draft_report", "")
messages = state["messages"]
revision_count = state.get("revision_count", 0)
# 获取之前的质量反馈
feedback = ""
for msg in reversed(messages):
if hasattr(msg, 'content') and '评分' in msg.content:
feedback = msg.content
break
prompt = f"""
请根据以下反馈修改研究报告：
当前报告：
{current_report}
质量反馈：
{feedback}
这是第 {revision_count + 1} 次修改，请重点改进被指出不足的部分。
"""
response = llm.invoke(prompt)
return {
"draft_report": response.content,
"revision_count": revision_count + 1,
"messages": [response]
}
def final_output(state: ResearchState) -> dict:
"""最终输出节点"""
draft_report = state.get("draft_report", "")
quality_score = state.get("quality_score", 0)
revision_count = state.get("revision_count", 0)
prompt = f"""
请将以下研究报告润色并输出最终版本：
{draft_report}
添加：
1. 报告摘要
2. 参考来源说明
"""
response = llm.invoke(prompt)
return {
"final_report": response.content,
"messages": [response]
}
def route_after_quality(state: ResearchState) -> str:
"""质量检查后的路由"""
score = state.get("quality_score", 0)
revision_count = state.get("revision_count", 0)
max_revisions = state.get("max_revisions", 3)
# 如果分数 >= 8 或已达到最大修改次数，生成最终报告
if score >= 8.0 or revision_count >= max_revisions:
return "final_output"
# 否则进入修改流程
return "revise_report"
# === 4. 构建工作流 ===
def build_research_workflow():
"""构建研究 Agent 工作流"""
workflow = StateGraph(ResearchState)
# 注册节点
workflow.add_node("understand", understand_topic)
workflow.add_node("search_web", search_web)
workflow.add_node("search_wiki", search_wiki)
workflow.add_node("synthesize", synthesize)
workflow.add_node("analyze", deep_analyze)
workflow.add_node("quality_check", quality_check)
workflow.add_node("generate_report", generate_report)
workflow.add_node("revise_report", revise_report)
workflow.add_node("final_output", final_output)
# 定义边
workflow.add_edge(START, "understand")
workflow.add_edge("understand", "search_web")
workflow.add_edge("understand", "search_wiki")
# 搜索完成后综合
workflow.add_edge("search_web", "synthesize")
workflow.add_edge("search_wiki", "synthesize")
# 综合后分析
workflow.add_edge("synthesize", "analyze")
# 分析后生成报告
workflow.add_edge("analyze", "generate_report")
# 生成报告后检查质量
workflow.add_edge("generate_report", "quality_check")
# 条件边：质量检查后路由
workflow.add_conditional_edges(
source="quality_check",
path=route_after_quality,
path_map={
"revise_report": "revise_report",
"final_output": "final_output"
}
)
# 修改后重新检查质量
workflow.add_edge("revise_report", "quality_check")
# 最终输出后结束
workflow.add_edge("final_output", END)
return workflow.compile()
# === 5. 运行示例 ===
def main():
"""主函数"""
print("🔬 多步研究 Agent 已启动")
print("=" * 60)
# 构建工作流
app = build_research_workflow()
# 定义初始状态
initial_state = {
"topic": "大语言模型对软件工程的影响",
"web_search_results": [],
"wiki_search_results": [],
"synthesized_info": "",
"analysis": "",
"key_points": [],
"draft_report": "",
"final_report": "",
"quality_score": 0.0,
"revision_count": 0,
"max_revisions": 3,
"messages": []
}
# 执行工作流（使用 stream 模式查看中间状态）
print("\n📊 开始研究...\n")
config = {"recursion_limit": 100} # 设置最大递归深度
for step in app.stream(initial_state, config=config):
node_name = list(step.keys())[0]
print(f"📍 执行节点: {node_name}")
# 获取最终结果
final_result = app.invoke(initial_state, config=config)
# 输出报告
print("\n" + "=" * 60)
print("📄 研究报告")
print("=" * 60)
print(final_result["final_report"])
print(f"\n📈 质量评分: {final_result['quality_score']:.1f}/10")
print(f"🔄 修改次数: {final_result['revision_count']}")
print(f"📚 参考来源: 网络搜索 + 维基百科")
if __name__ == "__main__":
main()六、配置文件示例
6.1 YAML 配置
# research_agent_config.yaml
# 多步研究 Agent 配置文件
agent:
name: "Multi-Step Research Agent"
version: "1.0.0"
description: "支持搜索、分析、审核的完整研究工作流"
# LLM 配置
llm:
provider: "openai"
model: "gpt-4"
temperature: 0.7
max_tokens: 2000
# 如果使用 API 代理
# base_url: "https://api.openai.com/v1"
# 搜索工具配置
search:
web:
enabled: true
tool: "duckduckgo"
max_results: 5
wiki:
enabled: true
tool: "wikipedia"
max_results: 3
# 工作流配置
workflow:
# 质量阈值
quality:
threshold: 8.0 # 质量评分阈值
max_revisions: 3 # 最大修改次数
# 节点超时设置（秒）
timeouts:
search: 30
analyze: 60
generate: 45
# 并发设置
concurrency:
search_nodes: 2 # 并行搜索节点数
# 持久化配置
persistence:
enabled: true
backend: "sqlite" # sqlite / postgres / memory
path: "./checkpoints"
# postgres 配置（生产环境）
# connection_string: "postgresql://user:pass@host:5432/dbname"
# 日志配置
logging:
level: "INFO" # DEBUG / INFO / WARNING / ERROR
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: "./logs/research_agent.log"6.2 Python 配置加载
"""
配置加载器
使用 Pydantic 进行配置验证
"""
from pydantic import BaseModel, Field
from typing import Optional, Literal
import yaml
import os
class LLMConfig(BaseModel):
"""LLM 配置"""
provider: Literal["openai", "azure", "anthropic"] = "openai"
model: str = "gpt-4"
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
max_tokens: int = Field(default=2000, gt=0)
base_url: Optional[str] = None
api_key: Optional[str] = None
class SearchConfig(BaseModel):
"""搜索工具配置"""
web: dict = Field(default_factory=lambda: {"enabled": True, "tool": "duckduckgo", "max_results": 5})
wiki: dict = Field(default_factory=lambda: {"enabled": True, "tool": "wikipedia", "max_results": 3})
class WorkflowConfig(BaseModel):
"""工作流配置"""
quality_threshold: float = Field(default=8.0, ge=0.0, le=10.0)
max_revisions: int = Field(default=3, ge=1)
timeouts: dict = Field(default_factory=lambda: {"search": 30, "analyze": 60, "generate": 45})
concurrency: dict = Field(default_factory=lambda: {"search_nodes": 2})
class PersistenceConfig(BaseModel):
"""持久化配置"""
enabled: bool = True
backend: Literal["sqlite", "postgres", "memory"] = "sqlite"
path: str = "./checkpoints"
connection_string: Optional[str] = None
class LoggingConfig(BaseModel):
"""日志配置"""
level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: Optional[str] = None
class AgentConfig(BaseModel):
"""Agent 完整配置"""
agent: dict
llm: LLMConfig
search: SearchConfig
workflow: WorkflowConfig
persistence: PersistenceConfig
logging: LoggingConfig
def load_config(config_path: str = "research_agent_config.yaml") -> AgentConfig:
"""
从 YAML 文件加载配置
Args:
config_path: 配置文件路径
Returns:
AgentConfig 实例
"""
# 如果文件不存在，使用默认配置
if not os.path.exists(config_path):
return AgentConfig(
agent={"name": "Research Agent", "version": "1.0.0"},
llm=LLMConfig(),
search=SearchConfig(),
workflow=WorkflowConfig(),
persistence=PersistenceConfig(),
logging=LoggingConfig()
)
with open(config_path, "r", encoding="utf-8") as f:
raw_config = yaml.safe_load(f)
return AgentConfig(**raw_config)
# 使用示例
if __name__ == "__main__":
config = load_config()
print(f"Agent: {config.agent['name']}")
print(f"Model: {config.llm.model}")
print(f"Quality Threshold: {config.workflow.quality_threshold}")七、总结
7.1 核心要点回顾
本文深入剖析了 LangGraph StateGraph 的设计原理和实战技巧：
状态机是复杂 Agent 的正确姿势
线性链无法处理分支、循环、异常
状态机提供可控、可追溯、可组合的工作流
状态设计的三个原则
精简：只存储必要的字段
类型安全：使用 TypedDict/Pydantic
可扩展：预留足够的灵活性
节点是纯函数
输入是状态，输出是增量更新
保持节点职责单一
边的三种类型
静态边：固定跳转
条件边：动态决策
循环边：条件边的特殊应用
人工介入和持久化
Human-in-the-loop 实现关键节点的审批
Checkpoint 机制支持断点续传和故障恢复
7.2 架构图回顾
┌─────────────────────────────────────────────────────────────────────┐
│ LangGraph StateGraph 设计模式总结 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ State（状态） │
│ ├─ 定义全局上下文 │
│ ├─ TypedDict/Pydantic 建模 │
│ └─ 通过节点增量更新 │
│ │
│ Node（节点） │
│ ├─ 业务逻辑执行单元 │
│ ├─ 输入 State，输出更新 │
│ └─ 单一职责原则 │
│ │
│ Edge（边） │
│ ├─ 静态边：add_edge() │
│ ├─ 条件边：add_conditional_edges() │
│ └─ 循环：条件边指向自身或其他节点 │
│ │
│ 工作流 = State + Nodes + Edges │
│ │
│ 进阶特性 │
│ ├─ Human-in-the-loop：关键节点人工审批 │
│ ├─ Checkpoint：状态持久化与恢复 │
│ └─ 并行执行：多个节点同时运行 │
│ │
└─────────────────────────────────────────────────────────────────────┘7.3 下一步学习建议
深入官方文档：LangGraph 官方提供了详尽的教程和示例
阅读源码：理解 StateGraph 的内部实现机制
实践项目：选择一个复杂场景，从头实现状态机
性能优化：学习如何优化大状态、减少内存占用
参考资料
本文首发于 2026 年 3 月 1 日，风格对标 PySuper（zhengxingtao.com）
评论区