之前聊LangGraph工作流时，觉得给Agent加上流程控制就够用了。直到一个真实需求砸过来：

老板说： "给我做一个行业调研助手，输入一个行业关键词，自动出一份调研报告——包括行业概述、竞品分析、投资建议三个部分。"

单Agent搞不了——一个AI既要搜行业信息、又要分析竞品、又要写投资建议，结果每块都浅尝辄止，报告像流水账，缺乏深度。

单LangGraph工作流也不够——虽然流程可控，但每个节点是同一个AI在干活，没有专业分工。就像让一个人同时当产品经理、设计师、开发，啥都干啥都半吊子，难以产出高质量内容。

多Agent协作就是干这个的——让多个Agent各司其职，分工合作。 一个负责搜索调研，一个负责竞品分析，一个负责写报告。就像公司里不同岗位的人各干各的活，最后拼成一份完整的交付物，显著提升专业度。

花了2周从单Agent改到多Agent协作，踩了4个坑。每一个都是"多人协作"场景下才会遇到的问题，分享出来帮你少走弯路。

先说结论

用Java人的理解：单Agent ≈ 一个全栈开发啥都干；LangGraph工作流 ≈ 一个后端按流程先后干不同的事；多Agent ≈ 前端+后端+DBA分工协作，有接口契约，有数据传递。

简单事单Agent，流程事LangGraph，专业事多Agent。 这句总结值得记住，帮你快速决策技术选型。

先看全貌：3个Agent怎么协作

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

3个Agent，各有各的专属Prompt和工具：

调度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"])

关键点：

用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}

Prompt工程在多Agent场景下比单Agent更重要。 因为每个Agent的输出是下一个Agent的输入——一个Agent偷懒，后面的全受影响，必须用结构化模板约束。

多Agent Prompt设计原则：

用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个层次：

宁可出一份"有标注的半成品报告"，也不要出一份"看起来完整但数据有误的报告"。 诚实标注比错误输出更有价值。

完整代码：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个坑的总结

从单Agent到多Agent的演进路径

别一上来就搞多Agent。 先用单Agent验证需求，再用LangGraph控制流程，最后发现一个AI干不过来时，才拆成多Agent。过早拆分 = 过度设计 = 难以维护，循序渐进才是正解。

这和Java架构演进一样的道理： 单体 → 模块化 → 微服务，别一上来就搞微服务。多Agent协作也是同样的演进逻辑，从简到繁逐步优化。