📌 前言

想象一下：让一个人同时当产品经理、架构师和程序员，他能做好吗？

“多个专长不同的 Agent 协作，比一个全能 Agent 更可靠。” —— LangGraph 设计哲学

本文带你从零掌握 LangGraph 多 Agent 协作开发，包含两个完整可运行项目、生产级最佳实践和踩坑总结。

你将学到：

多 Agent 协作的 3 种核心模式
LangGraph 核心概念 30 秒速通
三角色协作系统（产品经理 + 架构师 + 程序员）
研究员 vs 批评家辩论系统（95 行最小示例）
状态持久化、人工介入、错误处理
生产环境优化技巧

一、为什么需要多 Agent？

1.1 单 Agent 的三大局限

表格
局限 表现 后果
角色冲突 一个人既写需求又写代码，容易遗漏 质量下降
上下文爆炸 所有信息塞进一个 prompt，超出窗口限制 幻觉增多
无法并行 串行执行所有子任务 响应慢、成本高

1.2 多 Agent 协作的 3 种模式

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
┌─────────────────────────────────────────────────────┐
│ 模式一：流水线 (Pipeline) │
│ │
│ [Agent A] ──→ [Agent B] ──→ [Agent C] ──→ 输出 │
│ (产品经理) (架构师) (程序员) │
│ │
│ 特点：固定流程，每一步输出是下一步输入 │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│ 模式二：Supervisor（监督者） │
│ │
│ ┌──────────┐ │
│ │Supervisor│ │
│ └────┬─────┘ │
│ ┌───────┼───────┐ │
│ ▼ ▼ ▼ │
│ [Agent A] [Agent B] [Agent C] │
│ │
│ 特点：动态调度，Supervisor 决定下一步交给谁 │
└─────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│ 模式三：辩论 (Debate) │
│ │
│ [研究员] ◄──── 反驳 ────► [批评家] │
│ │ │ │
│ └──────────┬─────────────┘ │
│ ▼ │
│ [总结员] → 输出 │
│ │
│ 特点：多轮对抗，通过辩论提升输出质量 │
└─────────────────────────────────────────────────────┘

三种模式对比：

表格
模式 适用场景 复杂度 可控性 典型应用
流水线 固定流程，步骤明确 ⭐ 低 🟢 高 文档生成、数据处理管道
Supervisor 动态调度，任务不确定 ⭐⭐ 中 🟡 中 客服系统、复杂任务分解
辩论 质量优化，需要多轮迭代 ⭐⭐⭐ 高 🔴 低 代码审查、方案评审、研究分析

二、LangGraph 核心概念 30 秒速通

LangGraph 是 LangChain 团队推出的 Agent 编排框架，核心理念是用图（Graph）来定义 Agent 的工作流。

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
核心四要素：
┌────────────────────────────────────────────────────┐
│ │
│ State（状态）── 所有节点共享的数据容器 │
│ ↕ │
│ Node（节点）── 具体的处理逻辑（Agent/函数） │
│ ↕ │
│ Edge（边）── 节点之间的连接关系 │
│ ↕ │
│ Compile（编译）── 将图编译为可执行的 Runnable │
│ │
└────────────────────────────────────────────────────┘

一句话理解： State 是数据包，Node 是加工厂，Edge 是传送带，Compile 是启动生产线。

三、环境搭建

3.1 安装依赖

bash
1
2
3
4
5
pip install langgraph>=0.2.0 langchain-openai>=0.2.0 langchain-core>=0.3.0

状态持久化需要

pip install langgraph-checkpoint-sqlite>=0.1.0

3.2 配置环境变量

python
1
2
3
4
5
6
7
8
9
10
import os

OpenAI API 配置

os.environ[“OPENAI_API_KEY”] = “sk-your-api-key”
os.environ[“OPENAI_BASE_URL”] = “https://api.openai.com/v1” # 或其他兼容端点

验证安装

import langgraph
print(f"LangGraph 版本: {langgraph.version}")

💡 国内用户提示：可以使用智谱 GLM、通义千问等国产模型，只需修改 base_url 和 api_key。

四、实战一：三角色协作系统

我们要实现一个「产品经理 → 架构师 → 程序员」的流水线协作系统。用户输入需求，三个角色依次处理，最终输出完整的技术方案。

4.1 完整代码

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
“”"
LangGraph 多 Agent 协作：三角色流水线系统
产品经理 → 架构师 → 程序员
“”"
import operator
from typing import Annotated, TypedDict, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langgraph.graph import StateGraph, START, END

==================== Step 1: 定义 State ====================

class ProjectState(TypedDict):
“”“项目状态 - 所有节点共享”“”
requirement: str # 原始需求
prd: str # 产品需求文档
architecture: str # 架构设计方案
code: str # 代码实现
messages: Annotated[list, operator.add] # 消息列表（自动追加）

==================== Step 2: 定义 Agent 节点 ====================

llm = ChatOpenAI(model=“gpt-4o-mini”, temperature=0.7)

def product_manager(state: ProjectState) -> dict:
“”“产品经理：将需求转化为 PRD”“”
response = llm.invoke([
SystemMessage(content=“”"你是一位资深产品经理。
根据用户需求，输出简明的 PRD，包含：

1. 用户故事（As a… I want… So that…）
2. 核心功能点（不超过5个）
3. 验收标准
   保持简洁，每个部分3-5行。“”“),
   HumanMessage(content=f"用户需求：{state[‘requirement’]}”)
   ])
   return {
   “prd”: response.content,
   “messages”: [AIMessage(content=response.content, name=“PM”)]
   }

def architect(state: ProjectState) -> dict:
“”“架构师：根据 PRD 设计技术方案”“”
response = llm.invoke([
SystemMessage(content=“”"你是一位资深架构师。
根据产品需求文档(PRD)，输出架构设计，包含：

1. 技术栈选择及理由
2. 核心模块划分
3. 数据模型设计
4. API 接口定义
   保持实用，避免过度设计。“”“),
   HumanMessage(content=f"PRD：\n{state[‘prd’]}”)
   ])
   return {
   “architecture”: response.content,
   “messages”: [AIMessage(content=response.content, name=“Architect”)]
   }

def developer(state: ProjectState) -> dict:
“”“程序员：根据架构方案编写代码”“”
response = llm.invoke([
SystemMessage(content=“”"你是一位资深全栈工程师。
根据架构设计方案，输出核心代码实现，包含：

1. 项目结构（目录树）
2. 核心模块代码（Python）
3. 关键函数说明
   代码要可直接运行，添加必要注释。“”“),
   HumanMessage(content=f"架构方案：\n{state[‘architecture’]}”)
   ])
   return {
   “code”: response.content,
   “messages”: [AIMessage(content=response.content, name=“Developer”)]
   }

==================== Step 3: 条件路由（可选） ====================

def should_revise(state: ProjectState) -> Literal[“revise”, “end”]:
“”“判断是否需要返工（示例：简单演示条件路由）”“”
# 实际项目中可以让另一个 Agent 评审质量
messages = state.get(“messages”, [])
if len(messages) >= 3:
return “end”
return “revise”

==================== Step 4: 构建图 ====================

def build_pipeline():
“”“构建流水线图”“”
graph = StateGraph(ProjectState)

# 添加节点
graph.add_node("product_manager", product_manager)
graph.add_node("architect", architect)
graph.add_node("developer", developer)


# 添加边（流水线）
graph.add_edge(START, "product_manager")
graph.add_edge("product_manager", "architect")
graph.add_edge("architect", "developer")
graph.add_edge("developer", END)


# 编译
return graph.compile()

==================== Step 5: 执行 ====================

if name == “main”:
app = build_pipeline()

# 输入需求
result = app.invoke({
    "requirement": "开发一个支持多轮对话的客服机器人，能识别用户意图并自动回复",
    "messages": []
})


# 输出结果
print("=" * 60)
print("📋 产品需求文档 (PRD)")
print("=" * 60)
print(result["prd"])
print("\n" + "=" * 60)
print("🏗️ 架构设计方案")
print("=" * 60)
print(result["architecture"])
print("\n" + "=" * 60)
print("💻 代码实现")
print("=" * 60)
print(result["code"])

4.2 代码解析

执行流程：

plaintext
1
2
3
4
5
6
7
8
9
10
11
用户输入需求
│
▼
┌─────────────┐ PRD ┌─────────────┐ 架构方案 ┌─────────────┐
│ 产品经理 │ ──────────→ │ 架构师 │ ──────────→ │ 程序员 │
│ (Node 1) │ │ (Node 2) │ │ (Node 3) │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
最终输出

关键点说明：

表格
代码要素 作用 注意事项
TypedDict 定义 State 结构 字段类型要明确
Annotated[list, operator.add] 消息自动追加 不用手动 concat
add_edge(START, node) 设置入口 必须有一个起点
add_edge(node, END) 设置出口 流水线直接连 END
graph.compile() 编译为可执行对象 编译时会检查图合法性

五、实战二：研究员 vs 批评家辩论系统

辩论模式是多 Agent 协作中最有趣的模式。两个 Agent 互相反驳，最终由总结员给出结论。

5.1 完整代码（95 行最小示例）

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
“”"
LangGraph 多 Agent 辩论系统
研究员 ↔ 批评家 → 总结员
“”"
from typing import TypedDict, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, START, END

==================== State ====================

class DebateState(TypedDict):
topic: str # 辩论主题
research: str # 研究员观点
criticism: str # 批评家反驳
round: int # 当前轮次
max_rounds: int # 最大轮次
final_answer: str # 最终结论

==================== LLM ====================

llm = ChatOpenAI(model=“gpt-4o-mini”, temperature=0.7)

==================== Nodes ====================

def researcher(state: DebateState) -> dict:
“”“研究员：提出论点”“”
if state[“round”] == 0:
prompt = f"请对以下主题进行深入研究并给出专业分析：{state[‘topic’]}"
else:
prompt = f"""主题：{state[‘topic’]}
你之前的观点被批评家反驳了：
{state[‘criticism’]}

请针对反驳进行回应，补充论据，强化你的观点。“”"

response = llm.invoke([
    SystemMessage(content="你是一位严谨的研究员，擅长数据驱动的分析。每次回应要包含具体论据。"),
    HumanMessage(content=prompt)
])
return {"research": response.content, "round": state["round"] + 1}

def critic(state: DebateState) -> dict:
“”“批评家：反驳研究员”“”
response = llm.invoke([
SystemMessage(content=“你是一位犀利的批评家，擅长发现论证漏洞。指出逻辑缺陷、数据不足、忽略的反面论据。”),
HumanMessage(content=f"研究员的观点：\n{state[‘research’]}\n\n请提出你的批评。")
])
return {“criticism”: response.content}

def summarizer(state: DebateState) -> dict:
“”“总结员：综合双方观点给出结论”“”
response = llm.invoke([
SystemMessage(content=“你是中立的总结员。综合正反双方观点，给出平衡、全面的最终结论。”),
HumanMessage(content=f"""主题：{state[‘topic’]}

研究员观点：{state[‘research’]}

批评家意见：{state[‘criticism’]}

请给出最终综合结论。“”")
])
return {“final_answer”: response.content}

==================== 条件路由 ====================

def should_continue(state: DebateState) -> Literal[“critic”, “summarize”]:
“”“判断是否继续辩论”“”
if state[“round”] >= state[“max_rounds”]:
return “summarize”
return “critic”

==================== 构建图 ====================

def build_debate_graph():
graph = StateGraph(DebateState)

graph.add_node("researcher", researcher)
graph.add_node("critic", critic)
graph.add_node("summarizer", summarizer)


# 路由
graph.add_edge(START, "researcher")
graph.add_conditional_edges("researcher", should_continue, {
    "critic": "critic",
    "summarize": "summarizer"
})
graph.add_edge("critic", "researcher")  # 批评后回到研究员
graph.add_edge("summarizer", END)


return graph.compile()

==================== 执行 ====================

if name == “main”:
app = build_debate_graph()

result = app.invoke({
    "topic": "GPT-4 是否会在 3 年内被开源模型全面超越？",
    "research": "",
    "criticism": "",
    "round": 0,
    "max_rounds": 3,
    "final_answer": ""
})


print("🔬 研究员最终观点：")
print(result["research"])
print("\n🔍 批评家最终意见：")
print(result["criticism"])
print("\n📝 最终结论：")
print(result["final_answer"])

5.2 执行流程图

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
┌──────────┐
│ START │
└────┬─────┘
│
▼
┌──────────────┐
│ 研究员 │ ◄─────────────┐
│ (提出论点) │ │
└──────┬───────┘ │
│ │
round < max? │
╱ ╲ │
是 否 │
│ │ │
▼ ▼ │
┌──────────┐ ┌──────────┐ │
│ 批评家 │ │ 总结员 │ │
│ (反驳) │ │ (结论) │ │
└──────────┘ └────┬─────┘ │
│ │ │
└─────────────┘ │
(回到研究员) ───────────┘ ← 条件边循环

六、进阶功能

6.1 状态持久化（Checkpointer）

多轮对话中，我们需要保存 Agent 的中间状态。LangGraph 通过 Checkpointer 实现：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

创建 SQLite 连接

conn = sqlite3.connect(“checkpoints.db”, check_same_thread=False)
memory = SqliteSaver(conn)

编译时传入 checkpointer

app = graph.compile(checkpointer=memory)

使用 thread_id 区分不同会话

config = {“configurable”: {“thread_id”: “conversation-001”}}

第一次调用

result = app.invoke(
{“requirement”: “开发一个待办事项应用”, “messages”: []},
config=config
)

第二次调用（自动恢复上次状态）

result = app.invoke(
{“requirement”: “在上面基础上增加协作功能”, “messages”: []},
config=config
)

持久化方案对比：

表格
方案 适用场景 性能 部署复杂度
MemorySaver 开发测试 快（内存） ⭐
SqliteSaver 单机部署 中 ⭐⭐
PostgresSaver 生产环境 高 ⭐⭐⭐

6.2 人工介入（Human-in-the-Loop）

在关键节点让人类审批后再继续：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langgraph.checkpoint.memory import MemorySaver

在需要人工介入的节点添加 interrupt

app = graph.compile(
checkpointer=MemorySaver(),
interrupt_before=[“developer”] # 程序员节点前暂停
)

config = {“configurable”: {“thread_id”: “human-review-001”}}

执行到 developer 前会暂停

result = app.invoke(
{“requirement”: “开发用户认证模块”, “messages”: []},
config=config
)

此时可以查看中间状态

print(“架构方案（待审批）：”, result[“architecture”])

人类审批通过后，继续执行

可以直接修改 state

app.update_state(config, {“architecture”: “修改后的架构方案…”})

继续执行

result = app.invoke(None, config=config)

6.3 错误处理与重试

生产环境中，LLM 调用可能失败，需要重试机制：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import time
from tenacity import retry, stop_after_attempt, wait_exponential

class LLMAgent:
“”“带重试的 LLM Agent 封装”“”

def __init__(self, model="gpt-4o-mini", max_retries=3):
    self.llm = ChatOpenAI(
        model=model,
        temperature=0.7,
        max_retries=max_retries
    )


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True
)
def invoke(self, messages: list) -> str:
    """带重试的 LLM 调用"""
    try:
        response = self.llm.invoke(messages)
        return response.content
    except Exception as e:
        print(f"[LLM 调用失败] {e}，正在重试...")
        raise


def as_node(self, system_prompt: str, name: str):
    """将 Agent 转换为 LangGraph 节点函数"""
    def node(state: dict) -> dict:
        user_input = state.get("requirement", state.get("topic", ""))
        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_input)
        ]
        result = self.invoke(messages)
        return {"messages": result, f"{name}_output": result}
    return node

使用示例

pm_agent = LLMAgent(model=“gpt-4o-mini”)
pm_node = pm_agent.as_node(
system_prompt=“你是产品经理…”,
name=“pm”
)

七、生产实践

7.1 Send API 动态并行

当需要同时调用多个 Agent 并行工作时，使用 Send API：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langgraph.types import Send

def router(state: ProjectState) -> list:
“”“动态路由：根据需求拆分任务，并行分配给多个专家”“”
# 假设架构师分析后拆出了3个子任务
subtasks = [
{“task”: “设计数据库模型”, “domain”: “database”},
{“task”: “设计 API 接口”, “domain”: “api”},
{“task”: “设计前端页面”, “domain”: “frontend”}
]
# 每个子任务发送给对应的专家 Agent
return [
Send(f"expert_{t[‘domain’]}", {“task”: t[“task”], **state})
for t in subtasks
]

构建图时添加动态边

graph.add_conditional_edges(“architect”, router)

执行示意：

plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
┌──────────┐
│ 架构师 │ 拆分任务
└────┬─────┘
│ Send API
┌──────────┼──────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ DB 专家 │ │ API 专家 │ │ 前端专家 │ ← 并行执行
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
└──────────┼──────────┘
▼
┌──────────┐
│ 汇总 │ 合并结果
└──────────┘

7.2 子图嵌套（Subgraph）

复杂系统可以将每个 Agent 组封装为子图：

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

子图1：代码审查流程

review_graph = StateGraph(ReviewState)
review_graph.add_node(“lint_checker”, lint_checker)
review_graph.add_node(“security_reviewer”, security_reviewer)
review_graph.add_node(“review_summarizer”, review_summarizer)
review_graph.add_edge(START, “lint_checker”)
review_graph.add_edge(“lint_checker”, “security_reviewer”)
review_graph.add_edge(“security_reviewer”, “review_summarizer”)
review_graph.add_edge(“review_summarizer”, END)
compiled_review = review_graph.compile()

子图2：测试生成流程

test_graph = StateGraph(TestState)
test_graph.add_node(“test_writer”, test_writer)
test_graph.add_node(“test_runner”, test_runner)
test_graph.add_edge(START, “test_writer”)
test_graph.add_edge(“test_writer”, “test_runner”)
test_graph.add_edge(“test_runner”, END)
compiled_test = test_graph.compile()

主图：将子图作为节点

main_graph = StateGraph(MainState)
main_graph.add_node(“developer”, developer)
main_graph.add_node(“code_review”, compiled_review) # 子图作为节点
main_graph.add_node(“test_gen”, compiled_test) # 子图作为节点
main_graph.add_edge(START, “developer”)
main_graph.add_edge(“developer”, “code_review”)
main_graph.add_edge(“code_review”, “test_gen”)
main_graph.add_edge(“test_gen”, END)
main_app = main_graph.compile()

7.3 性能优化清单

表格
优化项 方法 效果
并行调用 使用 Send API 并行执行无依赖节点 耗时减少 50%+
流式输出 app.astream() 替代 app.invoke() 首字延迟降低
模型分级 简单任务用小模型，复杂任务用大模型 成本降低 60%+
缓存 对相同输入缓存 LLM 响应 减少重复调用
超时控制 为每个节点设置超时时间 防止死循环
Token 限制 截断过长的中间结果 避免上下文爆炸

python
1
2
3
4
5
6
7
8
9
10
11

流式输出示例

async for event in app.astream_events(
{“requirement”: “需求描述”, “messages”: []},
version=“v2”
):
kind = event[“event”]
if kind == “on_chat_model_stream”:
content = event[“data”][“chunk”].content
if content:
print(content, end=“”, flush=True)

八、避坑指南 🚧

坑 1：State 类型定义错误

python
1
2
3
4
5
6
7
8
9
10
11
12
13

❌ 错误：使用 dataclass

from dataclasses import dataclass

@dataclass
class MyState:
messages: list # LangGraph 不认识这个

✅ 正确：使用 TypedDict

from typing import TypedDict

class MyState(TypedDict):
messages: Annotated[list, operator.add] # 自动追加

坑 2：无限循环

python
1
2
3
4
5
6
7
8
9
10
11
12
13
14

❌ 条件边没有退出条件，导致无限循环

def route(state):
if state[“quality”] < 0.8:
return “improve” # 永远返回 improve
return “done”

✅ 设置最大重试次数

def route(state):
if state[“round”] >= state[“max_rounds”]:
return “done”
if state[“quality”] < 0.8:
return “improve”
return “done”

坑 3：消息格式混乱

python
1
2
3
4
5
6
7

❌ 混用字符串和 Message 对象

state[“messages”] = “some string” # 下一个节点期望的是 list

✅ 统一使用 Message 对象

from langchain_core.messages import AIMessage, HumanMessage
state[“messages”] = [AIMessage(content=“…”)]

坑 4：忘记编译

python
1
2
3
4
5
6
7

❌ 直接使用 graph 调用

graph.invoke({…}) # 报错！

✅ 先编译再调用

app = graph.compile()
app.invoke({…})

坑 5：Checkpointer 线程安全

python
1
2
3
4
5
6
7

❌ SqliteSaver 不是线程安全的

多线程共用一个 connection 会报错

✅ 使用连接池或 PostgresSaver

from langgraph.checkpoint.postgres import PostgresSaver

或者为每个线程创建独立连接

坑 6：Token 超限

python
1
2
3
4
5
6
7
8
9
10
11

❌ 不限制中间结果长度

state[“research”] = very_long_text # 可能超过 LLM 上下文

✅ 截断中间结果

def truncate(text: str, max_chars: int = 3000) -> str:
if len(text) <= max_chars:
return text
return text[:max_chars] + “\n…(已截断)”

state[“research”] = truncate(very_long_text)

九、总结与下一步

本文核心要点回顾

表格
知识点 核心内容
三种协作模式 流水线（固定）、Supervisor（动态）、辩论（对抗）
LangGraph 四要素 State → Node → Edge → Compile
状态管理 TypedDict 定义 + Annotated 自动聚合
持久化 SqliteSaver（开发）→ PostgresSaver（生产）
人工介入 interrupt_before 暂停 + update_state 修改
并行执行 Send API 动态分发
复杂系统 子图嵌套，分层管理

本专栏系列文章

表格
篇目 主题 状态
第1篇 大模型本地部署与 API 服务搭建 ✅ 已发布
第2篇 RAG 检索增强生成系统 ✅ 已发布
第3篇 Function Calling 工具调用 ✅ 已发布
第4篇 ReAct Agent 自主推理 ✅ 已发布
第5篇 Agent 记忆系统设计 ✅ 已发布
第6篇 LangGraph 多 Agent 协作（本文） ✅ 当前
第7篇 Agent 评估与可观测性 🔜 即将发布
第8篇 生产环境 Agent 部署与监控 🔜 即将发布

📢 作者说 ：多 Agent 系统是 AI 应用从"玩具"到"产品"的关键一步。掌握 LangGraph，你就掌握了 Agent 编排的核心能力。如果觉得有帮助， 点赞 + 收藏 + 关注 三连支持一下！

💬 有问题欢迎评论区讨论，看到必回！

本文代码已在 Python 3.10+ / LangGraph 0.2.x 环境下验证通过。完整代码仓库地址见评论区置顶。