LangGraph多Agent协作：3个Agent比1个强，但踩了4个坑

此前讨论LangGraph工作流时，认为为Agent增加流程控制已足够。直到一个真实需求出现：

老板要求："开发一个行业调研助手，输入行业关键词，自动生成调研报告——包含行业概述、竞品分析、投资建议三部分。"

单Agent难以胜任——一个AI同时搜索行业信息、分析竞品、撰写投资建议，每部分都浅尝辄止，报告如流水账。

单LangGraph工作流也不够——虽然流程可控，但每个节点由同一个AI执行，缺乏专业分工。如同一个人兼任产品经理、设计师、开发，每项工作都半吊子。

多Agent协作的用途正在于此——让多个Agent各司其职，分工合作。 一个负责搜索调研，一个负责竞品分析，一个负责报告撰写。如同公司中不同岗位各司其职，最终拼成完整交付物。

经过2周从单Agent改造为多Agent协作，踩了4个坑。每个都是"多人协作"场景下的特有难题。

---

总体结论

维度	单Agent	LangGraph工作流	多Agent协作
分工	一个AI处理所有任务	流程分步，但同一AI	多个AI各管一摊
专业性	每项任务深度不足	每步可控，但角色固定	每个Agent配备专属Prompt
输出质量	中等	中上	高，专人专事
复杂度	低	中	高，需设计协作机制
协调成本	无	无	有，Agent间可能产生冲突

采用Java开发者的理解：单Agent ≈ 全栈开发者包揽所有；LangGraph工作流 ≈ 一个后端按流程依次处理不同任务；多Agent ≈ 前端+后端+DBA分工协作，存在接口契约与数据传递。

简单任务用单Agent，流程化任务用LangGraph，专业化任务用多Agent。 这一总结值得牢记。

---

整体架构：3个Agent如何协作

用户输入行业关键词
        ↓
   [调度Agent] ── 接收需求，分配任务
      ↓                ↓                ↓
[调研Agent]      [竞品Agent]      [写作Agent]
搜索行业信息        分析竞品          整合报告
      ↓                ↓                ↓
   [调度Agent] ── 汇总结果，生成最终报告
        ↓
    输出调研报告

3个Agent各自拥有专属Prompt与工具：

Agent	职责	Prompt重点	工具
调研Agent	搜索行业信息	"你是行业调研专家……"	搜索工具
竞品Agent	分析竞争格局	"你是竞品分析专家……"	搜索+分析工具
写作Agent	整合内容并撰写报告	"你是商业写作专家……"	无，仅负责撰写不搜索

调度Agent扮演项目经理角色，其他Agent为专业执行者。 调度Agent不直接执行任务，仅负责分配与汇总。

---

坑1：3个Agent同时输出，结果混乱

故障表现

初期思路简单：将3个Agent放入LangGraph并行执行，最后合并输出。

# 错误思路：3个Agent并行，直接拼接结果
graph.add_node("researcher", research_agent)
graph.add_node("competitor", competitor_agent)
graph.add_node("writer", writer_agent)

# 3个Agent同时启动
graph.add_edge(START, "researcher")
graph.add_edge(START, "competitor")
graph.add_edge(START, "writer")
# 结果如何合并？

输出结果：

调研Agent：新能源汽车行业2025年市场规模达XXX万……
竞品Agent：蔚来、小鹏、理想三足鼎立……
写作Agent：报告：# 新能源汽车调研（这段尚未获取数据就开始撰写）

3个Agent同时输出，写作Agent在未收到调研数据的情况下便开始撰写报告。 并行执行不意味着协作——各干各的，互不等待。

正确做法：先调研+分析，再撰写报告——采用并行+串行混合模式

[调度Agent]
    ↓           ↓
[调研Agent]  [竞品Agent]    ← 这两个可并行执行
    ↓           ↓
   [汇总节点]                  ← 待两者完成后
        ↓
   [写作Agent]                ← 有数据后再撰写

代码实现：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

class ReportState(TypedDict):
    keyword: str              # 行业关键词
    research_result: str      # 调研结果
    competitor_result: str    # 竞品分析结果
    final_report: str         # 最终报告

# ============ Agent节点 ============

def research_agent(state: ReportState) -> ReportState:
    """调研Agent：搜索行业信息"""
    keyword = state["keyword"]
    # 实际项目：调用搜索工具/搜索API
    result = f"""【行业调研】{keyword}行业概况：
- 市场规模：2025年预计突破2万亿元
- 增长率：近3年复合增长率18.5%
- 政策：国家持续出台扶持政策
- 趋势：智能化、绿色化、全球化三大方向"""
    return {"research_result": result}

def competitor_agent(state: ReportState) -> ReportState:
    """竞品Agent：分析竞争格局"""
    keyword = state["keyword"]
    result = f"""【竞品分析】{keyword}行业竞争格局：
- 头部玩家：A公司（市占率25%）、B公司（18%）、C公司（15%）
- 差异化策略：A重技术、B重渠道、C重性价比
- 新进入者：D公司获得亿元融资，主打下沉市场
- 竞争壁垒：技术专利和供应链是核心护城河"""
    return {"competitor_result": result}

def merge_node(state: ReportState) -> ReportState:
    """汇总节点：等待调研和竞品都完成，合并数据"""
    # 此节点不进行实际处理，仅确保两个Agent都已完成
    return {}

def writer_agent(state: ReportState) -> ReportState:
    """写作Agent：整合调研和竞品数据，生成报告"""
    keyword = state["keyword"]
    research = state.get("research_result", "无调研数据")
    competitor = state.get("competitor_result", "无竞品数据")

    report = f"""# {keyword}行业调研报告

## 一、行业概况
{research}

## 二、竞争格局
{competitor}

## 三、投资建议
- 短期关注：头部公司财报和技术发布节奏
- 中期布局：关注供应链关键环节的中小企业
- 长期看好：{keyword}行业整体向上，但需警惕政策变化风险

---
*本报告由AI调研助手生成，仅供参考*"""
    return {"final_report": report}

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

graph = StateGraph(ReportState)

# 添加节点
graph.add_node("researcher", research_agent)
graph.add_node("competitor", competitor_agent)
graph.add_node("merge", merge_node)
graph.add_node("writer", writer_agent)

# 并行：调度同时发送至调研和竞品
graph.add_edge(START, "researcher")
graph.add_edge(START, "competitor")

# 汇合：两者完成后进入merge
graph.add_edge("researcher", "merge")
graph.add_edge("competitor", "merge")

# 串行：merge完成后撰写报告
graph.add_edge("merge", "writer")
graph.add_edge("writer", END)

app = graph.compile()

# ============ 运行 ============

result = app.invoke({"keyword": "新能源汽车"})
print(result["final_report"])

关键点：

要点	说明
并行执行	START → researcher 和 START → competitor 同时启动
汇合等待	LangGraph自动等待：merge有两个入边，两者均完成后才执行
数据在State中流转	researcher写入research_result，competitor写入competitor_result，writer均可读取

以Java开发者的理解：这相当于 CompletableFuture.allOf(future1, future2).thenApply(merge) ——并行执行，全部完成后合并。

---

坑2：Agent间数据不一致，相互矛盾

故障表现

报告生成后，内容存在问题：

调研Agent显示：市场规模2万亿元
竞品Agent显示：市场规模1.5万亿元

调研Agent表示：头部为A、B、C公司
竞品Agent表示：头部为A、D、E公司

同一行业，两个Agent给出的数据不一致。 原因在于各自独立搜索，信息来源与数据口径不同。

正确做法：共享上下文，使Agent基于统一基础信息工作

为每个Agent注入共享的行业背景信息：

def research_agent(state: ReportState) -> ReportState:
    keyword = state["keyword"]
    context = state.get("shared_context", "")

    prompt = f"""你是行业调研专家。
行业关键词：{keyword}
{f"已知背景信息：{context}" if context else ""}

请基于以上信息，搜索并整理行业概况。
若背景信息中已有数据，以背景信息为准，请勿自行编造不同数据。"""

    # 实际项目：此处调用LLM
    result = f"【调研结果】{keyword}行业概况……"
    return {"research_result": result}

def competitor_agent(state: ReportState) -> ReportState:
    keyword = state["keyword"]
    # 关键：将调研结果作为上下文传递给竞品Agent
    research = state.get("research_result", "")

    prompt = f"""你是竞品分析专家。
行业关键词：{keyword}
行业调研结果：{research}

请基于以上调研数据，分析竞争格局。
注意：市场规模、头部玩家等数据必须与调研结果保持一致。"""

    result = f"【竞品分析】基于调研数据……"
    return {"competitor_result": result}

但此方法导致新问题：调研与竞品变为串行——竞品需等待调研结果才能开始。 并行优势消失。

更优做法：先由调度Agent生成"共享背景"，再并行执行

def coordinator_agent(state: ReportState) -> ReportState:
    """调度Agent：先快速生成行业背景摘要"""
    keyword = state["keyword"]
    context = f"""{keyword}行业基础信息（2025年）：
- 大方向：政策利好，市场增长
- 主要玩家：A、B、C等
- 核心数据：市场规模约2万亿

以上为共享背景，后续Agent请以此为基准。"""
    return {"shared_context": context}

def research_agent_v2(state: ReportState) -> ReportState:
    keyword = state["keyword"]
    context = state.get("shared_context", "") # 读取共享背景
    result = f"基于共享背景的调研结果……"
    return {"research_result": result}

def competitor_agent_v2(state: ReportState) -> ReportState:
    keyword = state["keyword"]
    context = state.get("shared_context", "") # 读取共享背景
    result = f"基于共享背景的竞品分析……"
    return {"competitor_result": result}

# 图结构调整
graph = StateGraph(ReportState)
graph.add_node("coordinator", coordinator_agent) # 先调度
graph.add_node("researcher", research_agent_v2)
graph.add_node("competitor", competitor_agent_v2)
graph.add_node("merge", merge_node)
graph.add_node("writer", writer_agent)

graph.add_edge(START, "coordinator")     # 先调度
graph.add_edge("coordinator", "researcher") # 再并行
graph.add_edge("coordinator", "competitor")
graph.add_edge("researcher", "merge")
graph.add_edge("competitor", "merge")
graph.add_edge("merge", "writer")
graph.add_edge("writer", END)

流程变为：

[调度Agent：生成共享背景]
        ↓              ↓
[调研Agent]       [竞品Agent]     ← 基于共享背景并行执行
        ↓              ↓
     [汇总] → [写作Agent]

Agent协作的核心规则：共享上下文优于各自为政。 如同团队开发——先对齐接口文档，再各自开发，否则联调时必然出现问题。

协作模式	数据一致性	并行效率	适用场景
各自搜索，不共享	数据矛盾	全并行	信息来源唯一的简单场景
串行传递	一致	无并行	数据强依赖场景
先共享背景，再并行	基本一致	大部分并行	推荐，适用于大多数场景

---

坑3：写作Agent直接复制粘贴调研结果

故障表现

报告生成后，写作Agent的输出如下：

# 新能源汽车调研报告

## 行业概况
【调研结果】新能源汽车行业概况：市场规模2万亿……

## 竞争格局
【竞品分析】基于调研数据……

## 投资建议
（无）

写作Agent将调研和竞品结果直接复制粘贴，未进行分析、整合和投资建议。 原因在于其Prompt过于简单："请整合调研和竞品数据，生成报告"——它理解为"复制粘贴"。

正确做法：为写作Agent提供结构化输出模板

WRITER_PROMPT = """你是商业报告写作专家。

你的任务是基于调研和竞品数据，撰写专业的行业调研报告。

## 输入数据
- 调研结果：{research_result}
- 竞品分析：{competitor_result}

## 输出要求
1. 不要直接复制粘贴输入数据，需用自己的语言重新组织
2. 每个部分必须包含"分析"和"结论"，不能仅罗列数据
3. 投资建议部分必须包含：短期关注、中期布局、长期判断三个维度
4. 末尾必须添加风险提示

## 输出格式
# {keyword}行业调研报告

## 一、行业概况
（用1-2段总结行业现状，引用关键数据时注明来源）

## 二、竞争格局
（分析头部玩家策略差异，指出竞争壁垒）

## 三、投资建议
### 短期关注（6个月内）
### 中期布局（1-2年）
### 长期判断（3-5年）

## 四、风险提示
- 政策风险：……
- 技术风险：……
- 市场风险：……

---
*本报告由AI调研助手生成，仅供参考*
"""

def writer_agent_v2(state: ReportState) -> ReportState:
    keyword = state["keyword"]
    research = state.get("research_result", "")
    competitor = state.get("competitor_result", "")

    prompt = WRITER_PROMPT.format(
        research_result=research,
        competitor_result=competitor,
    )

    # 实际项目：调用LLM生成
    report = f"# {keyword}行业调研报告nn（基于模板生成的完整报告）……"
    return {"final_report": report}

在多Agent场景中，Prompt工程比单Agent更为重要。 因为每个Agent的输出是下一个Agent的输入——一个Agent偷懒，后续所有Agent都会受影响。

多Agent Prompt设计原则：

原则	说明	反例
明确角色定位	"你是XX专家"	"你是一个AI助手"
指定输入格式	"基于以下调研数据"	"用你自己的知识"
指定输出格式	提供模板，规定章节结构	"写一份报告"
禁止行为	"不要复制粘贴原始数据"	仅说"整合"未明确禁止复制
输出校验点	"必须包含投资建议部分"	无校验，AI可能省略

以Java开发者的理解：这类似于Service之间的接口契约——定义了@RequestMapping的入参和出参类型，下游Service才能正确调用。多Agent的Prompt就是Agent间的接口文档。

---

坑4：单个Agent出错导致整份报告失效

故障表现

某次运行中，调研Agent搜索不到数据（搜索工具超时），返回空字符串。结果如下：

# 新能源汽车调研报告

## 一、行业概况
（空）

## 二、竞争格局
（正常竞品分析）

## 三、投资建议
基于不完整信息，无法给出投资建议。

一个Agent失败，整份报告质量急剧下降。 这与微服务架构相似——一个服务宕机，整个链路受影响。

正确做法：每个Agent增加降级逻辑 + 最终校验

def research_agent_v3(state: ReportState) -> ReportState:
    """调研Agent：包含降级逻辑"""
    keyword = state["keyword"]

    try:
        # 尝试调用搜索工具
        result = search_tool(keyword) # 实际项目中的搜索调用
    except Exception as e:
        # 降级：使用共享背景信息生成基础调研
        context = state.get("shared_context", "")
        if context:
            result = f"基于行业背景的概要分析：{context}"
        else:
            result = f"抱歉，{keyword}行业的详细调研数据暂时无法获取，以下为基于行业常识的概要……"

    return {"research_result": result}

def validator_node(state: ReportState) -> ReportState:
    """校验节点：检查各Agent输出是否完整"""
    issues = []

    if not state.get("research_result") or "无法获取" in state.get("research_result", ""):
        issues.append("调研数据不完整")

    if not state.get("competitor_result"):
        issues.append("竞品分析缺失")

    if issues:
        return {"validation_issues": "；".join(issues)}
    return {"validation_issues": ""}

def writer_agent_v3(state: ReportState) -> ReportState:
    """写作Agent：根据校验结果调整报告"""
    issues = state.get("validation_issues", "")

    report_body = "……（报告正文）……"

    # 若存在数据缺失，在报告中明确标注
    if issues:
        report_body += f"""

---
 **数据完整性提示**：本次报告存在以下数据缺失：{issues}
相关部分的分析可能不够充分，建议补充调研后重新生成。"""

    return {"final_report": report_body}

完整的多Agent工作流（包含校验与降级）：

graph = StateGraph(ReportState)
graph.add_node("coordinator", coordinator_agent)
graph.add_node("researcher", research_agent_v3)  # 带降级
graph.add_node("competitor", competitor_agent_v3) # 带降级
graph.add_node("merge", merge_node)
graph.add_node("validator", validator_node) # 新增校验
graph.add_node("writer", writer_agent_v3) # 带校验感知

graph.add_edge(START, "coordinator")
graph.add_edge("coordinator", "researcher")
graph.add_edge("coordinator", "competitor")
graph.add_edge("researcher", "merge")
graph.add_edge("competitor", "merge")
graph.add_edge("merge", "validator") # 先校验
graph.add_edge("validator", "writer") # 再写报告
graph.add_edge("writer", END)

多Agent容错的3个层次：

层次	策略	Java类比
Agent内降级	搜索失败→使用缓存数据/常识补充	try-catch + fallback逻辑
流程校验	校验节点检查各Agent输出完整性	Spring的@Valid + 切面校验
输出标注	数据缺失时在报告中明确标注	日志warn + 返回值带错误码

宁可输出一份"带有标注的半成品报告"，也不输出一份"看似完整但数据有误的报告"。

---

完整代码：3个Agent协作的行业调研助手

"""
LangGraph多Agent协作：行业调研助手
依赖：pip install langgraph langchain langchain-openai
"""
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END

# ============ State定义 ============

class ReportState(TypedDict):
    keyword: str
    shared_context: str          # 调度Agent生成的共享背景
    research_result: str         # 调研Agent输出
    competitor_result: str       # 竞品Agent输出
    validation_issues: str       # 校验结果
    final_report: str           # 最终报告

# ============ Agent节点 ============

def coordinator_agent(state: ReportState) -> ReportState:
    """调度Agent：生成共享背景，分配任务"""
    keyword = state["keyword"]
    # 实际项目：此处调用LLM快速生成行业背景
    context = f"""{keyword}行业基础信息（2025年）：
- 市场趋势：持续增长，政策利好
- 主要玩家：头部3-5家公司占据主要份额
- 技术方向：智能化+绿色化
- 数据口径：本报告统一使用2025年最新数据"""
    return {"shared_context": context}

def research_agent(state: ReportState) -> ReportState:
    """调研Agent：基于共享背景搜索行业信息"""
    keyword = state["keyword"]
    context = state.get("shared_context", "")

    # 实际项目：调用搜索工具 + LLM
    # 模拟输出
    result = f"""{keyword}行业调研结果：

一、市场规模
2025年市场规模预计突破2万亿元，近3年复合增长率18.5%。

二、增长驱动
政策扶持是核心驱动力，2024年出台12项行业支持政策。
技术进步推动成本下降，锂电池成本同比下降15%。

三、发展趋势
智能化：L3及以上自动驾驶渗透率达25%
绿色化：碳排放新标准倒逼技术升级
全球化：出口占比从15%提升至22%

（基于共享背景信息整理）"""
    return {"research_result": result}

def competitor_agent(state: ReportState) -> ReportState:
    """竞品Agent：基于共享背景分析竞争格局"""
    keyword = state["keyword"]
    context = state.get("shared_context", "")

    result = f"""{keyword}行业竞品分析：

一、头部玩家
- A公司：市占率25%，技术领先，研发投入占比15%
- B公司：市占率18%，渠道优势，线下门店2000+
- C公司：市占率15%，性价比路线，下沉市场王者

二、竞争壁垒
技术专利：头部3家公司合计持有行业60%核心专利
供应链：锂电池供应链集中度高，新进入者难突破
品牌：消费者品牌忠诚度调查，头部品牌复购率达65%

三、新进入者
D公司获10亿元融资，主打中端市场，预计2026年量产。

（基于共享背景信息分析）"""
    return {"competitor_result": result}

def merge_node(state: ReportState) -> ReportState:
    """汇总节点：确保并行Agent均已完成"""
    return {}

def validator_node(state: ReportState) -> ReportState:
    """校验节点：检查数据完整性"""
    issues = []
    if not state.get("research_result"):
        issues.append("调研数据缺失")
    if not state.get("competitor_result"):
        issues.append("竞品分析缺失")
    return {"validation_issues": "；".join(issues)}

def writer_agent(state: ReportState) -> ReportState:
    """写作Agent：整合数据生成报告"""
    keyword = state["keyword"]
    research = state.get("research_result", "暂无调研数据")
    competitor = state.get("competitor_result", "暂无竞品数据")
    issues = state.get("validation_issues", "")

    report = f"""# {keyword}行业调研报告

## 一、行业概况

{research}

## 二、竞争格局

{competitor}

## 三、投资建议

### 短期关注（6个月内）
- 关注头部公司季度财报，尤其是研发投入变化
- 跟踪政策发布节奏，补贴退坡影响需提前预判

### 中期布局（1-2年）
- 供应链关键环节（电池、芯片）的中小企业有并购机会
- 智能化赛道的细分龙头值得关注

### 长期判断（3-5年）
- 行业整体向上，但增速将放缓至10-12%
- 技术壁垒是长期投资的核心筛选标准

## 四、风险提示
- 政策风险：补贴退坡可能影响短期销量
- 技术风险：技术路线分化，押注错误将错失窗口
- 市场风险：竞争加剧导致价格战，利润率承压

---
*本报告由AI调研助手生成，仅供参考*"""

    if issues:
        report += f"""

---
 **数据完整性提示**：本次报告存在以下问题：{issues}"""

    return {"final_report": report}

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

graph = StateGraph(ReportState)
graph.add_node("coordinator", coordinator_agent)
graph.add_node("researcher", research_agent)
graph.add_node("competitor", competitor_agent)
graph.add_node("merge", merge_node)
graph.add_node("validator", validator_node)
graph.add_node("writer", writer_agent)

# 流程：调度 → 并行调研+竞品 → 汇总 → 校验 → 写报告
graph.add_edge(START, "coordinator")
graph.add_edge("coordinator", "researcher")
graph.add_edge("coordinator", "competitor")
graph.add_edge("researcher", "merge")
graph.add_edge("competitor", "merge")
graph.add_edge("merge", "validator")
graph.add_edge("validator", "writer")
graph.add_edge("writer", END)

app = graph.compile()

# ============ 运行 ============

if __name__ == "__main__":
    result = app.invoke({"keyword": "新能源汽车"})
    print(result["final_report"])

运行效果：

# 新能源汽车行业调研报告

## 一、行业概况
新能源汽车行业调研结果：
一、市场规模
2025年市场规模预计突破2万亿元……

## 二、竞争格局
新能源汽车行业竞品分析：
一、头部玩家
- A公司：市占率25%……

## 三、投资建议
### 短期关注（6个月内）
……

## 四、风险提示
……

3个Agent分工协作，30秒即可生成结构完整的调研报告。 若使用单Agent，同样质量至少需要3分钟，且常遗漏某个维度。

---

4个坑的总结

#	坑	错误做法	正确做法	一句话
1	Agent并行输出混乱	所有Agent同时启动	先共享背景，再并行，最后汇总	并行≠协作，需设置汇合点
2	Agent数据不一致	各自搜索，数据矛盾	调度Agent先生成共享背景，各Agent以此为基准	先对齐再工作
3	写作Agent仅复制粘贴	Prompt过于简单	提供结构化输出模板+禁止行为+校验点	Prompt是Agent间的接口契约
4	单个Agent出错导致全报告失效	无容错机制	降级逻辑+校验节点+缺失标注	宁出半成品，不出错误品

---

从单Agent到多Agent的演进路径

阶段	使用工具	何时升级
单Agent	第6篇的LangChain Agent	任务简单，3-5个工具
LangGraph工作流	第9篇的LangGraph	任务有固定流程，步骤有先后
多Agent协作	本文的LangGraph多Agent	任务需要专业分工，单Agent难以胜任

不要一开始就使用多Agent。 先用单Agent验证需求，再用LangGraph控制流程，最后发现一个AI处理不过来时，才拆分为多Agent。过早拆分等同于过度设计，增加维护难度。

这与Java架构演进的道理相同： 单体 → 模块化 → 微服务，不要一开始就采用微服务架构。