CrewAI 深入学习：构建多智能体协作工作流

在之前的系列文章中，我们已经深入学习了 LangChain 基础框架、LangGraph 智能代理工作流、LlamaIndex 数据应用框架，以及最新的 MCP 互联互通标准。这些技术帮助我们构建了强大的单智能体应用，但面对越来越复杂的业务场景，单个 AI 智能体往往显得力不从心。

想象一下，当你需要完成一份市场调研报告时，你需要有人负责收集行业数据、有人负责分析竞争格局、有人负责撰写专业报告，甚至有人负责校对审核。如果能让多个专业的 AI 智能体像人类团队一样分工协作，那效率将会有质的飞跃。

这正是 CrewAI 要解决的问题 —— 它让我们能够快速构建专业的 AI 团队，让不同角色的智能体协同工作，共同完成复杂任务。

一、什么是 CrewAI？

CrewAI 是一个轻量级、高性能的 Python 框架，专为构建多智能体协作系统而设计。与其他框架不同，CrewAI 完全独立开发，不依赖 LangChain 等其他生态，却提供了极其简洁直观的 API，让开发者能够在几分钟内搭建起一个完整的 AI 协作团队。

CrewAI 的核心理念是模拟人类团队协作：就像一个公司里有市场部、研发部、测试部一样，我们可以为 AI 智能体分配不同的角色、目标和专业技能，让它们像真正的团队成员一样分工合作，共同完成目标。

截至 2025 年，CrewAI 已经拥有超过 4.7 万 GitHub Stars，成为企业级 AI 自动化的事实标准，全球超过 10 万开发者通过它构建了各种智能协作系统。

CrewAI 的核心优势

• 角色驱动设计：每个智能体都有明确的角色、目标和背景故事，让 AI 更专注于自己的专业领域

• 灵活的协作模式：支持顺序执行、层级管理等多种工作流模式

• 丰富的工具生态：内置搜索、网页抓取、代码执行等多种工具，也支持自定义工具

• 极简的 API 设计：几行代码就能搭建完整的多智能体系统，学习成本极低

• 生产就绪：企业级的安全性、可观测性和可扩展性

二、核心概念解析

在开始实战之前，我们需要先理解 CrewAI 的几个核心组件，它们就像搭建积木的基础模块：

1. Agent（智能体）

Agent 是 CrewAI 中最基础的单元，代表团队中的一个成员。每个 Agent 都有三个核心属性：

• Role（角色）：这个智能体的身份，比如 "资深程序员"、"市场研究员"

• Goal（目标）：这个智能体的工作目标，指导它的行为方向

• Backstory（背景故事）：智能体的背景设定，让它更好地理解自己的身份和工作方式

除此之外，Agent 还可以配置：

• Tools（工具）：智能体可以使用的工具，比如搜索、代码执行等

• Allow Delegation（是否允许委托）：是否允许这个智能体把任务委托给其他智能体

• Verbose（详细日志）：是否输出详细的执行日志，方便调试

2. Task（任务）

Task 是智能体需要完成的具体工作，每个任务都有：

• Description（描述）：任务的具体要求和内容

• Expected Output（预期输出）：任务完成后应该输出什么样的结果

• Agent（执行者）：由哪个智能体来执行这个任务

• Context（上下文）：这个任务依赖哪些其他任务的结果，实现任务之间的信息传递

3. Process（流程）

Process 定义了团队的协作模式，目前支持两种：

• Sequential（顺序执行）：任务按照定义的顺序一个接一个执行，前一个任务完成后才会开始下一个

• Hierarchical（层级管理）：自动创建一个管理者智能体，由它来规划任务、分配工作、验证结果，适合更复杂的项目

4. Crew（团队）

Crew 是最高层的组织单元，它把智能体、任务和流程组合在一起，形成一个完整的团队。当你定义好所有内容后，只需要调用 kickoff\(\) 方法，整个团队就会自动开始工作。

Crews vs Flows

CrewAI 还提供了 Flows 组件，用于处理结构化的工作流：

• Crews：适合开放的、需要自主探索的任务，比如研究、创意写作

• Flows：适合需要精确控制、确定性执行的流程，比如 API 编排、审批流程

三、环境搭建

在开始实战之前，我们先搭建开发环境。CrewAI 对 Python 版本要求是 3.10 以上，建议使用虚拟环境来管理依赖。

1. 创建虚拟环境

# 创建虚拟环境
conda create -n crewai-env python=3.13 -y

# 激活虚拟环境
conda activate crewai-env

2. 安装 CrewAI

# 安装 CrewAI 核心库
pip install crewai crewai-tools

# 如果需要使用本地模型（比如 Ollama），还需要安装对应的依赖
pip install langchain-ollama

3. 配置 API 密钥

CrewAI 默认使用 OpenAI 的模型，你需要配置 OpenAI 的 API 密钥：

# 设置环境变量
export OPENAI_API_KEY="your-openai-api-key"

如果你使用本地模型（比如 Ollama），可以这样配置：

export OPENAI_API_BASE="http://localhost:11434/v1"
export OPENAI_API_KEY="EMPTY"
export OPENAI_MODEL_NAME="qwen3"

四、实战教程 1：构建软件开发虚拟团队

我们先来做一个简单的例子：构建一个软件开发团队，包含程序员和测试工程师，让它们协作完成一个用户管理系统的开发。

完整代码

import os
from crewai import Agent, Task, Crew, Process

# 如果你使用本地 Ollama 模型，取消下面的注释
# os.environ['OPENAI_API_BASE'] = 'http://localhost:11434/v1'
# os.environ["OPENAI_API_KEY"] = "EMPTY"
# os.environ["OPENAI_MODEL_NAME"] = "openai/qwen3"

class DevelopmentCrew:
    def __init__(self, development_description, testing_description):
        # 1. 定义程序员智能体
        self.programmer = Agent(
            role="资深程序员",
            goal="编写高质量、可维护的代码，实现功能需求",
            backstory="你是一位经验丰富的程序员，精通多种编程语言和设计模式。" +
            "你注重代码质量，善于编写清晰、高效的代码，并且有很强的逻辑思维能力。",
            verbose=True,
            max_execution_time=300
        )
        
        # 2. 定义测试工程师智能体
        self.qa_engineer = Agent(
            role="测试工程师",
            goal="确保代码质量，发现并报告缺陷，验证功能完整性",
            backstory="你是一位严谨的测试工程师，对细节极其敏感。" +
            "你擅长设计测试用例，能够发现各种边界条件和潜在问题，" +
            "确保交付的代码达到高质量标准。",
            verbose=True,
            max_execution_time=300
        )
        
        # 3. 定义开发任务
        self.development_task = Task(
            description=development_description,
            agent=self.programmer,
            expected_output="完整的Python代码实现，包含所有要求的功能和必要的注释"
        )
        
        # 4. 定义测试任务，依赖开发任务的结果
        self.testing_task = Task(
            description=testing_description,
            agent=self.qa_engineer,
            expected_output="详细的测试报告，包含测试用例、发现的问题和改进建议",
            context=[self.development_task]
        )
    
    # 创建并运行开发团队
    def invoke(self):
        dev_crew = Crew(
            agents=[self.programmer, self.qa_engineer],
            tasks=[self.development_task, self.testing_task],
            process=Process.sequential,  # 顺序执行任务
            verbose=True
        )
        
        print("开始软件开发流程...")
        print("第一阶段：程序员进行功能开发")
        print("第二阶段：测试工程师进行质量验证")
        
        result = dev_crew.kickoff()
        
        print("\n开发流程完成，最终结果:")
        print(result)
        return result.raw

if __name__ == "__main__":
    # 定义任务需求
    dev_description = """
    开发一个用户管理系统，包含以下功能：
    1. 用户注册（用户名、邮箱、密码）
    2. 用户登录验证
    3. 用户信息查询
    4. 密码重置功能
    请使用Python编写清晰的代码，包含必要的注释和错误处理。
    """
    
    test_description = """
    对开发的用户管理系统进行全面测试：
    1. 设计测试用例覆盖正常流程和异常情况
    2. 进行边界值测试和错误处理测试
    3. 验证所有功能是否符合需求
    4. 提供详细的测试报告和改进建议
    """
    
    # 运行团队
    dev_crew = DevelopmentCrew(dev_description, test_description)
    dev_crew.invoke()

代码解析

1. 智能体定义：我们创建了两个智能体，一个是程序员，一个是测试工程师，每个都有自己的角色、目标和背景故事

2. 任务定义：开发任务由程序员执行，测试任务由测试工程师执行，并且测试任务依赖开发任务的结果

3. 团队组建：我们把两个智能体和两个任务组合成一个团队，使用顺序执行的流程

4. 启动执行：调用 kickoff\(\) 方法，团队就会自动开始工作

运行结果

当你运行这段代码后，你会看到：

1. 首先，程序员智能体开始工作，分析需求，编写用户管理系统的代码

2. 然后，测试工程师拿到程序员写的代码，开始设计测试用例，进行测试

3. 最后，测试工程师输出详细的测试报告，包括发现的问题和改进建议

整个过程就像一个真实的开发团队在工作一样，完全自动化完成！

五、实战教程 2：构建市场研究团队

接下来我们来做一个更复杂的例子，构建一个市场研究团队，包含研究员、数据分析师和报告撰写员，让它们协作完成一份完整的市场调研报告。这个例子我们还会用到 CrewAI 的工具，让智能体能够联网搜索信息。

完整代码

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# 配置搜索和网页抓取工具
# SerperDevTool 是一个联网搜索工具，需要配置 SERPER_API_KEY 环境变量
# export SERPER_API_KEY="your-serper-api-key"
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# 1. 定义研究员智能体
researcher = Agent(
    role='市场研究员',
    goal='收集和分析最新的行业趋势和竞争对手信息',
    backstory="""你是一位经验丰富的市场研究专家，擅长从各种数据源
    中提取有价值的商业洞察。你总是能发现别人忽略的重要信息。""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool]  # 给研究员配置搜索和网页抓取工具
)

# 2. 定义数据分析师智能体
analyst = Agent(
    role='数据分析师',
    goal='对收集的数据进行深度分析，提供可执行的商业建议',
    backstory="""你是一位数据分析专家，擅长从复杂数据中发现模式
    和趋势。你的分析总是精准且具有前瞻性。""",
    verbose=True,
    allow_delegation=False
)

# 3. 定义报告撰写员智能体
writer = Agent(
    role='商业报告撰写员',
    goal='将分析结果转化为清晰、有说服力的商业报告',
    backstory="""你是一位专业的商业写作专家，擅长将复杂的数据
    分析转化为易懂的商业洞察和建议。""",
    verbose=True,
    allow_delegation=False
)

# 定义任务
def create_market_research_tasks(topic: str):
    """创建市场研究任务序列"""
    
    # 数据收集任务
    research_task = Task(
        description=f"""
        对{topic}进行全面的市场研究，包括：
        1. 行业现状和发展趋势
        2. 主要竞争对手分析
        3. 目标客户群体分析
        4. 市场机会和威胁
        5. 相关技术发展动态
        请确保信息的准确性和时效性。
        """,
        agent=researcher,
        expected_output="详细的市场研究数据和初步分析"
    )
    
    # 数据分析任务
    analysis_task = Task(
        description="""
        基于收集的数据进行深度分析：
        1. 市场规模和增长潜力评估
        2. 竞争格局和优势分析
        3. 消费者需求和行为模式
        4. 技术趋势对市场的影响
        5. 投资价值和风险评估
        提供量化的分析结果和预测。
        """,
        agent=analyst,
        expected_output="深度数据分析报告和商业洞察"
    )
    
    # 报告撰写任务
    writing_task = Task(
        description="""
        基于研究和分析结果，撰写专业的市场研究报告：
        1. 执行摘要
        2. 市场概况和趋势分析
        3. 竞争对手深度分析
        4. 机会与挑战评估
        5. 战略建议和行动计划
        报告应该结构清晰，数据支撑充分，建议可执行。
        """,
        agent=writer,
        expected_output="完整的市场研究报告，包含战略建议"
    )
    
    return [research_task, analysis_task, writing_task]

# 创建团队
def create_research_crew(topic: str):
    """创建市场研究团队"""
    tasks = create_market_research_tasks(topic)
    
    crew = Crew(
        agents=[researcher, analyst, writer],
        tasks=tasks,
        process=Process.sequential,  # 按顺序执行任务
        verbose=2
    )
    return crew

# 执行市场研究
def conduct_market_research(topic: str):
    """执行完整的市场研究流程"""
    print(f"🔍 启动市场研究项目：{topic}")
    crew = create_research_crew(topic)
    result = crew.kickoff()
    print("✅ 市场研究完成！")
    return result

if __name__ == "__main__":
    # 研究AI编程工具市场
    research_topic = "AI编程工具和代码生成器市场"
    
    try:
        report = conduct_market_research(research_topic)
        print("\n" + "="*50)
        print("📊 市场研究报告")
        print("="*50)
        print(report)
        
        # 保存报告到文件
        with open("market_report.md", "w", encoding="utf-8") as f:
            f.write(str(report))
        print("\n📄 报告已保存到 market_report.md")
    except Exception as e:
        print(f"❌ 研究过程中发生错误：{e}")

这个例子的亮点

1. 工具的使用：我们给研究员配置了搜索工具和网页抓取工具，让它能够联网获取最新的市场信息

2. 多步骤协作：三个智能体分工明确，研究员负责收集数据，分析师负责分析数据，写作者负责整理报告

3. 自动化流程：整个过程完全自动化，你只需要输入研究主题，就能得到一份完整的专业报告

六、高级特性

1. YAML 配置方式

对于复杂的项目，CrewAI 支持用 YAML 文件来配置智能体和任务，这样可以更好地分离配置和代码：

# config/agents.yaml
researcher:
  role: >
    {topic} Senior Researcher
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You're a seasoned researcher with a knack for uncovering the latest
    developments in {topic}.

writer:
  role: >
    Tech Writer
  goal: >
    Create compelling content about {topic}
  backstory: >
    You're a technical writer specializing in AI topics.

# config/tasks.yaml
research_task:
  description: >
    Conduct a thorough research about {topic}
  expected_output: >
    A list with the most relevant findings
  agent: researcher

write_task:
  description: >
    Write an article about {topic}
  expected_output: >
    A full article with at least 3 paragraphs
  agent: writer

2. 自定义工具

除了内置的工具，你还可以很方便地创建自定义工具：

from crewai_tools import tool

@tool
def calculate_market_growth(market_size: float, growth_rate: float, years: int) -> str:
    """
    计算未来几年的市场规模增长
    
    Args:
        market_size: 当前市场规模（亿元）
        growth_rate: 年增长率（百分比）
        years: 预测年数
    """
    result = market_size
    for i in range(years):
        result = result * (1 + growth_rate / 100)
    
    return f"预测{years}年后市场规模将达到 {result:.2f} 亿元"

# 然后就可以把这个工具分配给智能体
analyst = Agent(
    ...
    tools=[calculate_market_growth]
)

3. 层级流程

对于更复杂的项目，你可以使用层级流程，CrewAI 会自动创建一个管理者来协调工作：

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.hierarchical,  # 层级管理模式
    manager_llm=ChatOpenAI(model="gpt-4")
)

在这种模式下，管理者会：

• 分析整体目标，制定执行计划

• 把任务分配给合适的智能体

• 验证每个任务的结果

• 协调智能体之间的协作

八、总结

在这篇文章中，我们深入学习了 CrewAI 框架，从核心概念到实战案例，一步步构建了两个完整的多智能体协作系统：软件开发团队和市场研究团队。

现在已经有很多成熟的多agent协作IDE 如Trae 本文主要是通过crewAI的学习了解构建多智能体协作工作流的原理性知识。
如果你对我们的系列文章感兴趣，可以回顾之前的内容：

• [MCP (Model Context Protocol) 深入学习：构建 AI Agent 互联互通标准]

• [LangGraph 深入学习：构建智能代理工作流]

• [LangChain 深入学习]

• [AI Agent 开发实战：LlamaIndex 框架深度应用指南]