LangGraph Workflows & Agents 设计模式

LangGraph Workflows & Agents 设计模式 快速入门

一句话卖点：LangChain 官方的”Agent 建筑设计图”——把业界验证过的 6 种 Agentic Workflow 模式拆成可直接复用的代码模板，从此不再从零设计 Agent 架构，直接搭积木。

这是什么？适合谁？

LangGraph Workflows & Agents 是 LangChain 团队维护的Agentic Workflow 设计模式官方文档库。它系统性地总结了 6 种经过工业验证的可复用模式——从最简单的 Prompt Chaining 到复杂的多 Agent 协作——每种模式都有概念解释、架构图、和可运行的 Python 代码示例。

核心区分：文档将 AI 应用分为两大类——

• Workflows（工作流）：代码路径预定义，LLM 按固定顺序执行。包括 Prompt Chaining、Parallelization、Routing、Orchestrator-Worker、Evaluator-Optimizer 五种模式；
• Agents（智能体）：LLM 动态决定自己的流程和工具使用，路径不固定。Agent 自己选择调用哪些工具、何时停止、何时请求人类介入。 六种模式的速览： | 模式 | 一句话 | 适用场景 | |------|--------|----------| | Prompt Chaining | A 的输出作为 B 的输入，串行执行 | 文档生成（大纲→正文→润色）、多步翻译 | | Parallelization | 多个 LLM 调用同时执行 | 多维度分析、多语种翻译、分块摘要 | | Routing | 分类输入后分发到专门的处理分支 | 客服意图路由、代码语言检测分发 | | Orchestrator-Worker | 中心调度器动态拆任务给专业 Worker | 多 Agent 研究、复杂软件开发 | | Evaluator-Optimizer | 生成→评估→反馈→重写循环 | 代码审查自动修、文案迭代优化 | | Agent | LLM 自主决策工具调用和流程 | 聊天助手、自主编程 Agent | 适合谁？AI 工程师/架构师——需要为团队设计可维护的 Agent 系统架构；LLM 应用开发者——从单次 Prompt 调用升级到多步 Workflow，需要成熟的设计参考；技术管理者——评估团队 Agent 架构是否合理，避免重复造轮子；Agent 框架开发者——理解业界标准模式，设计自己的 Agent DSL 或编排层。

不适合：只想用一句话调 LLM 的极简需求（直接用 ChatGPT API 即可）；完全无编程基础（需要 Python + LangChain 基础）；对 LangChain 生态有强烈排斥（但设计模式本身可脱离 LangChain 实现）。

准备工作

1. Python 3.11+：LangGraph 要求 Python ≥ 3.9，推荐 3.11 以上以获得更好的类型提示支持；
2. LLM API Key：至少一个——推荐 OpenAI API Key（GPT-4o / GPT-4.1）或 Anthropic API Key（Claude 3.5 Sonnet / Claude 4）；
3. 基本 Python 知识：理解 async/await、类型注解（Type Hints）、Pydantic 模型；
4. （推荐）LangChain 基础：了解 ChatModel、Tool、PromptTemplate 等概念会让学习更顺畅；
5. 一个明确的使用场景：带着一个真实需求来学效果最好——比如”我要做一个客服工单自动分类→处理→回复的流水线”。

4 步快速上手

第 1 步：安装 LangGraph 并理解核心概念

pip install langgraph langchain langchain-openai
# 如果用 Anthropic Claude
pip install langchain-anthropic

LangGraph 有三个核心概念，必须理解才能读懂所有模式： ① StateGraph（状态图）——整个 Workflow 的容器，所有 Node 共享一个 State 字典：

from typing import TypedDict
from langgraph.graph import StateGraph
class State(TypedDict):
query: str
category: str
result: str
graph = StateGraph(State)

② Node（节点）——图中执行具体逻辑的函数，接收 State 返回更新：

def classify(state: State) -> dict:
# 用 LLM 分类用户意图
result = llm.invoke(f"分类这句话: {state['query']}")
return {"category": result}

③ Edge（边）——连接节点的有向箭头。分两种：

• 普通边（add_edge）：固定从 A → B
• 条件边（add_conditional_edges）：根据 State 动态选择下一个 Node 第一个模式：Prompt Chaining（提示链） 这是最简单的模式——把任务拆成串行步骤，上一步的输出是下一步的输入：

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
class ChainState(TypedDict):
topic: str
outline: str
draft: str
final: str
llm = ChatOpenAI(model="gpt-4o")
def generate_outline(state: ChainState) -> dict:
prompt = f"为'{state['topic']}'生成一个3段式大纲"
return {"outline": llm.invoke(prompt).content}
def write_draft(state: ChainState) -> dict:
prompt = f"根据大纲写正文：\n{state['outline']}"
return {"draft": llm.invoke(prompt).content}
def polish(state: ChainState) -> dict:
prompt = f"润色以下文章，使其更简洁有力：\n{state['draft']}"
return {"final": llm.invoke(prompt).content}
# 构建图
builder = StateGraph(ChainState)
builder.add_node("outline", generate_outline)
builder.add_node("draft", write_draft)
builder.add_node("polish", polish)
builder.set_entry_point("outline")
builder.add_edge("outline", "draft")
builder.add_edge("draft", "polish")
builder.add_edge("polish", END)
graph = builder.compile()
# 运行
result = graph.invoke({"topic": "AI Agent 的未来趋势"})
print(result["final"])

执行流程：topic 输入 → generate_outline → write_draft → polish → 输出 final

验证方法：print(result["final"]) 能看到完整的三段式润色文章。如果输出不连贯，检查每个 Node 的 prompt 是否把上一步结果完整传入了。

第 2 步：掌握 Routing 与 Parallelization 组合模式

在实际项目中，Routing 和 Parallelization 经常组合使用：先路由分类，再对不同类型的请求并行执行不同的处理。

Routing（路由）——根据输入内容将请求分发到不同的处理分支：

from typing import Literal

from langgraph.graph import StateGraph, END

class RouteState(TypedDict):

query: str

intent: str

response: str

def classify_intent(state: RouteState) -> dict:

prompt = """将用户问题分类为以下之一：

- technical: 技术问题（bug、代码、部署）

- billing: 账单/付费问题

- general: 一般咨询

只输出分类标签。"""

intent = llm.invoke(

f"{prompt}\n\n用户问题：{state['query']}"

).content.strip().lower()

return {"intent": intent}

def route_by_intent(state: RouteState) -> Literal["tech_support", "billing_support", "general_support"]:

if state["intent"] == "technical":

return "tech_support"

elif state["intent"] == "billing":

return "billing_support"

else:

return "general_support"

def tech_support(state: RouteState) -> dict:

return {"response": llm.invoke(f"你是技术专家，回答：{state['query']}").content}

def billing_support(state: RouteState) -> dict:

return {"response": llm.invoke(f"你是账单专员，回答：{state['query']}").content}

def general_support(state: RouteState) -> dict:

return {"response": llm.invoke(f"回答用户问题：{state['query']}").content}

# 构建路由图

builder = StateGraph(RouteState)

builder.add_node("classify", classify_intent)

builder.add_node("tech_support", tech_support)

builder.add_node("billing_support", billing_support)

builder.add_node("general_support", general_support)

builder.set_entry_point("classify")

builder.add_conditional_edges("classify", route_by_intent, {

"tech_support": "tech_support",

"billing_support": "billing_support",

"general_support": "general_support",

})

builder.add_edge("tech_support", END)

builder.add_edge("billing_support", END)

builder.add_edge("general_support", END)

router_graph = builder.compile()

Parallelization（并行化）——同时执行多个独立的 LLM 调用：

from langgraph.graph import StateGraph, END

import operator

from typing import Annotated

class ParallelState(TypedDict):

text: str

# Send API 并行：每个元素独立进入节点

sections: list[str]

# 汇总结果用 operator.add 合并

summaries: Annotated[list[str], operator.add]

def split_sections(state: ParallelState) -> dict:

"""将文本拆分为多个 section，返回 Send 列表"""

# 模拟拆分

sections = state["text"].split("\n\n")

return {"sections": sections}

from langgraph.types import Send

def continue_to_summarize(state: ParallelState):

"""为每个 section 生成一个 Send 对象，并行执行"""

return [

Send("summarize", {"text": section})

for section in state["sections"]

]

def summarize(state: dict) -> dict:

"""每个 section 独立摘要（并行执行）"""

summary = llm.invoke(f"用一句话摘要：{state['text']}").content

return {"summaries": [summary]}

builder = StateGraph(ParallelState)

builder.add_node("split", split_sections)

builder.add_node("summarize", summarize)

builder.set_entry_point("split")

builder.add_conditional_edges("split", continue_to_summarize, ["summarize"])

builder.add_edge("summarize", END)

parallel_graph = builder.compile()

关键理解：Send API 是 LangGraph 实现并行化的核心——continue_to_summarize 返回的每个 Send 对象都会创建一个独立的 summarize 执行实例，它们并行运行不互相等待。

第 3 步：构建 Orchestrator-Worker 多 Agent 系统

Orchestrator-Worker 是 LangGraph 中最强大的模式：一个中央 Orchestrator（调度器）分析任务后动态分配给多个专业 Worker，Worker 各自拥有独立的工具和 System Prompt。

from typing import TypedDict, Annotated

import operator

from langgraph.graph import StateGraph, END

class OrchestratorState(TypedDict):

task: str

plan: list[dict]          # 调度计划: [{step, assigned_worker}]

worker_results: Annotated[list[dict], operator.add]

final_report: str

# === 定义三个专业 Worker ===

def researcher(state: dict) -> dict:

"""研究员 Worker：搜索和收集信息"""

prompt = f"""你是研究员。收集关于以下主题的关键信息：

{state['task']}

返回：3-5 个关键发现，每条带来源。"""

result = llm.invoke(prompt).content

return {"worker_results": [{"worker": "researcher", "result": result}]}

def analyst(state: dict) -> dict:

"""分析师 Worker：分析数据、找趋势"""

prompt = f"""你是数据分析师。基于研究结果，分析以下主题：

{state['task']}

找出 3 个关键趋势和 2 个风险点。"""

result = llm.invoke(prompt).content

return {"worker_results": [{"worker": "analyst", "result": result}]}

def writer(state: dict) -> dict:

"""撰稿 Worker：将分析结果写成报告"""

# 从 state 中取出前面 Worker 的结果

researcher_output = next(

(w["result"] for w in state.get("worker_results", [])

if w["worker"] == "researcher"), ""

)

analyst_output = next(

(w["result"] for w in state.get("worker_results", [])

if w["worker"] == "analyst"), ""

)

prompt = f"""你是报告撰写人。基于以下信息撰写一份结构化报告：

## 研究结果

{researcher_output}

## 分析洞察

{analyst_output}

## 原始任务

{state['task']}

请生成包含摘要、主要发现、趋势分析、风险提示四个部分的报告。"""

result = llm.invoke(prompt).content

return {"final_report": result}

# === 构建图 ===

builder = StateGraph(OrchestratorState)

# 所有 Worker 并行启动

builder.add_node("researcher", researcher)

builder.add_node("analyst", analyst)

builder.add_node("writer", writer)

builder.set_entry_point("researcher")

builder.set_entry_point("analyst")  # 多入口 = 并行

builder.add_edge("researcher", "writer")

builder.add_edge("analyst", "writer")

builder.add_edge("writer", END)

orchestrator_graph = builder.compile()

# 运行

result = orchestrator_graph.invoke({

"task": "2026 年 AI Agent 开源框架的市场竞争格局分析"

})

print(result["final_report"])

执行流程：researcher 和 analyst 并行执行 → writer 等待两者完成 → 汇总生成报告

扩展思路：在真实项目中，你可以在 writer 后面再接一个 reviewer Worker 做质量审核，用 add_conditional_edges 判断是否需要退回 writer 重写——这自然引入了下一个模式：Evaluator-Optimizer。

第 4 步：引入 Evaluator-Optimizer 迭代优化

Evaluator-Optimizer 模式的核心是一个”生成→评估→反馈→重写”循环：

from typing import Literal

class EvalState(TypedDict):

task: str

draft: str

score: int

feedback: str

iteration: int

def generate(state: EvalState) -> dict:

"""生成初稿或根据反馈修改"""

if state.get("feedback"):

prompt = f"""重写以下内容，解决反馈中提到的问题：

原始任务：{state['task']}

当前草稿：{state['draft']}

反馈意见：{state['feedback']}

请生成改进版本。"""

else:

prompt = f"完成任务：{state['task']}"

return {

"draft": llm.invoke(prompt).content,

"iteration": state.get("iteration", 0) + 1

}

def evaluate(state: EvalState) -> dict:

"""评估草稿质量"""

prompt = f"""评估以下输出质量（满分 10 分）。

任务：{state['task']}

输出：{state['draft']}

评分标准：

- 完整性（是否覆盖所有要求）

- 准确性（事实是否正确）

- 清晰度（是否易于理解）

返回格式：

Score: X/10

Feedback: 具体改进建议（如无问题写 "None"）"""

response = llm.invoke(prompt).content

# 解析评分

import re

score_match = re.search(r"Score:\s*(\d+)/10", response)

score = int(score_match.group(1)) if score_match else 5

feedback = response.split("Feedback:")[-1].strip() if "Feedback:" in response else ""

return {"score": score, "feedback": feedback}

def should_continue(state: EvalState) -> Literal["generate", "end"]:

"""决定是否继续迭代"""

if state["score"] >= 8 or state["iteration"] >= 3:

return "end"

return "generate"

# 构建 Evaluator-Optimizer 图

builder = StateGraph(EvalState)

builder.add_node("generate", generate)

builder.add_node("evaluate", evaluate)

builder.set_entry_point("generate")

builder.add_edge("generate", "evaluate")

builder.add_conditional_edges(

"evaluate", should_continue,

{"generate": "generate", "end": END}

)

eval_graph = builder.compile()

result = eval_graph.invoke({

"task": "写一篇 2026 年 AI Agent 技术趋势短文（300 字）"

})

print(f"最终评分: {result['score']}/10")

print(f"迭代次数: {result['iteration']}")

print(f"最终输出:\n{result['draft']}")

循环逻辑：generate → evaluate → 不达标 → generate（用反馈重写）→ evaluate → 达标 → END

常见踩坑

1. State 字段未定义导致 KeyError：所有 Node 中写入的 key 必须在 TypedDict 的 State 类中声明，LangGraph 运行时会严格校验——忘记定义字段是最常见的初学者错误；

2. 并行节点中存在隐式数据依赖：两个节点被设为并行（多入口），但 Worker B 的 prompt 里硬编码了”基于 Worker A 的结果”——这会导致 B 读到空数据。确保并行节点之间完全独立；

3. 条件边返回值与目标节点名不匹配：route_by_intent 返回 "tech_support" 但图中注册的节点叫 "tech_support_handler"——LangGraph 会抛出 InvalidUpdateError；

4. Evaluator-Optimizer 无退出条件导致无限循环：should_continue 里忘记设最大迭代次数上限——LLM 有时永远给不出 10/10 的评分，导致死循环消耗大量 token；

5. Send API 返回值错误：continue_to_* 函数必须返回 list[Send]，而不是 dict 或 list[str]——返回类型错误是并行化失败的首要原因；

6. Orchestrator 拆分粒度太细：把任务拆成 20 个微小 Worker，每个 Worker 只做一行代码的事——调度开销 > 实际计算，建议 2-5 个 Worker；

7. 遗忘 operator.add 导致并行结果被覆盖：并行节点同时写同一个 State 字段时，必须用 Annotated[list, operator.add] 而非普通 list——否则后完成的节点会覆盖先完成的结果。

初级用法

• 内容生成流水线：Prompt Chaining 模式，大纲→正文→润色→配图描述，一键生成完整文章；

• 多语种翻译服务：Parallelization 模式，一篇文档同时翻译为英/日/韩/法四种语言，并行执行互不阻塞；

• 智能客服路由：Routing 模式，用户问题分类为”退换货”/“技术故障”/“投诉”后分别进入对应处理流程；

• 代码审查助手：Evaluator-Optimizer 模式，AI 生成代码 → Review → 发现 bug → 自动修复 → 再 Review，直到通过。

高级玩法

• 模式组合嵌套：外层 Routing + 内层 Evaluator-Optimizer——先路由到”代码生成”分支，内部跑生成→审查→修复循环，最后输出通过审查的代码；

• Human-in-the-Loop 打断：在 Evaluator-Optimizer 的评估节点后用 interrupt() 挂起，等人类审核打分后再决定继续迭代还是结束——LangGraph 原生支持；

• 持久化与断点续跑：LangGraph 支持 SQLite/Postgres checkpoint，长时间运行的 Orchestrator-Worker 任务可以在断点恢复——适合需要数小时的研究任务；

• 跨 Worker 记忆共享：在 Orchestrator-Worker 模式中加入共享记忆层（InMemoryStore 或 PostgresSaver），让 Worker 之间可以交换中间发现——类似团队白板；

• 动态 Worker 注册：不从代码写死 Worker 列表，而是在 Runtime 从配置文件/YAML 中动态加载 Worker 的 System Prompt 和工具集——实现”热插拔”Agent 系统；

• 与 LangSmith 集成调试：所有 LangGraph 图的执行都可以自动追踪到 LangSmith，可视化每个 Node 的输入输出、耗时、token 消耗——这对于调试复杂 Workflow 至关重要。

小技巧

1. 从最简单模式开始验证：不要一上来就搭 Orchestrator-Worker。先用 Prompt Chaining 验证你的 LLM 在这个任务上的基础表现，确认模型选对了再加复杂度；

2. 用 graph.get_graph().draw_mermaid_png() 可视化：LangGraph 内置 Mermaid 图导出，画出来比看代码直观 10 倍——尤其是条件边多了以后；

3. Prompt 模板化管理：把每个 Node 的 System Prompt 抽到单独的 .txt 或 YAML 文件中管理，而不是硬编码在 Python 代码里——方便非技术同事调试提示词；

4. State 设计先于代码：拿到需求后先画 State 字段图——这个 Workflow 从头到尾需要哪些数据？哪些是全局共享？哪些是 Node 局部？画清楚 State 再写代码能避免 50% 的返工；

5. 小模型做路由，大模型做生成：Routing 分类任务可以用便宜的 GPT-4o-mini / Claude Haiku，把 GPT-4o / Claude Sonnet 留给真正需要推理的 Worker——成本可降 60-80%；

6. 用 TypedDict 的 total=False 标记可选字段：State 里不是所有字段一开始就有值——draft 在第一步没有，final 在中间步骤也没有。标记为可选可以避免类型检查告警；

7. 条件边返回值加类型注解：def route(state) -> Literal["a", "b", "c"]——加了 Literal 类型后，IDE 能自动补全和校验，重构时如果改了节点名但忘了改路由返回值，类型检查器会直接报错。

常见问题 FAQ

Q1: LangGraph 和 LangChain 是什么关系？必须一起用吗？

A: LangGraph 是 LangChain 生态中的编排框架（类似”控制流引擎”），LangChain 提供 LLM 调用、工具封装、Prompt 模板等基础能力。你可以只用 LangGraph + 原生 OpenAI SDK，不依赖 LangChain——LangGraph 的 StateGraph 本身对 LLM 提供者无感。

Q2: 这些模式和 Anthropic 的《Building Effective Agents》文章有什么关系？

A: LangChain 的这篇文档直接引用了 Anthropic 的理论框架作为设计哲学。Anthropic 的文章提出了”增强型 LLM”概念和 Workflow vs Agent 的二分法；LangChain 将其落地为 LangGraph 中可运行的代码。两篇配合阅读效果最好。

Q3: 6 种模式都要学吗？还是只学需要的？

A: 建议按依赖关系学：Prompt Chaining（基础）→ Routing + Parallelization（核心组合）→ Orchestrator-Worker（进阶）→ Evaluator-Optimizer（质量保障）。Agent 模式理解概念即可，具体实现建议用 LangChain 的 create_agent API 或直接看 LangGraph Agent 教程。

Q4: LangGraph 有可视化编辑器吗？

A: 有 LangGraph Studio（桌面应用 + 云版），可以可视化编辑 StateGraph、拖拽节点和边、实时调试。但目前 Studio 更偏向调试和监控，生产级 Workflow 还是建议代码定义。另外可以用 draw_mermaid_png() 导出静态图做文档。

Q5: 除了 LangGraph，还有其他方式实现这些模式吗？

A: 这些是设计模式，不绑定任何框架。你可以用 CrewAI 实现 Orchestrator-Worker、用 AWS Step Functions 实现 Prompt Chaining、用手写 asyncio 实现 Parallelization。LangGraph 的价值在于用图抽象统一了所有模式——一套语法覆盖全部 6 种。如果你只需要一两种模式，完全可以用更轻量的方案。

Q6: 生产环境部署有哪些注意事项？

A: (1) 使用 PostgresSaver 而非内存 MemorySaver 做 checkpoint——进程重启后状态不丢失；(2) 设置 recursion_limit 防止无限循环——默认 25 步，Evaluator-Optimizer 场景可能需要调高到 50；(3) 配置 LangSmith 追踪所有执行，出问题时能回溯每个 Node 的输入输出；(4) 设置 Node 级别的超时和重试策略（LangGraph 内置支持 RetryPolicy）；(5) 用 astream_events() 替代 invoke() 做流式输出，提升用户体验。

进阶学习建议

第 1-2 小时：跑通所有模式

• 按本文的 4 步逐一复制代码到 Jupyter Notebook 或 .py 文件

• 用自己的 API Key 跑通 Prompt Chaining → Routing+Parallel → Orchestrator-Worker → Evaluator-Optimizer

• 每跑通一个模式，尝试把 prompt 替换成你自己的业务场景

第 3-4 小时：组合模式解决真实问题

• 选一个你手头的真实需求（如”自动分析 GitHub Issue 并生成修复 PR”）

• 设计需要哪些模式组合：Routing（判断 Issue 类型）→ Orchestrator-Worker（分派给分析+修复 Worker）→ Evaluator-Optimizer（代码 Review 循环）

• 实现一个可运行的 MVP——哪怕 Workers 只做 mock 输出

第 1 周：深入 LangGraph 特性

• 学习 checkpoint 持久化、Human-in-the-Loop interrupt、streaming 三种生产级特性

• 用 LangSmith 可视化你的 Workflow 执行轨迹

• 将 State 从 TypedDict 升级到 Pydantic BaseModel（更强大的数据校验）

第 2-4 周：生产化

• 加入错误处理和重试策略（RetryPolicy）

• 用 PostgresSaver 替代内存 checkpoint

• 用 FastAPI 包装成 REST API 供前端调用

• 配置日志、监控、成本追踪

推荐资源：

• LangGraph 官方文档：langchain.com/docs

• LangGraph 教程合集：langchain-ai.lang.chat

• Anthropic《Building Effective Agents》：anthropic.com/engineering

• LangGraph GitHub 仓库 + 示例：github.com/langchain-ai/langgraph

• LangGraph Cookbook（社区示例集）：github.com/langchain-ai/langgraph-cookbook

本文基于 LangChain/LangGraph 官方文档和公开资料整理，AI 辅助生成，MagicNetWorld 尚未完成独立实测。代码示例来自官方文档的简化改编版本，完整可运行示例请参见 LangGraph 官方文档。如有错误或过时信息，请通过 contact@magicnetworld.com 反馈。