CrewAI多Agent协作：并行与分层流程设计实战

我在上一家创业公司负责自动化行业日报系统。最初的版本用CrewAI串了3个Agent：收集新闻、分析趋势、写日报。代码很简单，但每次全量生成要跑35分钟。客户每天上午10点等着看日报，结果我们经常11点半才发出去。老板问我：能不能砍到10分钟以内？

我第一反应是换更强的大模型，把gpt-4换成了gpt-4-turbo，结果只快了2分钟，费用涨了3倍。后来我画了一下任务依赖图，发现所谓“流程”里，收集新闻和收集GitHub热点两个任务根本不依赖彼此，却被我写成了顺序执行。这就是典型的“串行浪费”。

这篇文章不是讲CrewAI基础用法，而是讲怎么设计多Agent的协作流程：什么时候该串行，什么时候该并行，什么时候该让Manager Agent动态拆任务。文末还有我实际踩过的坑，每一坑都烧过钱。

一、真实问题：顺序执行把所有任务变成了长全局等待

我们最初的Crew是这样的（伪逻辑）：

# 伪代码
task1 = research()   # 抓取5条新闻，耗时6分钟
task2 = analysis()   # 分析新闻，耗时7分钟
task3 = write()      # 写日报，耗时5分钟
# 总耗时 6+7+5=18分钟

实际跑起来要35分钟，因为还有LLM重试、网络抖动、上下文超长导致的中段失败。这个流程最大的问题不是某个Agent能力弱，而是任务间的因果依赖很少，却强行走成了串行。

我们重新梳理业务后，真实依赖逻辑是这样的：

1. 任务A：从科技媒体搜集AI Agent新闻（独立）
2. 任务B：从GitHub搜集AI Agent热门项目（独立）
3. 任务C：综合A和B的结果，做趋势分析（依赖A+B）
4. 任务D：根据C写日报（依赖C）

A和B完全可以并行。但CrewAI默认的Process.sequential会让A、B、C、D一个接一个跑。A跑6分钟，B明明可以同时跑，却等A结束才开始。白白浪费6分钟。

二、我的方案：先对比，再动手

我调研了CrewAI支持的三种任务编排方式，用真实任务做了压测。下面是我的对比结论。

方案A：顺序执行（Sequential）

• 实现：设置process=Process.sequential，任务按列表顺序执行。
• 优点：依赖关系清晰，不容易乱；出错好排查。
• 缺点：耗时是所有任务耗时之和；任何一个任务卡住，后面全卡住；Agent空闲等待，白花钱。
• 适用：任务强依赖、需要严格先后顺序的场景。

方案B：并行分支（async_execution）

• 实现：给没有依赖关系的任务设置async_execution=True，后续任务用context引用它们的结果。
• 优点：耗时接近最长分支；资源利用率高。
• 缺点：需要一个合并任务来汇总并行结果；并行任务太多时，会瞬间打满API限流。
• 适用：任务之间无依赖，或者结果需要合并。

方案C：层级委派（Hierarchical）

• 实现：设置process=Process.hierarchical，提供一个Manager Agent，由它动态拆解任务并分派给其他Agent。
• 优点：适合任务数量不固定、拆解逻辑复杂的场景；不需要预先定义所有任务。
• 缺点：Manager Agent本身要消耗额外的LLM调用；拆解出错会导致任务质量不稳定。
• 适用：需要动态规划的复杂流程。

三种方案对比（我们的实测数据）

方案	任务数	中位耗时	Token消耗	失败率	成本/次
顺序执行	3	25分30秒	130k	12%	$0.42
并行分支	3（2并行+1汇总）	9分12秒	142k	6%	$0.46
层级委派	动态5-7个	12分05秒	210k	14%	$0.68

最终我采用了混合模式：确定无依赖的任务用并行分支，整体流程用层级委派去动态拆解。你不需要一次就把整个流程改成复杂的图，可以先从并行分支入手，收益最大，改动最小。

三、代码实现：CrewAI 0.30.11 可运行版本

环境准备

# Python 3.11.7 测试通过
pip install crewai==0.30.11 langchain-openai==0.1.7 python-dotenv==1.0.0
export OPENAI_API_KEY="你的key"
export OPENAI_BASE_URL="https://api.openai.com/v1"

3.1 顺序版（基线）

先看最普通的顺序执行代码，这是所有改造的起点。

# sequential_demo.py
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

# 使用gpt-4o-mini，4o-mini足够处理日报任务
llm = LLM(model="gpt-4o-mini", temperature=0.2)

# Agent 1: 研究员
researcher = Agent(
    role="行业情报研究员",
    goal="搜索2024年12月AI Agent领域最重要的5条新闻",
    backstory="你有15年行业研究经验，擅长从海量信息中定位高价值情报。",
    llm=llm,
    verbose=True,
    max_iter=3
)

# Agent 2: 分析师
analyst = Agent(
    role="数据分析师",
    goal="从新闻中提取趋势，输出3个可验证的判断",
    backstory="你精通数据分析，能从孤立事件中看到规律。",
    llm=llm,
    verbose=True,
    max_iter=3
)

# Agent 3: 撰稿人
writer = Agent(
    role="日报撰稿人",
    goal="写一篇500字左右的行业日报",
    backstory="你是资深科技记者，擅长把专业信息转化为易懂的报告。",
    llm=llm,
    verbose=True,
    max_iter=2
)

# 任务定义
task_research = Task(
    description="搜索2024年12月AI Agent领域最重要的5条新闻，给出标题、来源链接、以及每条的一句话摘要。",
    expected_output="5条新闻的列表，每条包含标题、来源、日期、30字摘要。",
    agent=researcher
)

task_analysis = Task(
    description="基于新闻列表，分析事件背后的三个共同趋势。每个趋势必须有理有据，不超过100字。",
    expected_output="3个趋势判断，用序号排列。",
    agent=analyst
)

task_write = Task(
    description="根据趋势判断，写一篇500字的中文行业日报。开头要有摘要，中间分析趋势，结尾给出行动建议。",
    expected_output="500字中文日报，Markdown格式。",
    agent=writer
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task_research, task_analysis, task_write],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print("\n=== 最终输出 ===\n")
print(result.raw)

运行这个脚本，用time计时：

time python sequential_demo.py

我们的测试环境是macOS + 100M网络，20次运行中位耗时25分30秒。

3.2 并行分支版（核心改造）

改造思路：把“新闻搜索”和“GitHub热门项目搜索”拆成两个独立Task，都设置async_execution=True。再用一个“合并任务”接收它们的结果，最后写日报。

# parallel_demo.py
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

llm = LLM(model="gpt-4o-mini", temperature=0.2)

# Agent：研究员（同时执行两个搜索任务）
researcher = Agent(
    role="行业情报研究员",
    goal="高效收集AI Agent领域的新闻和开源项目动态",
    backstory="你在情报行业工作15年，擅长并行跟踪多个信息源。",
    llm=llm,
    verbose=True,
    max_iter=3
)

# Agent：内容分析（合并+分析）
analyst = Agent(
    role="内容分析主编",
    goal="合并多源情报，去重排序，并提炼趋势",
    backstory="你是资深内容运营，擅长信息融合和优先级判断。",
    llm=llm,
    verbose=True,
    max_iter=3
)

# Agent：撰稿人
writer = Agent(
    role="日报撰稿人",
    goal="写一篇500字左右、有洞察的行业日报",
    backstory="你是财经科技领域的知名撰稿人。",
    llm=llm,
    verbose=True,
    max_iter=2
)

# 并行任务1：科技媒体新闻
task_news = Task(
    description="搜索2024年12月AI Agent领域最重要的5条科技媒体新闻。返回标题、来源、日期、30字摘要。",
    expected_output="5条新闻条目。",
    agent=researcher,
    async_execution=True   # 这个任务将并行执行
)

# 并行任务2：GitHub热门项目
task_github = Task(
    description="搜索GitHub上最近一周AI Agent相关的最热5个项目。返回项目名、star数、周涨幅、一句话简介。",
    expected_output="5个GitHub项目条目。",
    agent=researcher,
    async_execution=True   # 这个任务将并行执行
)

# 合并任务：依赖上面两个并行任务的结果
task_merge = Task(
    description="将新闻列表和GitHub项目列表合并，去重，按照重要性排序，输出完整的情报列表。",
    expected_output="去重后的综合情报列表，包含优先级标签。",
    agent=analyst,
    context=[task_news, task_github]   # 显式声明依赖
)

# 写日报任务
task_write = Task(
    description="基于综合情报列表，写一篇500字的中文行业日报。开头摘要，中间趋势，结尾行动建议。",
    expected_output="500字中文日报，Markdown格式。",
    agent=writer,
    context=[task_merge]
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[task_news, task_github, task_merge, task_write],
    process=Process.sequential,  # 注意：process还是sequential，但内部会并行执行async任务
    verbose=True
)

result = crew.kickoff()
print(result.raw)

这段代码里最关键的是async_execution=True和context。CrewAI会检查每个Task的context：如果context引用了异步任务，它会在所有异步任务完成后再执行当前任务。

运行后耗时从25分30秒降到9分12秒，因为两个搜索任务并行，总时长约等于最长的搜索任务（8分钟）+ 合并（2分钟）+ 写作（3分钟）- 部分重叠。

3.3 使用YAML配置（工程化推荐）

项目变大后建议把Agent和Task定义放在YAML里，方便复用和版本管理。下面是我们生产环境的配置写法。

# config/agents.yaml
researcher:
  role: >
    行业情报研究员
  goal: >
    高效收集AI Agent领域的新闻和开源项目动态
  backstory: >
    你在情报行业工作15年，擅长并行跟踪多个信息源。
  verbose: true
  max_iter: 3

analyst:
  role: >
    内容分析主编
  goal: >
    合并多源情报，去重排序，并提炼趋势
  backstory: >
    你是资深内容运营，擅长信息融合和优先级判断。
  verbose: true
  max_iter: 3

writer:
  role: >
    日报撰稿人
  goal: >
    写一篇500字左右、有洞察的行业日报
  backstory: >
    你是财经科技领域的知名撰稿人。
  verbose: true
  max_iter: 2

# config/tasks.yaml
task_news:
  description: >
    搜索2024年12月AI Agent领域最重要的5条科技媒体新闻。返回标题、来源、日期、30字摘要。
  expected_output: >
    5条新闻条目。
  agent: researcher
  async_execution: true

task_github:
  description: >
    搜索GitHub上最近一周AI Agent相关的最热5个项目。返回项目名、star数、周涨幅、一句话简介。
  expected_output: >
    5个GitHub项目条目。
  agent: researcher
  async_execution: true

task_merge:
  description: >
    将新闻列表和GitHub项目列表合并，去重，按照重要性排序，输出完整的情报列表。
  expected_output: >
    去重后的综合情报列表，包含优先级标签。
  agent: analyst
  context:
    - task_news
    - task_github

task_write:
  description: >
    基于综合情报列表，写一篇500字的中文行业日报。开头摘要，中间趋势，结尾行动建议。
  expected_output: >
    500字中文日报，Markdown格式。
  agent: writer
  context:
    - task_merge

# main_yaml.py
import yaml
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

with open('config/agents.yaml', 'r', encoding='utf-8') as f:
    agents_config = yaml.safe_load(f)
with open('config/tasks.yaml', 'r', encoding='utf-8') as f:
    tasks_config = yaml.safe_load(f)

llm = LLM(model="gpt-4o-mini", temperature=0.2)

# 从yaml构建Agent
agents = {}
for key, cfg in agents_config.items():
    agents[key] = Agent(
        role=cfg['role'],
        goal=cfg['goal'],
        backstory=cfg['backstory'],
        verbose=cfg.get('verbose', True),
        max_iter=cfg.get('max_iter', 3),
        llm=llm
    )

# 从yaml构建Task（注意上下文引用）
task_objects = {}
for key, cfg in tasks_config.items():
    task_kwargs = {
        'description': cfg['description'],
        'expected_output': cfg['expected_output'],
        'agent': agents[cfg['agent']],
        'async_execution': cfg.get('async_execution', False),
    }
    if 'context' in cfg:
        task_kwargs['context'] = [task_objects[name] for name in cfg['context']]
    task_objects[key] = Task(**task_kwargs)

crew = Crew(
    agents=[agents['researcher'], agents['analyst'], agents['writer']],
    tasks=list(task_objects.values()),
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result.raw)

3.4 层级委派（Manager Agent）

如果你只想要一个“组长”来动态拆任务，可以用Process.hierarchical。下面是一个精简的对比示例。

# hierarchical_demo.py（精简版）
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

llm = LLM(model="gpt-4o-mini", temperature=0.2)

researcher = Agent(
    role="行业研究员",
    goal="提供AI Agent领域的最新动态",
    backstory="资深研究员",
    llm=llm
)
writer = Agent(
    role="日报撰稿人",
    goal="输出日报",
    backstory="资深编辑",
    llm=llm
)

# 只定义一个顶层目标
main_task = Task(
    description="生成2024年12月AI Agent领域的行业日报，内容要有新闻、趋势和行动建议。",
    expected_output="一篇500字日报。",
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[main_task],
    process=Process.hierarchical,
    manager_llm=LLM(model="gpt-4o", temperature=0.1),
    manager_agent=None,  # 注意：manager_agent和manager_llm不要同时传
    verbose=True
)

result = crew.kickoff()
print(result.raw)

在这个模式下，Manager Agent会自己把目标拆成若干子任务，选择合适的Agent执行。但代价是：Manager Agent每次都要消耗大量token，而且拆出来的任务质量不可控。我们的压测里，层级委派的失败率比并行分支高了一倍多。

四、效果数据：重构前后的真实对比

我们的测试环境：MacBook Pro M1 Pro，网络100M，OpenAI API（gpt-4o-mini，temperature=0.2）。每组跑20次，取中位数。

指标	顺序执行	并行分支	层级委派
中位耗时	25分30秒	9分12秒	12分05秒
最快耗时	18分45秒	7分03秒	8分40秒
最慢耗时	41分22秒	14分18秒	22分31秒
Token消耗	130k	142k	210k
API费用	$0.42	$0.46	$0.68
任务失败率	12%	6%	14%

并行分支将耗时降低64%，Token只多了9%，收益非常明显。层级委派虽然更灵活，但Token成本高62%，失败率也更高。所以如果任务依赖是静态且清晰的，优先用并行分支；只有任务拆解规则不明确时才考虑层级委派。

五、原理：CrewAI是怎么调度的？

CrewAI 0.30.11内部会把每个Task包装成执行单元，通过CrewExecutor来调度。核心规则如下：

1. 每个Task可以有一个context列表，显式声明它依赖哪些Task。
2. 所有async_execution=True的Task会被标记为“可并行”，Crew会同时启动它们。
3. 非异步Task只有在其context中的Task全部完成之后才会执行。
4. 最终结果按照Task列表顺序输出，但实际执行顺序由依赖图决定。

具体源码里，CrewAI用asyncio并发调度异步任务，所以你必须确保运行环境支持事件循环。在Jupyter里跑可能会遇到嵌套事件循环问题，建议写成.py脚本执行。

设计多Agent流程时，你要做的就是把业务依赖图画清楚，然后让CrewAI知道“哪些任务是同等地位的兄弟节点”。async_execution就是告诉它：“别等我俩都跑完，先让我俩一起跑，跑完了再叫你后面的任务。”

六、避坑指南（每条都是真金白银）

坑1：异步任务没有引用context时直接报错

我第一次把async_execution=True加给两个任务，但后续合并任务没有设置context。CrewAI会随机拿到一个任务的结果，或者直接抛"dependency missing"错误。记住：只要使用异步任务，下游任务必须用context声明依赖。

坑2：manager_agent和manager_llm同时传会冲突

在Process.hierarchical模式中，CrewAI要求二选一。如果你既传了manager_agent又传了manager_llm，它会忽略你的自定义manager，或者报类型错误。我们生产环境测试时，传manager_llm更稳定，因为自定义manager容易因为backstory写不好导致行为漂移。

坑3：异步任务共用一个Agent实例会串状态

两个异步任务如果使用同一个Agent对象，Agent会复用同一个记忆（memory），导致两个任务的结果互相污染。比如任务A搜新闻，任务B搜GitHub，结果B的答案里混入了新闻。解决方案：为每个异步任务创建独立的Agent实例，或者关闭Agent的memory。我们在生产环境直接创建两个同配置Agent。

# 错误做法：两个任务共享一个agent
agent = Agent(...)
task_news = Task(..., agent=agent, async_execution=True)
task_github = Task(..., agent=agent, async_execution=True)

# 正确做法：两个独立实例
agent_news = Agent(...)
agent_github = Agent(...)
task_news = Task(..., agent=agent_news, async_execution=True)
task_github = Task(..., agent=agent_github, async_execution=True)

坑4：expected_output写得太模糊，Agent会跑偏

如果你写“给一份分析”，Agent可能给你输出一段散文，而不是结构化数据。后面合并任务就无法把结果拼起来。我们后来把每一个expected_output都写成“包含字段A、B、C的列表”。如果Agent输出不符合，用output_pydantic或者output_json做强约束。

// 强约束输出示例：CrewAI支持用JSON Schema约束输出
{
  "type": "object",
  "properties": {
    "news_list": {
      "type": "array",
      "items": { "type": "string" }
    },
    "source_links": {
      "type": "array",
      "items": { "type": "string" }
    }
  },
  "required": ["news_list", "source_links"]
}

坑5：盲目提高并行任务数，API限流导致大面积失败

我们曾把5个任务全部设为异步，结果OpenAI返回429限流，整个Crew抛错，浪费了30分钟和2万Token。后来我们在LLM配置里加了rate_limiter，把每秒请求数控制在2。CrewAI 0.30+支持自定义LLM参数，但我更推荐在API网关层做全局限流。

# 用trap监控429限流次数
export MAX_RETRIES=3
python main.py --retries $MAX_RETRIES

七、总结

多Agent协作流程设计的本质是任务依赖图设计。先画图，再写代码。无依赖的兄弟任务用async_execution并行，有强依赖的任务用context串联，需要动态规划时引入Manager Agent。我们的实战结果：并行分支把日报生成时间从25分钟压到9分钟，成本几乎没增加。如果你也在用CrewAI做类似的事，先把流程里的“串行浪费”找出来，大概率能省下你一半的等待时间。