LangGraph Multi-Agent：团队协作的艺术

发表于2026-02-19|更新于2026-04-29|技术系列LangGraph

|浏览量:

想象一下：你让AI写一篇投资研报。它搜索了资料、分析了数据、写出了报告，但最后你发现——数据引用错了，逻辑链条断了，结论站不住脚。

这时候你可能会想：如果有个专门的研究员负责搜集资料，有个专门的分析师负责数据验证，还有个专门的写手负责行文逻辑，最后还有个审核员把关质量，该多好？

这就是多智能体（Multi-Agent）的核心思想：让专业的人做专业的事。

一、什么时候该拆分多个Agent？

在LangGraph的世界里，单个Agent就像是一个全能选手。它能思考、能调用工具、能产出结果。但当任务变得复杂时，这个”全能选手”开始力不从心。

单Agent的瓶颈

想象你在用ChatGPT完成一个复杂任务：

1
2
3

你：帮我分析苹果公司2024年财报，预测2025年股价走势

AI：（开始搜索...下载财报...分析数据...写报告...）

听起来不错？但问题很快就来了：

工具太多：财报分析需要搜索、下载、PDF解析、数据计算、可视化等十几种工具，Agent经常选错
上下文爆炸：搜索返回的内容、PDF的原始文本、中间计算结果混在一起，Agent迷失在信息海洋里
角色混乱：一会儿是研究员在搜集资料，一会儿是分析师在算市盈率，角色跳来跳去，逻辑连贯性变差
错误难追溯：最后发现某个数据引用错了，但不知道是在哪个环节出的问题

多Agent的优势

现在我们换一种方式：

1	用户请求 → 研究员（搜集资料） → 写手（撰写初稿） → 审核员（检查质量） → 最终报告

每个Agent只做一件事，但做到极致：

角色	专业领域	工具集	优势
研究员	信息搜集	搜索引擎、PDF解析、数据库查询	快速定位权威数据源
写手	内容创作	文档编辑、格式排版	逻辑清晰、行文流畅
审核员	质量控制	对比检查、事实校验	发现错误、提出修改意见

拆分的黄金法则

不是每个任务都需要多Agent。什么时候该拆分？问自己三个问题：

1. 任务是否涉及多个专业领域？

需要搜资料 + 做计算 + 写文档 → 考虑拆分
只是简单问答 → 单Agent足够

2. 是否需要不同角色对结果进行验证？

研报需要事实核查 → 专门加个审核员Agent
闲聊对话 → 不需要

3. 工具数量是否超过10个？

工具太多会让Agent”选择困难” → 按职责分组
3-5个工具 → 单个Agent能handle

二、三种协作模式

LangGraph提供了三种多Agent协作模式，就像三种不同的团队协作方式。

模式一：监督者模式（Supervisor）—— 项目经理制

这是最常见的模式。一个Supervisor Agent像项目经理一样，负责任务分配和流程控制，Worker Agent们各司其职，完成自己负责的部分。

工作方式：

          ┌──────────────┐
          │  Supervisor  │
          │   (监督者)    │
          └──────┬───────┘
                 │
     ┌──────────┼──────────┐
     │          │          │
     ▼          ▼          ▼
┌────────┐ ┌────────┐ ┌────────┐
│Worker 1│ │Worker 2│ │Worker 3│
│研究员  │ │ 写手   │ │审核员  │
└────────┘ └────────┘ └────────┘

适用场景：

任务流程相对固定（先做A，再做B，最后做C）
需要明确的责任划分
每个Worker的输出是下一个Worker的输入

优点：

流程清晰，易于理解和调试
Supervisor可以统一监控进度
每个Worker只需关注自己的职责

缺点：

Supervisor是单点，可能成为瓶颈
流程僵化，不够灵活

模式二：协作模式（Collaboration）—— 头脑风暴制

所有Agent共享同一个”对话白板”，任何人都可以随时发言、补充信息、提出质疑。

工作方式：

┌─────────────────────────────────────────┐
│         共享消息白板                    │
│  ┌─────────────────────────────────┐   │
│  │ 研究员：找到了苹果2024年营收数据 │   │
│  │ 分析师：我算了一下，同比增长8%    │   │
│  │ 写手：这个数据放在第一段可以吗？  │   │
│  │ 审核员：来源可靠吗？             │   │
│  │ 研究员：来自官方财报第15页...     │   │
│  └─────────────────────────────────┘   │
│         ↑        ↑        ↑            │
│    ┌────┴────────┴────────┴────┐      │
│    │  研究员   分析师   写手   审核员  │
│    └────────────────────────────┘      │
└─────────────────────────────────────────┘

适用场景：

需要集思广益的创意任务
信息需要反复验证和补充
团队成员需要频繁交流

优点：

信息高度透明，所有人都能看见全貌
灵活性强，可以根据讨论动态调整方向
容易发现盲点

缺点：

消息白板会越来越长，上下文膨胀
容易出现”七嘴八舌”的混乱
难以确定什么时候该结束

模式三：层级团队（Hierarchical）—— 集团公司制

一个Agent本身可以是一个完整的多Agent系统。就像集团公司：总部管理着多个子公司，每个子公司下面又有自己的团队。

工作方式：

┌─────────────────────────────────────┐
│           CEO Agent                 │
│        （战略决策层）                │
└───────────┬─────────────┬───────────┘
            │             │
    ┌───────▼──────┐ ┌────▼────────┐
    │  财务部      │ │  研发部      │
    │ （子系统）    │ │ （子系统）   │
    └───────┬──────┘ └─────┬───────┘
            │              │
    ┌───────▼──────┐ ┌────▼────────┐
    │会计→出纳→审计│ │前端→后端→测试│
    └──────────────┘ └─────────────┘

适用场景：

超大规模项目
需要分层管理的复杂系统
不同团队可以并行工作

优点：

可扩展性极强
模块化程度高，子团队可复用
并行效率高

缺点：

复杂度极高
调试困难
信息在层层传递中可能失真

三、消息传递 vs 共享状态

多Agent之间如何交流？有两种主流方式。

方式一：消息传递（Message Passing）

就像发微信：Agent A给Agent B发一条消息，消息包含明确的发送者、接收者和内容。

class Message:
    sender: str      # 谁发的
    receiver: str    # 发给谁
    content: str     # 内容是什么
    msg_type: str    # 消息类型：任务/回复/询问

类比：微信私聊、邮件、Slack消息

特点：

点对点通信，明确知道谁在跟谁说话
消息有明确的边界，不会混在一起
可以追踪每条消息的传递路径

方式二：共享状态（Shared State）

就像白板会议：所有人看着同一块白板，任何人都可以在上面写东西，所有人都能看见。

class SharedState:
    messages: list      # 所有对话历史
    findings: dict      # 研究发现
    current_task: str   # 当前任务
    final_output: str   # 最终结果

类比：腾讯文档、Notion、共享白板

特点：

信息对所有Agent透明
不需要指定”发给谁”
状态会持续累积，需要定期清理

该怎么选？

场景	推荐方式	原因
监督者模式	消息传递	Supervisor需要明确分配任务给特定Worker
协作模式	共享状态	所有人都要看到完整的讨论过程
层级团队	混合使用	层内用共享状态，层间用消息传递

四、实战：研报生成小组

好了，理论知识讲完，我们来实战。我们要构建一个三Agent系统：

研究员（Researcher）：搜集资料，找出关键数据
写手（Writer）：根据资料撰写研报
审核员（Reviewer）：检查事实准确性、逻辑完整性

完整代码实现

import asyncio
from typing import TypedDict, Annotated, Sequence, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent
from operator import add

# ============================================
# 1. 定义共享状态
# ============================================

class ReportState(TypedDict):
    """研报生成小组的共享状态"""
    messages: Annotated[Sequence[BaseMessage], add]  # 对话历史
    topic: str                                       # 研报主题
    research_data: str                               # 研究发现
    draft_report: str                                # 研报初稿
    final_report: str                                # 最终报告
    review_feedback: str                             # 审核意见
    next_step: Literal["research", "write", "review", "revise", "end"]  # 下一步

# ============================================
# 2. 创建三个Agent
# ============================================

# 初始化模型
llm = ChatOpenAI(model="gpt-4", temperature=0)

# ---- 研究员 Agent ----
research_tools = [
    # 这里可以放搜索工具、数据库查询工具等
    # 示例中我们先用简单的模拟
]

research_agent = create_react_agent(
    model=llm,
    tools=research_tools,
    prompt="""你是一位专业的财经研究员。你的任务是搜集和整理资料。

工作流程：
1. 理解研报主题
2. 列出需要搜集的关键信息（财务数据、市场表现、行业动态等）
3. 整理成结构化的研究发现

输出格式：

【研究发现】

关键数据：(数据项) = (数值)，来源：(来源)
关键数据：(数据项) = (数值)，来源：(来源)
…

【数据摘要】
用2-3句话概括核心发现


注意：
- 每个数据都要有来源说明
- 如果数据不确定，明确标注"待验证"
"""
)

# ---- 写手 Agent ----
write_tools = []  # 写手主要靠模型能力

writer_agent = create_react_agent(
    model=llm,
    tools=write_tools,
    prompt="""你是一位资深的财经撰稿人。你的任务是根据研究员提供的资料撰写研报。

研报结构：
1. 标题：简洁有力，突出核心观点
2. 摘要：3-5句话概括全文
3. 核心发现：分点列出3-5个关键发现
4. 详细分析：
   - 财务分析
   - 市场分析
   - 风险因素
5. 结论与建议：明确的投资建议

写作要求：
- 逻辑清晰，论证充分
- 数据引用要准确
- 语言专业但不晦涩
- 适当使用小标题增强可读性
"""
)

# ---- 审核员 Agent ----
review_tools = []

reviewer_agent = create_react_agent(
    model=llm,
    tools=review_tools,
    prompt="""你是一位严格的研报审核员。你的任务是检查研报质量。

审核维度：
1. 数据准确性：数据是否有矛盾？来源是否可靠？
2. 逻辑完整性：论证是否充分？结论是否有依据？
3. 表达清晰度：语言是否流畅？结构是否合理？

输出格式：

【审核结果】

数据准确性：(通过/需修改) - 具体说明
逻辑完整性：(通过/需修改) - 具体说明
表达清晰度：(通过/需修改) - 具体说明

【修改建议】

【最终判定】
通过 / 需要修改


如果判定"需要修改"，必须给出具体修改建议。
"""
)

# ============================================
# 3. 定义节点逻辑
# ============================================

def research_node(state: ReportState):
    """研究员节点：搜集资料"""
    topic = state["topic"]
    
    print("\n🔍 研究员开始工作...")
    
    # 调用研究员Agent
    result = research_agent.invoke({
        "messages": [HumanMessage(content=f"请搜集关于'{topic}'的资料，整理成研究发现。")]
    })
    
    research_data = result["messages"][-1].content
    
    print("✅ 研究员完成资料搜集")
    
    return {
        "messages": result["messages"],
        "research_data": research_data,
        "next_step": "write"
    }

def write_node(state: ReportState):
    """写手节点：撰写研报"""
    topic = state["topic"]
    research_data = state["research_data"]
    
    print("\n✍️ 写手开始撰写研报...")
    
    # 如果是修改，要带上审核意见
    review_feedback = state.get("review_feedback", "")
    
    if review_feedback:
        prompt = f"""请根据以下信息修改研报：

研报主题：{topic}

研究发现：
{research_data}

之前的审核意见：
{review_feedback}

请修改研报，解决审核提出的问题。
"""
    else:
        prompt = f"""请根据以下研究发现撰写研报：

研报主题：{topic}

研究发现：
{research_data}

请撰写一份专业的投资研报。
"""
    
    result = writer_agent.invoke({
        "messages": [HumanMessage(content=prompt)]
    })
    
    draft_report = result["messages"][-1].content
    
    print("✅ 写手完成研报撰写")
    
    return {
        "messages": result["messages"],
        "draft_report": draft_report,
        "next_step": "review"
    }

def review_node(state: ReportState):
    """审核员节点：检查质量"""
    draft_report = state["draft_report"]
    
    print("\n👁️ 审核员开始检查...")
    
    result = reviewer_agent.invoke({
        "messages": [HumanMessage(content=f"请审核以下研报：\n\n{draft_report}")]
    })
    
    review_feedback = result["messages"][-1].content
    
    # 判断审核结果
    if "通过" in review_feedback and "需要修改" not in review_feedback:
        print("✅ 审核通过！")
        return {
            "messages": result["messages"],
            "review_feedback": review_feedback,
            "final_report": draft_report,
            "next_step": "end"
        }
    else:
        print("⚠️ 审核不通过，需要修改")
        return {
            "messages": result["messages"],
            "review_feedback": review_feedback,
            "next_step": "write"  # 返回写手修改
        }

def router(state: ReportState):
    """路由函数：决定下一步执行哪个节点"""
    return state["next_step"]

# ============================================
# 4. 构建工作流图
# ============================================

# 创建图
workflow = StateGraph(ReportState)

# 添加节点
workflow.add_node("research", research_node)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)

# 设置入口
workflow.set_entry_point("research")

# 添加边
workflow.add_edge("research", "write")

# 审核节点的条件路由
workflow.add_conditional_edges(
    "review",
    router,
    {
        "write": "write",      # 需要修改，返回写手
        "end": END             # 审核通过，结束
    }
)

# 写手节点路由到审核
workflow.add_edge("write", "review")

# 编译图
app = workflow.compile()

# ============================================
# 5. 运行示例
# ============================================

async def generate_report(topic: str):
    """生成研报的主函数"""
    
    print(f"\n{'='*50}")
    print(f"开始生成研报：{topic}")
    print(f"{'='*50}")
    
    # 初始状态
    initial_state = {
        "messages": [],
        "topic": topic,
        "research_data": "",
        "draft_report": "",
        "final_report": "",
        "review_feedback": "",
        "next_step": "research"
    }
    
    # 运行工作流
    result = await app.ainvoke(initial_state)
    
    print(f"\n{'='*50}")
    print("研报生成完成！")
    print(f"{'='*50}\n")
    
    return result["final_report"]

# 运行示例
if __name__ == "__main__":
    topic = "苹果公司2024年财报分析及2025年投资展望"
    report = asyncio.run(generate_report(topic))
    
    print("\n【最终研报】\n")
    print(report)

代码解析

1. 状态定义（State）

class ReportState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add]
    topic: str
    research_data: str
    draft_report: str
    final_report: str
    review_feedback: str
    next_step: str

这是三个Agent的”共享白板”。Annotated[..., add]表示这个字段的内容可以累加，而不是覆盖。

2. 三个Agent定义

每个Agent都是一个create_react_agent，有专门的系统提示词定义其角色。

3. 节点函数

每个节点函数接收当前状态，执行任务，返回更新后的状态。注意返回的是变更的部分，不是完整状态。

4. 路由逻辑

1 2	def router(state: ReportState): return state["next_step"]

审核节点根据审核结果设置next_step，路由函数读取这个值决定下一步走向。

5. 条件边

workflow.add_conditional_edges(
    "review",
    router,
    {
        "write": "write",  # 需要修改
        "end": END         # 审核通过
    }
)

这是监督者模式的典型用法——根据条件决定流程走向。

运行效果

运行这段代码，你会看到类似这样的输出：

==================================================
开始生成研报：苹果公司2024年财报分析及2025年投资展望
==================================================

🔍 研究员开始工作...
✅ 研究员完成资料搜集

✍️ 写手开始撰写研报...
✅ 写手完成研报撰写

👁️ 审核员开始检查...
⚠️ 审核不通过，需要修改

✍️ 写手开始撰写研报...
✅ 写手完成研报撰写

👁️ 审核员开始检查...
✅ 审核通过！

==================================================
研报生成完成！
==================================================

【最终研报】
...