第7篇：CrewAI多智能体协作 —— 构建虚拟研究团队

当一个Agent的任务复杂度超出单兵作战的能力边界时，多Agent协作便成为必然选择。CrewAI正是为这种场景而生的框架——它通过“角色驱动”的设计哲学，让多个AI智能体像人类团队一样分工协作。本文从Agent、Task、Crew、Process四大核心组件入手，构建一个“研究员→撰稿人→审稿人→报告整合者”的研究助手团队，通过YAML配置与Python代码的双重视角，展示如何定义角色分工、任务依赖与执行流程，并深入分析CrewAI与LangGraph的设计差异。

---

一、为什么需要多智能体系统？

单一智能体在处理复杂任务时面临一个根本性的困境：一个模型在同一时刻只能有一个“思维重心”。以“撰写一份研究报告”为例，如果让单个Agent完成全部工作——调研、写作、审稿、整合——它必须在不同任务模式间反复切换，结果往往是每个环节都“及格但不出彩”。

多智能体系统的核心价值在于专业化分工：

• 专业化分工：每个智能体聚焦特定领域，研究员只负责搜集信息，撰稿人只负责文字表达
• 容错性提升：通过智能体间的交叉验证降低错误率，审稿人能发现撰稿人遗漏的逻辑漏洞
• 动态扩展：新增业务场景时只需添加对应智能体，无需重构整个系统
• 可观测性增强：通过智能体交互日志实现精准故障定位

CrewAI正是围绕这一理念设计的框架。它不试图提供一个“万能Agent”，而是提供一套工具，让你能够组装一个由专业Agent构成的虚拟团队。

---

二、CrewAI的四大核心组件

CrewAI的架构围绕四个核心概念展开：

组件	含义	类比
Agent（智能体）	具有特定角色、目标和背景故事的执行单元	团队中的“员工”
Task（任务）	分配给Agent的具体工作，包含描述和预期输出	员工要完成的“工作项”
Crew（团队）	Agent和Task的集合，定义整体协作策略	整个“项目组”
Process（流程）	任务执行的编排方式（顺序/层级）	“工作流程”

2.1 Agent：定义你的“员工”

Agent是CrewAI中最基础的执行单元。每个Agent通过三个要素定义其行为边界：

• Role（角色） ：Agent的身份标签，如“高级研究员”或“专业撰稿人”。角色决定了Agent的思维框架。
• Goal（目标） ：Agent要达成的终极目的，如“针对指定主题搜索并收集最新资料”。目标是Agent决策的“北极星”。
• Backstory（背景故事） ：Agent的“人设”，为其行为提供上下文。例如，“你是一位经验丰富的研究员，擅长从海量数据中提炼核心观点”。

这三个要素共同塑造了Agent的行为模式，使其在不同场景下能够“进入角色”。

2.2 Task：定义“工作项”

Task是Agent需要完成的具体工作。一个Task包含以下关键属性：

属性	说明
description	任务的详细描述，Agent据此理解要做什么
expected_output	预期输出的格式和内容描述
agent	负责执行该任务的Agent
context	依赖的其他Task的输出，用于构建任务间的数据流
tools	该任务可用的工具列表

Task的设计原则是：描述要足够具体，让Agent知道“做成什么样算成功”。模糊的Task描述是Agent输出质量不佳的最常见原因。

2.3 Process：定义“协作方式”

Process决定了Task的执行策略。CrewAI支持两种Process：

Process	行为	适用场景
Sequential（顺序）	任务按定义顺序依次执行，前一个Task的输出作为后一个Task的上下文	流程清晰的流水线任务
Hierarchical（层级）	由一个Manager Agent动态分配任务给其他Agent，根据能力匹配和输出质量进行委派	复杂决策、需要动态调整的任务

Sequential适合流程固定的场景，如“研究→写作→审稿→整合”。Hierarchical则更适合需要动态决策的场景——Manager Agent会根据Agent的能力和当前进度动态调整任务分配。

2.4 Crew：组装“团队”

Crew是将Agent、Task和Process组装在一起的容器。它定义了：

• 哪些Agent参与协作
• 哪些Task需要完成
• 以什么Process执行
• 执行过程中的各种配置（缓存、日志、记忆等）

---

三、实战：构建“研究助手团队”

我们以“研究助手团队”为例，展示如何从零构建一个CrewAI应用。这个团队包含四个Agent，分别负责研究、写作、审稿和报告整合。

3.1 项目结构与初始化

使用CrewAI CLI创建项目骨架：

crewai create crew research_crew
cd research_crew

CLI生成的项目结构如下：

research_crew/
├── pyproject.toml
├── .env
└── src/
    └── research_crew/
        ├── __init__.py
        ├── main.py          # 入口脚本
        ├── crew.py          # Crew定义
        ├── tools/           # 自定义工具
        └── config/
            ├── agents.yaml  # Agent配置
            └── tasks.yaml   # Task配置

关键设计理念：CrewAI将配置（YAML）与实现（Python）分离。Agent和Task的定义写在YAML文件中，Python代码只负责加载和执行。这样修改Agent的角色描述或Task的预期输出时，不需要改动任何Python代码。

3.2 定义Agent（agents.yaml）

# config/agents.yaml
researcher:
  role: >
    高级研究员
  goal: >
    针对用户指定的主题，搜索并收集最相关、最新的资料和信息
  backstory: >
    你是一位经验丰富的研究员，擅长使用搜索引擎快速定位关键信息。
    你注重信息的准确性和权威性，能够从大量数据中提炼出核心观点。
  verbose: true
  allow_delegation: false

writer:
  role: >
    专业撰稿人
  goal: >
    根据研究员提供的资料，撰写一篇结构清晰、语言流畅的文章草稿
  backstory: >
    你是一位才华横溢的作家，能够将复杂的技术内容转化为通俗易懂的文字。
    你擅长组织文章结构，使逻辑连贯且引人入胜。
  verbose: true
  allow_delegation: false

critic:
  role: >
    资深审稿人
  goal: >
    审阅撰稿人提供的草稿，提出具体的修改建议和改进意见
  backstory: >
    你是一位严谨的编辑，拥有多年内容审核经验。你能够发现逻辑漏洞、语法错误，
    并给出建设性的改进方向。
  verbose: true
  allow_delegation: false

reporter:
  role: >
    报告整合专家
  goal: >
    综合研究员、撰稿人和审稿人的成果，生成最终的结构化研究报告
  backstory: >
    你是一位专业的报告撰写专家，善于将分散的信息整合成一份完整、专业的报告。
    你注重格式规范和可读性，输出Markdown格式的报告。
  verbose: true
  allow_delegation: false

3.3 定义Task（tasks.yaml）

# config/tasks.yaml
research_task:
  description: >
    对主题“{topic}”进行深入研究，收集至少5条高质量的资料摘要。
    每条摘要需包含来源和核心观点。
  expected_output: >
    一个包含至少5个条目的列表，每个条目格式为：
    - 来源：URL或文献名
    - 观点：一句话总结
  agent: researcher

write_task:
  description: >
    基于研究员收集的资料，撰写一篇约500字的文章草稿。
    文章需包含引言、主体（分段论述）和结论。
  expected_output: >
    一篇结构完整的Markdown格式文章草稿。
  agent: writer
  context:
    - research_task

critic_task:
  description: >
    审阅撰稿人提供的文章草稿，从逻辑、语法、可读性三个方面提出至少3条具体修改建议。
  expected_output: >
    一个包含建议列表的文本，每条建议格式为“- 建议内容”。
  agent: critic
  context:
    - write_task

report_task:
  description: >
    根据研究员资料、撰稿人草稿和审稿人建议，整合生成一份最终的研究报告。
    报告应包含：标题、摘要、引言、详细内容、结论、参考文献。
  expected_output: >
    一份完整的Markdown格式研究报告，结构清晰，引用规范。
  agent: reporter
  context:
    - research_task
    - write_task
    - critic_task

关键设计：context字段定义了Task之间的依赖关系。CrewAI会根据context自动构建执行顺序——write_task依赖research_task的输出，因此research_task会先执行，其输出作为write_task的上下文传入。

3.4 加载并执行Crew（crew.py）

# crew.py
import os
import yaml
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

class ResearchCrew:
    def __init__(self):
        self.llm = ChatOpenAI(
            model="deepseek-chat",
            openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
            openai_api_base="https://api.deepseek.com/v1",
            temperature=0.7
        )
        self.agents = self._load_agents()
        self.tasks = None
    
    def _load_agents(self):
        """从YAML加载Agent定义"""
        with open("config/agents.yaml", "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)
        
        agents = {}
        for name, cfg in config.items():
            agents[name] = Agent(
                role=cfg["role"],
                goal=cfg["goal"],
                backstory=cfg["backstory"],
                llm=self.llm,
                verbose=cfg.get("verbose", True),
                allow_delegation=cfg.get("allow_delegation", False)
            )
        return agents
    
    def _load_tasks(self, topic: str):
        """从YAML加载Task定义，动态替换{topic}占位符"""
        with open("config/tasks.yaml", "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)
        
        tasks = []
        task_map = {}
        for name, cfg in config.items():
            # 替换{topic}占位符
            description = cfg["description"].replace("{topic}", topic)
            
            # 解析context依赖
            context = []
            if "context" in cfg:
                for ctx_name in cfg["context"]:
                    if ctx_name in task_map:
                        context.append(task_map[ctx_name])
            
            task = Task(
                description=description,
                expected_output=cfg["expected_output"],
                agent=self.agents[cfg["agent"]],
                context=context if context else None
            )
            tasks.append(task)
            task_map[name] = task
        return tasks
    
    def run(self, topic: str):
        """执行研究团队"""
        self.tasks = self._load_tasks(topic)
        
        crew = Crew(
            agents=list(self.agents.values()),
            tasks=self.tasks,
            process=Process.sequential,
            verbose=2
        )
        
        result = crew.kickoff()
        return result

if __name__ == "__main__":
    crew = ResearchCrew()
    result = crew.run("人工智能在医疗领域的应用")
    print("\n" + "="*60)
    print("📄 最终研究报告")
    print("="*60)
    print(result)

3.5 执行流程

当crew.kickoff()被调用时，CrewAI按以下顺序执行：

1. 解析依赖图：根据Task的context字段构建执行顺序图
2. 顺序执行：
   • research_task → 研究员Agent搜索资料
   • write_task → 撰稿人Agent基于研究员输出撰写草稿
   • critic_task → 审稿人Agent审阅草稿并提出建议
   • report_task → 报告整合者Agent综合所有输入生成最终报告
3. 结果传递：每个Task的输出自动作为下一个Task的context输入
4. 返回最终结果：report_task的输出作为整个Crew的返回值

---

四、CrewAI vs LangGraph：两种编排哲学的碰撞

CrewAI和LangGraph是当前多Agent编排领域最具代表性的两个框架，但它们的设计哲学截然不同：

维度	CrewAI	LangGraph
编排方式	角色驱动的声明式配置	图结构的过程式编程
核心抽象	Agent、Task、Crew	State、Node、Edge
流程控制	通过Task的context隐式构建依赖	通过条件边显式控制分支和循环
配置方式	YAML配置（推荐）	Python代码
灵活性	中等（结构清晰但扩展受限）	极高（可自由控制每一步）
适用场景	标准化协作流程	复杂决策系统

CrewAI的优势在于“开箱即用”。你只需要定义“谁做什么”，框架自动处理任务间的数据传递和执行顺序。代码量少，结构清晰，非常适合构建标准化的多Agent流水线。

LangGraph的优势在于“精细控制”。你可以显式定义状态转换、条件分支、循环重试，甚至可以插入人工审核节点。适合构建需要复杂决策逻辑的系统。

如何选择？

• 如果任务是固定流程的流水线（如研究→写作→审稿→报告），CrewAI是最自然的选择
• 如果任务涉及动态决策、循环迭代、条件分支（如需要根据中间结果决定下一步），LangGraph更合适

---

五、进阶实践：工具集成与任务委托

5.1 为Agent集成搜索工具

from crewai.tools import BaseTool

class BochaSearchTool(BaseTool):
    name: str = "博查搜索"
    description: str = "搜索互联网信息，输入关键词返回摘要和链接"
    
    def _run(self, query: str) -> str:
        # 调用博查API实现搜索
        # ...
        return search_results

在Agent定义中绑定工具：

search_tool = BochaSearchTool()
researcher = Agent(
    role="高级研究员",
    tools=[search_tool],  # 研究员拥有搜索能力
    # ...
)

5.2 任务委托（allow_delegation）

当allow_delegation=True时，Agent可以将子任务委托给其他Agent：

researcher = Agent(
    role="高级研究员",
    allow_delegation=True,  # 允许委托
    # ...
)

这赋予了Agent自主决策的能力——它可以根据任务复杂度决定是否需要求助其他Agent。例如，研究员在搜索过程中发现需要数据分析，可以委托给数据分析Agent。

---

六、总结：CrewAI的核心价值

CrewAI的价值不在于提供一个“更强大的Agent”，而在于提供一套让多个专业Agent高效协作的框架。其核心设计原则是：

1. 角色即边界：通过Role、Goal、Backstory定义Agent的专业领域，避免职责模糊
2. 配置即文档：YAML配置文件本身就是团队的工作说明书
3. 依赖即流程：Task的context字段天然定义了执行顺序，无需额外编排代码

当你需要构建一个多步骤、多角色的协作系统时，CrewAI提供了一条从“定义角色”到“执行流程”的最短路径。

思考与动手建议：

1. 使用CrewAI CLI创建一个新项目，在agents.yaml中定义三个不同角色的Agent（如“产品经理”、“开发者”、“测试工程师”），观察它们如何协作完成一个简单任务。
2. 尝试将Process从sequential改为hierarchical，观察Manager Agent如何动态分配任务。
3. 为Researcher Agent集成一个真实的搜索API（如博查或DuckDuckGo），对比有无搜索工具时研究报告的质量差异。