LangGraph 复杂状态机设计模式
系统掌握 LangGraph 中的状态机设计模式，从基础模式到高级组合，提供可复用的代码模板和最佳实践。
LangGraph 复杂状态机设计模式
1. 引言：为什么需要状态机设计模式
1.1 复杂工作流的挑战
在前一篇文章《LangChain 与 LangGraph 深度解析》中，我们介绍了 LangGraph 的基本概念和用法。然而，当面对真实世界的复杂应用时，仅仅知道 API 是远远不够的。开发者经常面临以下挑战：
- 工作流复杂度爆炸 — 随着功能增加，状态机图变得难以理解和维护
- 重复造轮子 — 相似的逻辑在不同项目中重复实现
- 设计决策困难 — 面对复杂需求，不知道如何组织状态和转换
- 调试困难 — 状态流转不清晰，问题难以定位
- 协作成本 — 团队成员对状态机设计缺乏统一理解
1.2 设计模式的价值
设计模式是软件工程的宝贵资产，它们提供了：
✅ 经过验证的解决方案 — 解决常见问题的最佳实践
✅ 通用语言 — 团队间高效沟通的基础
✅ 可复用模板 — 快速构建可靠系统
✅ 避免陷阱 — 前人踩过的坑，后人不必再踩
在 LangGraph 中，设计模式同样重要。通过掌握常见模式，你可以：
- 快速搭建复杂工作流
- 提高代码可维护性
- 降低调试成本
- 与社区高效交流
1.3 LangGraph 的模式支持
LangGraph 的设计天然适合模式化：
from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Literal, Annotated
# 状态定义 — 模式的基础
class WorkflowState(TypedDict):
input: str
output: str
metadata: dict
# 节点 — 模式的原子单元
def node_function(state: WorkflowState) -> WorkflowState:
return {"output": "result"}
# 边 — 模式的连接方式
workflow = StateGraph(WorkflowState)
workflow.add_node("step", node_function)
workflow.add_edge(START, "step")
workflow.add_edge("step", END)
# 编译 — 模式的实例化
app = workflow.compile()
LangGraph 的核心抽象（状态、节点、边、条件）为设计模式提供了坚实基础。
1.4 本文结构
本文将系统介绍 LangGraph 中的设计模式，按复杂度递增组织：
- 基础模式 — 顺序、分支、并行
- 循环模式 — 迭代、重试
- 协作模式 — 多 Agent 架构
- 人机协作模式 — 人类审核与干预
- 高级模式 — 模式组合与嵌套
- 反模式与最佳实践 — 避免常见陷阱
每个模式都包含：
- 模式描述与适用场景
- 完整代码实现
- 可视化图结构
- 变体与扩展建议
让我们开始探索这些强大的设计模式！
2. 基础模式：顺序与分支
2.1 线性流水线模式
模式描述：最简单的模式，节点按固定顺序依次执行，无分支无循环。
适用场景：
- 数据预处理流水线
- 多步骤内容生成
- 顺序验证流程
核心特征：
- 节点顺序固定
- 无条件判断
- 状态单向流动
代码实现：
from langgraph.graph import StateGraph, END, START
from typing import TypedDict
# 定义状态
class PipelineState(TypedDict):
raw_data: str
cleaned_data: str
analyzed_data: str
final_output: str
# 节点 1：数据清洗
def clean_data(state: PipelineState) -> PipelineState:
"""清洗原始数据"""
cleaned = state["raw_data"].strip().lower()
return {"cleaned_data": cleaned}
# 节点 2：数据分析
def analyze_data(state: PipelineState) -> PipelineState:
"""分析清洗后的数据"""
words = state["cleaned_data"].split()
analysis = f"共 {len(words)} 个单词"
return {"analyzed_data": analysis}
# 节点 3：生成输出
def generate_output(state: PipelineState) -> PipelineState:
"""生成最终输出"""
output = f"原始：{state['cleaned_data']}\n分析：{state['analyzed_data']}"
return {"final_output": output}
# 构建流水线
workflow = StateGraph(PipelineState)
# 添加节点
workflow.add_node("clean", clean_data)
workflow.add_node("analyze", analyze_data)
workflow.add_node("generate", generate_output)
# 设置顺序边
workflow.add_edge(START, "clean")
workflow.add_edge("clean", "analyze")
workflow.add_edge("analyze", "generate")
workflow.add_edge("generate", END)
# 编译
pipeline = workflow.compile()
# 执行
result = pipeline.invoke({"raw_data": " Hello World "})
print(result["final_output"])
可视化：
START → clean → analyze → generate → END
变体：
- 带日志的流水线：每个节点记录执行日志
- 带进度的流水线：状态中包含进度信息
- 可中断流水线：支持在任意节点暂停
2.2 条件分支模式
模式描述：根据状态内容决定下一步执行路径，实现条件逻辑。
适用场景：
- 基于输入类型的处理
- 质量检查与路由
- A/B 测试分流
核心特征：
- 条件函数决定路径
- 多个可能的分支
- 分支可能汇合
代码实现：
from typing import Literal
class RouterState(TypedDict):
input_type: str
content: str
result: str
# 条件函数：根据输入类型路由
def route_by_type(state: RouterState) -> Literal["text_processor", "image_processor", "error_handler"]:
"""根据输入类型选择处理路径"""
input_type = state["input_type"].lower()
if input_type == "text":
return "text_processor"
elif input_type == "image":
return "image_processor"
else:
return "error_handler"
# 分支 1：文本处理
def process_text(state: RouterState) -> RouterState:
"""处理文本内容"""
result = f"文本处理：{len(state['content'])} 个字符"
return {"result": result}
# 分支 2：图像处理
def process_image(state: RouterState) -> RouterState:
"""处理图像内容"""
result = f"图像处理：分析 {state['content']}"
return {"result": result}
# 分支 3：错误处理
def handle_error(state: RouterState) -> RouterState:
"""处理未知类型"""
result = f"错误：不支持的类型 {state['input_type']}"
return {"result": result}
# 构建路由图
workflow = StateGraph(RouterState)
# 添加节点
workflow.add_node("text_processor", process_text)
workflow.add_node("image_processor", process_image)
workflow.add_node("error_handler", handle_error)
# 设置入口和条件边
workflow.add_edge(START, "router")
workflow.add_conditional_edges(
"router", # 从哪个节点出发
route_by_type, # 条件函数
{
"text_processor": "text_processor",
"image_processor": "image_processor",
"error_handler": "error_handler"
}
)
# 所有分支汇合到 END
workflow.add_edge("text_processor", END)
workflow.add_edge("image_processor", END)
workflow.add_edge("error_handler", END)
# 编译
router = workflow.compile()
# 测试不同路径
print(router.invoke({"input_type": "text", "content": "Hello"}))
print(router.invoke({"input_type": "image", "content": "image_url"}))
print(router.invoke({"input_type": "audio", "content": "audio_data"}))
可视化：
→ text_processor →
↘
START → router → image_processor → END
↘
→ error_handler →
高级技巧：
多层级路由：
def primary_router(state: RouterState) -> Literal["category_a", "category_b"]:
"""一级路由"""
return "category_a" if len(state["content"]) > 100 else "category_b"
def secondary_router(state: RouterState) -> Literal["fast_path", "thorough_path"]:
"""二级路由"""
return "fast_path" if state.get("urgent") else "thorough_path"
# 嵌套条件边
workflow.add_conditional_edges("entry", primary_router, {...})
workflow.add_conditional_edges("category_a", secondary_router, {...})
动态条件：
def dynamic_threshold(state: RouterState) -> Literal["approve", "review"]:
"""基于动态阈值的条件"""
threshold = state.get("threshold", 0.5)
score = state.get("confidence_score", 0)
return "approve" if score >= threshold else "review"
2.3 并行执行模式
模式描述：多个节点同时执行，提高处理效率。
适用场景：
- 多源数据收集
- 并行特征提取
- 多模型投票
核心特征：
- 节点并发执行
- 结果需要合并
- 可能涉及同步等待
代码实现：
import asyncio
from typing import List
class ParallelState(TypedDict):
query: str
search_results: List[str]
analysis_results: List[str]
combined_output: str
# 并行任务 1：网络搜索
async def web_search(state: ParallelState) -> ParallelState:
"""执行网络搜索"""
await asyncio.sleep(0.5) # 模拟异步操作
return {"search_results": [f"搜索结果 {i}" for i in range(3)]}
# 并行任务 2：知识库查询
async def knowledge_base_query(state: ParallelState) -> ParallelState:
"""查询知识库"""
await asyncio.sleep(0.3)
return {"analysis_results": [f"知识条目 {i}" for i in range(2)]}
# 合并结果
def merge_results(state: ParallelState) -> ParallelState:
"""合并并行结果"""
combined = {
"search": state.get("search_results", []),
"knowledge": state.get("analysis_results", [])
}
output = f"搜索：{len(combined['search'])}条，知识：{len(combined['knowledge'])}条"
return {"combined_output": output}
# 构建并行图
workflow = StateGraph(ParallelState)
# 添加并行节点
workflow.add_node("search", web_search)
workflow.add_node("knowledge", knowledge_base_query)
workflow.add_node("merge", merge_results)
# 并行分支
workflow.add_edge(START, "search")
workflow.add_edge(START, "knowledge")
# 汇合点
workflow.add_edge("search", "merge")
workflow.add_edge("knowledge", "merge")
workflow.add_edge("merge", END)
# 编译
parallel_app = workflow.compile()
# 执行
result = parallel_app.invoke({"query": "人工智能"})
print(result["combined_output"])
可视化：
→ search →
↘ \
START → merge → END
/ /
→ knowledge →
注意事项：
- 并行节点必须异步（async def）才能真正并发
- 合并节点需要等待所有并行分支完成
- 状态合并策略需要精心设计（覆盖、合并、选择）
状态合并策略：
from typing import Annotated
def merge_lists(left: List[str], right: List[str]) -> List[str]:
"""合并两个列表"""
return left + right
class MergedState(TypedDict):
results: Annotated[List[str], merge_lists]
3. 循环模式：迭代与重试
3.1 固定次数迭代模式
模式描述：节点执行固定次数，每次迭代更新状态。
适用场景：
- 渐进式内容优化
- 多轮推理
- 分步细化
核心特征：
- 计数器控制迭代
- 状态累积
- 确定性终止
代码实现：
from typing import Literal
class IterationState(TypedDict):
prompt: str
current_output: str
iteration: int
max_iterations: int
history: list
def generate_content(state: IterationState) -> IterationState:
"""生成或优化内容"""
# 模拟 LLM 生成
if state["iteration"] == 0:
output = f"初始版本：{state['prompt']}"
else:
output = f"优化版本 {state['iteration']}: 基于 '{state['current_output']}' 改进"
return {
"current_output": output,
"history": state.get("history", []) + [output]
}
def check_iterations(state: IterationState) -> Literal["continue", "end"]:
"""检查是否继续迭代"""
if state["iteration"] < state["max_iterations"]:
return "continue"
return "end"
# 构建迭代图
workflow = StateGraph(IterationState)
workflow.add_node("generate", generate_content)
workflow.add_edge(START, "generate")
workflow.add_conditional_edges(
"generate",
check_iterations,
{
"continue": "generate", # 循环回自身
"end": END
}
)
# 在 generate 节点后自动增加迭代计数
# 实际使用中可以通过状态更新器实现
iteration_app = workflow.compile()
result = iteration_app.invoke({
"prompt": "写一篇关于 AI 的文章",
"current_output": "",
"iteration": 0,
"max_iterations": 3,
"history": []
})
print(f"最终输出：{result['current_output']}")
print(f"迭代历史：{result['history']}")
可视化：
┌─────────────┐
↓ │
START → generate ────┘ (最多 N 次)
↓
END
3.2 条件终止循环模式
模式描述：循环执行直到满足特定条件，而非固定次数。
适用场景：
- 直到答案满意
- 直到收敛
- 直到达到质量标准
核心特征：
- 条件函数判断终止
- 可能需要最大次数保护
- 状态质量评估
代码实现：
class ConvergenceState(TypedDict):
query: str
answer: str
quality_score: float
threshold: float
iteration: int
max_iterations: int
def refine_answer(state: ConvergenceState) -> ConvergenceState:
"""优化答案"""
# 模拟答案优化和质量评估
iteration = state["iteration"]
# 模拟质量分数逐渐提高
quality_score = min(0.95, 0.5 + iteration * 0.15)
answer = f"优化后的答案 (质量：{quality_score:.2f})"
return {
"answer": answer,
"quality_score": quality_score,
"iteration": iteration + 1
}
def should_continue_refining(state: ConvergenceState) -> Literal["refine", "done"]:
"""判断是否继续优化"""
# 达到质量阈值或最大迭代次数则停止
if state["quality_score"] >= state["threshold"]:
return "done"
if state["iteration"] >= state["max_iterations"]:
return "done"
return "refine"
# 构建条件终止图
workflow = StateGraph(ConvergenceState)
workflow.add_node("refine", refine_answer)
workflow.add_edge(START, "refine")
workflow.add_conditional_edges(
"refine",
should_continue_refining,
{
"refine": "refine",
"done": END
}
)
convergence_app = workflow.compile()
result = convergence_app.invoke({
"query": "量子计算原理",
"answer": "",
"quality_score": 0.0,
"threshold": 0.85,
"iteration": 0,
"max_iterations": 10
})
print(f"最终答案：{result['answer']}")
print(f"质量分数：{result['quality_score']:.2f}")
print(f"迭代次数：{result['iteration']}")
3.3 带退避的重试模式
模式描述：操作失败时自动重试，使用退避策略避免频繁失败。
适用场景：
- API 调用失败重试
- 网络请求重试
- 不稳定操作容错
核心特征：
- 失败检测
- 退避策略（线性、指数）
- 最大重试次数
代码实现：
import time
import random
from typing import Literal
class RetryState(TypedDict):
task: str
result: str
attempt: int
max_attempts: int
base_delay: float
errors: list
def execute_task(state: RetryState) -> RetryState:
"""执行可能失败的任务"""
attempt = state["attempt"]
# 模拟随机失败（前两次失败，第三次成功）
if attempt < 2:
error = f"第 {attempt} 次尝试失败：临时错误"
return {
"errors": state.get("errors", []) + [error],
"result": None
}
else:
return {
"result": f"任务成功完成 (第 {attempt} 次尝试)",
"errors": state.get("errors", [])
}
def check_result(state: RetryState) -> Literal["retry", "success", "give_up"]:
"""检查结果并决定下一步"""
if state["result"] is not None:
return "success"
if state["attempt"] >= state["max_attempts"]:
return "give_up"
return "retry"
def apply_backoff(state: RetryState) -> RetryState:
"""应用退避延迟"""
# 指数退避：delay = base_delay * 2^(attempt-1)
delay = state["base_delay"] * (2 ** (state["attempt"] - 1))
# 添加随机抖动
jitter = delay * 0.2 * (random.random() - 0.5)
actual_delay = delay + jitter
print(f"第 {state['attempt']} 次失败，{actual_delay:.2f}秒后重试...")
# time.sleep(actual_delay) # 实际使用取消注释
return {"attempt": state["attempt"] + 1}
# 构建重试图
workflow = StateGraph(RetryState)
workflow.add_node("execute", execute_task)
workflow.add_node("backoff", apply_backoff)
workflow.add_edge(START, "execute")
workflow.add_conditional_edges(
"execute",
check_result,
{
"retry": "backoff",
"success": END,
"give_up": END
}
)
workflow.add_edge("backoff", "execute")
retry_app = workflow.compile()
result = retry_app.invoke({
"task": "调用外部 API",
"result": None,
"attempt": 1,
"max_attempts": 5,
"base_delay": 1.0,
"errors": []
})
print(f"最终结果：{result['result']}")
print(f"失败记录：{result['errors']}")
退避策略变体：
# 线性退避
delay = state["base_delay"] * state["attempt"]
# 指数退避（推荐）
delay = state["base_delay"] * (2 ** state["attempt"])
# 斐波那契退避
def fibonacci(n):
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
delay = state["base_delay"] * fibonacci(state["attempt"])
4. 协作模式：多 Agent 架构
4.1 主管 - 工人模式（Manager-Worker）
模式描述：主管 Agent 负责任务分解和协调，工人 Agent 负责具体执行。
适用场景：
- 复杂任务分解
- 并行任务分配
- 结果整合
核心特征：
- 分层架构
- 任务分发
- 结果汇总
代码实现：
from typing import List, Literal
class ManagerWorkerState(TypedDict):
task: str
subtasks: List[str]
results: List[str]
current_subtask: int
final_output: str
def manager_decompose(state: ManagerWorkerState) -> ManagerWorkerState:
"""主管：分解任务"""
task = state["task"]
# 模拟任务分解
subtasks = [
f"子任务 1: 研究 {task} 的背景",
f"子任务 2: 分析 {task} 的核心问题",
f"子任务 3: 提出 {task} 的解决方案"
]
return {"subtasks": subtasks, "current_subtask": 0}
def worker_execute(state: ManagerWorkerState) -> ManagerWorkerState:
"""工人：执行子任务"""
current = state["current_subtask"]
subtask = state["subtasks"][current]
# 模拟执行
result = f"[完成] {subtask}\n详细内容：..."
results = state.get("results", []) + [result]
return {
"results": results,
"current_subtask": current + 1
}
def check_all_subtasks_done(state: ManagerWorkerState) -> Literal["continue", "aggregate"]:
"""检查是否所有子任务完成"""
if state["current_subtask"] < len(state["subtasks"]):
return "continue"
return "aggregate"
def manager_aggregate(state: ManagerWorkerState) -> ManagerWorkerState:
"""主管：汇总结果"""
results = state["results"]
final_output = "=== 任务执行报告 ===\n\n"
for i, result in enumerate(results, 1):
final_output += f"{i}. {result}\n\n"
final_output += "=== 总结 ===\n所有子任务已完成。"
return {"final_output": final_output}
# 构建主管 - 工人图
workflow = StateGraph(ManagerWorkerState)
workflow.add_node("manager_decompose", manager_decompose)
workflow.add_node("worker_execute", worker_execute)
workflow.add_node("manager_aggregate", manager_aggregate)
workflow.add_edge(START, "manager_decompose")
workflow.add_edge("manager_decompose", "worker_execute")
workflow.add_conditional_edges(
"worker_execute",
check_all_subtasks_done,
{
"continue": "worker_execute",
"aggregate": "manager_aggregate"
}
)
workflow.add_edge("manager_aggregate", END)
mw_app = workflow.compile()
result = mw_app.invoke({
"task": "设计一个推荐系统",
"subtasks": [],
"results": [],
"current_subtask": 0,
"final_output": ""
})
print(result["final_output"])
可视化：
START → manager_decompose → worker_execute ──(循环)──┐
↓ │
check ──→│ │
↓ │
manager_aggregate ←─┘
↓
END
4.2 投票与共识模式
模式描述：多个 Agent 独立给出答案，通过投票或共识机制确定最终结果。
适用场景：
- 提高答案准确性
- 减少单个模型偏差
- 重要决策
核心特征：
- 并行独立评估
- 投票/共识机制
- 可能包含仲裁
代码实现：
from typing import List, Literal
class VotingState(TypedDict):
question: str
votes: List[str]
current_voter: int
total_voters: int
final_answer: str
def voter_answer(state: VotingState) -> VotingState:
"""投票者：给出答案"""
voter_id = state["current_voter"]
# 模拟不同投票者的答案
answers = {
0: "答案 A: 基于方法 1 的分析",
1: "答案 A: 基于方法 2 的分析",
2: "答案 B: 基于方法 3 的分析"
}
answer = answers.get(voter_id, f"答案：投票者 {voter_id}")
votes = state.get("votes", []) + [answer]
return {
"votes": votes,
"current_voter": voter_id + 1
}
def tally_votes(state: VotingState) -> VotingState:
"""统计票数"""
votes = state["votes"]
# 简单统计（实际应使用更复杂的语义相似度）
vote_counts = {}
for vote in votes:
answer_type = vote.split(":")[0] # 提取答案类型
vote_counts[answer_type] = vote_counts.get(answer_type, 0) + 1
# 找出得票最多的答案
winner = max(vote_counts, key=vote_counts.get)
final_answer = f"{winner} (得票：{vote_counts[winner]}/{len(votes)})"
return {"final_answer": final_answer, "vote_counts": vote_counts}
# 构建投票图
workflow = StateGraph(VotingState)
workflow.add_node("voter", voter_answer)
workflow.add_node("tally", tally_votes)
workflow.add_edge(START, "voter")
workflow.add_conditional_edges(
"voter",
lambda s: "tally" if s["current_voter"] >= s["total_voters"] else "voter",
{
"voter": "voter",
"tally": "tally"
}
)
workflow.add_edge("tally", END)
voting_app = workflow.compile()
result = voting_app.invoke({
"question": "哪种算法最适合此场景？",
"votes": [],
"current_voter": 0,
"total_voters": 3,
"final_answer": ""
})
print(f"最终答案：{result['final_answer']}")
print(f"投票统计：{result.get('vote_counts', {})}")
4.3 流水线协作模式
模式描述：多个 Agent 按顺序协作，每个负责特定阶段。
适用场景：
- 内容创作流水线
- 代码开发流程
- 数据分析流程
核心特征：
- 阶段划分清晰
- 状态传递
- 专业化分工
代码实现：
class PipelineState(TypedDict):
topic: str
research: str
outline: str
draft: str
review: str
final_article: str
def researcher(state: PipelineState) -> PipelineState:
"""研究阶段：收集信息"""
research = f"""
关于 {state['topic']} 的研究结果:
- 关键点 1: ...
- 关键点 2: ...
- 数据支持：...
"""
return {"research": research}
def outliner(state: PipelineState) -> PipelineState:
"""大纲阶段：组织结构"""
outline = f"""
文章大纲 - {state['topic']}:
1. 引言
2. 核心概念
3. 案例分析
4. 最佳实践
5. 总结
"""
return {"outline": outline}
def writer(state: PipelineState) -> PipelineState:
"""写作阶段：生成草稿"""
draft = f"""
# {state['topic']}
## 引言
基于研究：{state['research'][:50]}...
## 正文
按照大纲展开...
## 总结
核心观点总结...
"""
return {"draft": draft}
def reviewer(state: PipelineState) -> PipelineState:
"""评审阶段：质量检查"""
review = """
评审意见:
✅ 结构清晰
✅ 论据充分
⚠️ 建议增加案例
✅ 语言流畅
评分：85/100
"""
return {"review": review}
def finalizer(state: PipelineState) -> PipelineState:
"""定稿阶段：整合输出"""
final_article = f"""
{state['draft']}
---
评审反馈：{state['review']}
"""
return {"final_article": final_article}
# 构建流水线协作图
workflow = StateGraph(PipelineState)
workflow.add_node("researcher", researcher)
workflow.add_node("outliner", outliner)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
workflow.add_node("finalizer", finalizer)
# 顺序连接
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "outliner")
workflow.add_edge("outliner", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_edge("reviewer", "finalizer")
workflow.add_edge("finalizer", END)
pipeline_app = workflow.compile()
result = pipeline_app.invoke({"topic": "机器学习入门", "research": "", "outline": "", "draft": "", "review": "", "final_article": ""})
print(result["final_article"])
5. 人机协作模式
5.1 人类审核节点
模式描述：在关键节点暂停执行，等待人类审核或决策。
适用场景：
- 内容发布前审核
- 高风险决策确认
- 质量把关
核心特征：
- 中断执行流
- 等待人类输入
- 恢复执行
代码实现：
from langgraph.graph import interrupt
from typing import Literal
class HumanInTheLoopState(TypedDict):
content: str
human_approved: bool
human_feedback: str
iteration: int
def generate_content(state: HumanInTheLoopState) -> HumanInTheLoopState:
"""生成内容"""
if state["iteration"] == 0:
content = "初始版本：这是一篇测试文章..."
else:
content = f"修订版本 {state['iteration']}: 根据反馈 '{state['human_feedback']}' 修改..."
return {"content": content}
def human_review(state: HumanInTheLoopState) -> HumanInTheLoopState:
"""人类审核节点"""
print(f"\n=== 等待人类审核 ===")
print(f"当前内容：{state['content']}")
# 中断执行，等待人类输入
# 实际使用中，这里会暂停并等待外部输入
approval = input("是否批准？(y/n/r 修订): ")
if approval == 'y':
return {"human_approved": True, "human_feedback": ""}
elif approval == 'n':
return {"human_approved": False, "human_feedback": ""}
else:
feedback = input("提供修订意见：")
return {"human_approved": False, "human_feedback": feedback}
def decide_next(state: HumanInTheLoopState) -> Literal["approve", "revise", "reject"]:
"""决定下一步"""
if state["human_approved"]:
return "approve"
elif state["iteration"] >= 3:
return "reject" # 达到最大修订次数
return "revise"
# 构建人机协作图
workflow = StateGraph(HumanInTheLoopState)
workflow.add_node("generate", generate_content)
workflow.add_node("review", human_review)
workflow.add_edge(START, "generate")
workflow.add_edge("generate", "review")
workflow.add_conditional_edges(
"review",
decide_next,
{
"approve": END,
"revise": "generate",
"reject": END
}
)
# 编译时设置中断点
app = workflow.compile(interrupt_before=["review"])
# 执行
result = app.invoke({
"content": "",
"human_approved": False,
"human_feedback": "",
"iteration": 0
})
print(f"\n最终结果：{result['content']}")
使用 interrupt 工具：
from langgraph.graph import interrupt
def human_approval_node(state):
"""使用 interrupt 工具"""
# 抛出中断，暂停执行
human_decision = interrupt({
"content": state["content"],
"question": "是否批准此内容？"
})
return {
"human_approved": human_decision.get("approved", False),
"human_feedback": human_decision.get("feedback", "")
}
5.2 中断与恢复
模式描述：支持在任意点暂停和恢复执行，保持状态持久化。
核心特征：
- 状态检查点
- 线程管理
- 断点续传
代码实现：
from langgraph.checkpoint.memory import MemorySaver
# 创建检查点存储器
checkpointer = MemorySaver()
# 编译时传入
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
# 创建线程
config = {"configurable": {"thread_id": "conversation_123"}}
# 执行到中断点
result = app.invoke(
{"content": "初始内容", "iteration": 0},
config
)
# 此时暂停，等待人类输入
# 恢复执行，传入人类决策
result = app.invoke(
{"human_approved": True, "human_feedback": ""},
config
)
# 从断点继续执行
多线程管理：
# 不同用户使用不同线程
user1_config = {"configurable": {"thread_id": "user_001"}}
user2_config = {"configurable": {"thread_id": "user_002"}}
# 独立执行，互不干扰
app.invoke({"input": "user1 data"}, user1_config)
app.invoke({"input": "user2 data"}, user2_config)
5.3 反馈循环
模式描述：人类反馈直接影响后续生成，形成闭环优化。
代码实现：
class FeedbackLoopState(TypedDict):
prompt: str
output: str
feedback: str
quality_score: float
iterations: int
def generate(state: FeedbackLoopState) -> FeedbackLoopState:
"""生成内容"""
# 根据反馈调整生成
context = state["feedback"] if state["feedback"] else "无反馈"
output = f"生成内容 (基于反馈：{context})"
return {"output": output}
def get_feedback(state: FeedbackLoopState) -> FeedbackLoopState:
"""获取人类反馈"""
# 模拟反馈（实际为真实人类输入）
feedback = "建议增加更多细节"
score = 0.8
return {
"feedback": feedback,
"quality_score": score
}
def should_improve(state: FeedbackLoopState) -> Literal["improve", "done"]:
"""判断是否需要改进"""
if state["quality_score"] >= 0.9:
return "done"
if state["iterations"] >= 5:
return "done"
return "improve"
workflow = StateGraph(FeedbackLoopState)
workflow.add_node("generate", generate)
workflow.add_node("feedback", get_feedback)
workflow.add_edge(START, "generate")
workflow.add_edge("generate", "feedback")
workflow.add_conditional_edges(
"feedback",
should_improve,
{
"improve": "generate",
"done": END
}
)
feedback_app = workflow.compile()
result = feedback_app.invoke({
"prompt": "写一篇文章",
"output": "",
"feedback": "",
"quality_score": 0.0,
"iterations": 0
})
print(f"最终输出：{result['output']}")
print(f"质量分数：{result['quality_score']}")
6. 高级模式：模式组合
6.1 嵌套状态机
模式描述：状态机中包含子状态机，处理复杂分层逻辑。
适用场景：
- 复杂任务分层
- 模块化设计
- 代码复用
代码实现：
# 子图：内容生成模块
def create_content_subgraph():
"""创建内容生成子图"""
class ContentState(TypedDict):
topic: str
draft: str
refined: str
def write_draft(state: ContentState) -> ContentState:
return {"draft": f"草稿：{state['topic']}"}
def refine_draft(state: ContentState) -> ContentState:
return {"refined": f"精修：{state['draft']}"}
subgraph = StateGraph(ContentState)
subgraph.add_node("write", write_draft)
subgraph.add_node("refine", refine_draft)
subgraph.add_edge(START, "write")
subgraph.add_edge("write", "refine")
subgraph.add_edge("refine", END)
return subgraph.compile()
# 主图：使用子图
class MainState(TypedDict):
topic: str
content: str
reviewed: bool
def prepare_topic(state: MainState) -> MainState:
"""准备主题"""
return {"topic": state["topic"].upper()}
def review_content(state: MainState) -> MainState:
"""审核内容"""
return {"reviewed": True}
main_workflow = StateGraph(MainState)
# 添加子图为节点
content_subgraph = create_content_subgraph()
main_workflow.add_node("content_generator", content_subgraph)
main_workflow.add_node("prepare", prepare_topic)
main_workflow.add_node("review", review_content)
main_workflow.add_edge(START, "prepare")
main_workflow.add_edge("prepare", "content_generator")
main_workflow.add_edge("content_generator", "review")
main_workflow.add_edge("review", END)
main_app = main_workflow.compile()
result = main_app.invoke({"topic": "人工智能", "content": "", "reviewed": False})
print(f"结果：{result}")
6.2 模式混合实践
模式描述：在实际应用中组合多种模式。
综合案例：智能内容生产系统
class ContentSystemState(TypedDict):
topic: str
research_results: list
draft: str
human_approved: bool
feedback: str
iteration: int
final_content: str
# 1. 并行研究（并行模式）
async def parallel_research(state: ContentSystemState) -> ContentSystemState:
tasks = [
search_web(state["topic"]),
query_knowledge_base(state["topic"]),
analyze_trends(state["topic"])
]
results = await asyncio.gather(*tasks)
return {"research_results": results}
# 2. 迭代生成（循环模式）
def iterative_generation(state: ContentSystemState) -> ContentSystemState:
context = "\n".join(state["research_results"])
feedback = state.get("feedback", "")
draft = f"基于研究和反馈 {feedback} 生成的内容..."
return {"draft": draft}
# 3. 人类审核（人机协作模式）
def human_approval(state: ContentSystemState) -> ContentSystemState:
approval = interrupt({"draft": state["draft"]})
return {
"human_approved": approval.get("approved", False),
"feedback": approval.get("feedback", "")
}
# 4. 多模式组合
workflow = StateGraph(ContentSystemState)
workflow.add_node("research", parallel_research)
workflow.add_node("generate", iterative_generation)
workflow.add_node("approve", human_approval)
workflow.add_edge(START, "research")
workflow.add_edge("research", "generate")
workflow.add_edge("generate", "approve")
workflow.add_conditional_edges(
"approve",
lambda s: "generate" if not s["human_approved"] else "end",
{"generate": "generate", "end": END}
)
system_app = workflow.compile(interrupt_before=["approve"])
7. 反模式与最佳实践
7.1 常见设计陷阱
陷阱 1：状态爆炸
# ❌ 反模式：状态包含过多字段
class BadState(TypedDict):
field1: str
field2: str
# ... 50+ 字段
field50: str
# ✅ 最佳实践：状态模块化
class ResearchState(TypedDict):
query: str
results: list
class WritingState(TypedDict):
research: ResearchState
draft: str
陷阱 2：循环依赖
# ❌ 反模式：节点 A 依赖 B，B 依赖 A
def node_a(state):
return node_b(state) # 直接调用其他节点
# ✅ 最佳实践：通过状态传递数据
def node_a(state):
return {"data": process(state["input"])}
def node_b(state):
return {"result": use(state["data"])}
陷阱 3：忽略错误处理
# ❌ 反模式：没有错误处理
def risky_node(state):
result = might_fail()
return {"result": result}
# ✅ 最佳实践：try-catch 或重试模式
def safe_node(state):
try:
result = might_fail()
except Exception as e:
return {"error": str(e), "result": None}
return {"result": result}
7.2 性能优化建议
1. 并行化独立任务
# 使用并行模式加速独立操作
async def parallel_processing(state):
results = await asyncio.gather(
task1(state["data"]),
task2(state["data"]),
task3(state["data"])
)
return {"results": results}
2. 缓存中间结果
from functools import lru_cache
@lru_cache(maxsize=128)
def expensive_computation(input_data):
return compute(input_data)
3. 流式处理大数据
def stream_processor(state):
for chunk in state["large_data"]:
yield process(chunk)
7.3 可维护性原则
1. 单一职责
每个节点只做一件事，便于测试和复用。
2. 明确的状态契约
class State(TypedDict):
"""
状态说明：
- input: 用户输入（只读）
- output: 节点输出（写入）
- metadata: 元数据（可选）
"""
input: str
output: str
metadata: dict
3. 可视化文档
# 生成可视化图
app.get_graph().draw_mermaid_png(output_path="workflow.png")
8. 实战案例：完整工作流构建
8.1 需求分析
场景：构建一个智能客服系统
需求：
- 理解用户意图
- 检索相关知识
- 生成回答
- 质量检查
- 人类审核（必要时）
- 发送回答
8.2 模式选择
- 条件分支：意图识别路由
- 并行执行：多源知识检索
- 循环模式：回答优化
- 人机协作：复杂问题审核
8.3 代码实现
class CustomerServiceState(TypedDict):
user_query: str
intent: str
knowledge: list
draft_answer: str
quality_score: float
human_reviewed: bool
final_answer: str
# 意图识别
def classify_intent(state: CustomerServiceState) -> CustomerServiceState:
# 模拟意图分类
intent = "technical_support" # 实际使用 LLM
return {"intent": intent}
# 并行知识检索
async def retrieve_knowledge(state: CustomerServiceState) -> CustomerServiceState:
results = await asyncio.gather(
search_faq(state["user_query"]),
search_docs(state["intent"]),
query_database(state["user_query"])
)
return {"knowledge": [r for r in results if r]}
# 生成回答
def generate_answer(state: CustomerServiceState) -> CustomerServiceState:
context = "\n".join(state["knowledge"])
answer = f"基于 {len(state['knowledge'])} 个知识源的回答..."
return {"draft_answer": answer}
# 质量检查
def check_quality(state: CustomerServiceState) -> CustomerServiceState:
# 模拟质量评分
score = 0.85
return {"quality_score": score}
# 人类审核（低质量时）
def human_review(state: CustomerServiceState) -> CustomerSystemState:
if state["quality_score"] < 0.8:
decision = interrupt({"answer": state["draft_answer"]})
return {"human_reviewed": True, "final_answer": decision["revised_answer"]}
return {"human_reviewed": False, "final_answer": state["draft_answer"]}
# 构建完整工作流
workflow = StateGraph(CustomerServiceState)
workflow.add_node("classify", classify_intent)
workflow.add_node("retrieve", retrieve_knowledge)
workflow.add_node("generate", generate_answer)
workflow.add_node("check", check_quality)
workflow.add_node("review", human_review)
workflow.add_edge(START, "classify")
workflow.add_edge("classify", "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "check")
workflow.add_edge("check", "review")
workflow.add_edge("review", END)
cs_app = workflow.compile(interrupt_before=["review"])
# 执行
result = cs_app.invoke({"user_query": "如何重置密码？"})
print(f"最终回答：{result['final_answer']}")
8.4 测试与调试
# 单元测试单个节点
def test_classify_intent():
state = {"user_query": "密码问题"}
result = classify_intent(state)
assert result["intent"] == "account_management"
# 集成测试完整流程
def test_full_workflow():
result = cs_app.invoke({"user_query": "测试问题"})
assert "final_answer" in result
assert len(result["final_answer"]) > 0
# 可视化调试
cs_app.get_graph().draw_mermaid_png("customer_service.png")
9. 总结与资源
9.1 模式速查表
| 模式 | 适用场景 | 复杂度 |
|---|---|---|
| 线性流水线 | 顺序处理 | ⭐ |
| 条件分支 | 路由决策 | ⭐⭐ |
| 并行执行 | 并发任务 | ⭐⭐ |
| 固定迭代 | 次数确定 | ⭐⭐ |
| 条件终止 | 质量导向 | ⭐⭐⭐ |
| 重试退避 | 容错处理 | ⭐⭐ |
| 主管 - 工人 | 任务分解 | ⭐⭐⭐ |
| 投票共识 | 提高准确性 | ⭐⭐⭐ |
| 人类审核 | 质量把关 | ⭐⭐⭐ |
| 嵌套状态机 | 分层设计 | ⭐⭐⭐⭐ |
9.2 学习路径建议
- 入门：掌握线性、条件、并行基础模式
- 进阶：学习循环和协作模式
- 高级：实践模式组合和嵌套
- 专家：自定义模式和框架扩展
9.3 推荐资源
9.4 下一步
- 在实际项目中应用这些模式
- 贡献新的模式到社区
- 关注 LangGraph 官方更新
感谢阅读！ 希望这篇文章能帮助你构建更强大、更可靠的 LangGraph 应用。🚀