Agent框架横评（四）：CrewAI让Multi-Agent协作像搭乐高一样简单

一、开场：为什么需要多Agent

大家好，我是老金。

单个Agent能力有限。

现实任务需要：

• 有人负责搜索
• 有人负责写作
• 有人负责审核

CrewAI让多Agent协作变得简单。

二、CrewAI是什么

2.1 定位

CrewAI是一个Multi-Agent协作框架，核心概念：

• Agent：具有特定角色的AI角色
• Task：分配给Agent的任务
• Crew：Agent团队
• Process：协作流程（顺序/并行/层级）

2.2 核心思想

┌─────────────────────────────────────────────────────────┐
│                 CrewAI核心思想                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│     Agent = 角色 + 目标 + 工具                          │
│                                                         │
│     Task = 描述 + 期望 + Agent分配                      │
│                                                         │
│     Crew = Agent团队 + Process + 协作模式               │
│                                                         │
└─────────────────────────────────────────────────────────┘

2.3 架构

┌─────────────────────────────────────────────────────────┐
│                  CrewAI架构                              │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │                    Crew                         │   │
│  │              编排多个Agent协作                   │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │                  Agents                        │   │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐        │   │
│  │  │ Researcher│  │ Writer  │  │ Editor  │        │   │
│  │  └─────────┘  └─────────┘  └─────────┘        │   │
│  └────────────────────┬────────────────────────────┘   │
│                       ↓                                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │                  Tasks                         │   │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐        │   │
│  │  │ Research │  │ Writing │  │ Editing │        │   │
│  │  └─────────┘  └─────────┘  └─────────┘        │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
└─────────────────────────────────────────────────────────┘

三、快速上手

3.1 安装

pip install crewai
pip install 'crewai[tools]'  # 包含内置工具

3.2 第一个Crew

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# 创建LLM
llm = ChatOpenAI(model="gpt-4")

# 创建Agent
researcher = Agent(
    role="研究员",
    goal="收集并分析最准确的信息",
    backstory="你是一个专业的研究员，擅长从多个来源收集信息。",
    llm=llm,
    verbose=True
)

writer = Agent(
    role="作家",
    goal="把复杂的信息写成易懂的文章",
    backstory="你是一个资深科技作家，擅长把技术内容写得通俗易懂。",
    llm=llm,
    verbose=True
)

# 创建Task
research_task = Task(
    description="研究AI Agent的最新发展趋势",
    agent=researcher,
    expected_output="一份详细的研究报告"
)

write_task = Task(
    description="基于研究报告，写一篇博客文章",
    agent=writer,
    expected_output="一篇1000字的博客文章"
)

# 创建Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential  # 顺序执行
)

# 启动
result = crew.kickoff()
print(result)

四、Agent详解

4.1 Agent定义

from crewai import Agent

agent = Agent(
    role="角色名称",           # Agent的角色
    goal="最终目标",           # Agent追求的目标
    backstory="背景故事",      # Agent的身份设定
    llm=llm,                  # 使用的LLM
    tools=[],                 # 可用工具
    verbose=False,            # 是否详细输出
    allow_delegation=False,    # 是否允许委托任务
    max_iter=5,               # 最大迭代次数
    max_rpm=10                # 每分钟最大请求数
)

4.2 内置工具

from crewai.tools import BaseTool
from crewai.tools import (
    SearchTool,
    FileReadTool,
    FileWriteTool,
    SerperDevTool,
    BrowserbaseLoadTool
)

# 搜索工具
search_tool = SearchTool()

# 文件读写
read_tool = FileReadTool()
write_tool = FileWriteTool()

# Serper搜索
serper_tool = SerperDevTool(
    api_key="xxx"
)

# 浏览器工具
browser_tool = BrowserbaseLoadTool(
    api_key="xxx",
    project_id="xxx"
)

# 在Agent中使用
agent = Agent(
    role="研究员",
    goal="提供准确的信息",
    tools=[search_tool, serper_tool]
)

4.3 自定义工具

from crewai.tools import BaseTool
from pydantic import Field

class MyCustomTool(BaseTool):
    name: str = "my_custom_tool"
    description: str = "执行自定义任务的工具"

    def _run(self, query: str) -> str:
        """同步执行"""
        # 你的逻辑
        result = do_something(query)
        return result

    async def _arun(self, query: str) -> str:
        """异步执行"""
        result = await do_something_async(query)
        return result

# 使用
custom_tool = MyCustomTool()

agent = Agent(
    role="分析师",
    goal="提供数据分析",
    tools=[custom_tool]
)

五、Task详解

5.1 Task定义

from crewai import Task

task = Task(
    description="具体任务描述",     # 任务描述
    agent=agent,                   # 分配的Agent（可选）
    expected_output="期望输出",     # 期望的输出格式
    tools=[],                     # 可用工具（可选）
    async_execution=False,         # 是否异步执行
    context=[],                   # 上下文任务（结果会传给后续任务）
    output_file="result.md"       # 输出到文件（可选）
)

5.2 任务依赖

# Task 1: 研究
research_task = Task(
    description="研究量子计算的最新进展",
    agent=researcher,
    expected_output="量子计算研究报告"
)

# Task 2: 写作（依赖研究结果）
writing_task = Task(
    description="基于研究报告写一篇文章",
    agent=writer,
    context=[research_task],  # 依赖research_task的结果
    expected_output="一篇完整的文章"
)

# Task 3: 审核（依赖写作结果）
review_task = Task(
    description="审核并改进文章",
    agent=editor,
    context=[writing_task],  # 依赖writing_task的结果
    expected_output="改进后的文章"
)

5.3 输出格式

# 指定输出格式
task = Task(
    description="生成数据分析报告",
    expected_output="""
    报告格式：
    1. 执行摘要
    2. 数据概览
    3. 关键发现
    4. 建议
    """,
    output_file="reports/analysis.md"
)

六、Crew详解

6.1 协作流程

from crewai import Crew, Process

# 顺序流程：一个接一个
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential
)

# 并行流程：同时执行不同任务
crew_parallel = Crew(
    agents=[researcher_1, researcher_2, researcher_3],
    tasks=[task_1, task_2, task_3],
    process=Process.hierarchical  # 层级
)

# 层级流程：Manager协调
crew_hierarchical = Crew(
    agents=[manager, researcher, writer],
    tasks=[...],
    process=Process.hierarchical,
    manager_agent=manager  # 指定Manager
)

6.2 层级流程

# Manager Agent
manager = Agent(
    role="项目经理",
    goal="协调团队高效完成任务",
    backstory="经验丰富的项目经理，擅长资源调配和进度控制"
)

# 层级Crew
crew = Crew(
    agents=[manager, researcher, writer, designer],
    tasks=[...],
    process=Process.hierarchical,
    manager_agent=manager,
    manager_llm=llm  # Manager专用LLM
)

6.3 Crew配置

crew = Crew(
    agents=[...],
    tasks=[...],
    process=Process.sequential,

    # 记忆功能
    memory=True,  # 启用记忆
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small"
    },

    # 共享知识
    shared_tasks=False,

    # Verbose
    verbose=2,  # 0=silent, 1=minimal, 2=full

    # 限制
    max_iter=15,
    max_rpm=30
)

七、实战案例

7.1 营销内容生成

from crewai import Agent, Task, Crew, Process

# 创建Agent
market_researcher = Agent(
    role="市场研究员",
    goal="了解目标受众和市场趋势",
    backstory="专业市场分析师，擅长用户调研和竞品分析",
    tools=[search_tool]
)

copywriter = Agent(
    role="文案撰写师",
    goal="创作吸引人的营销文案",
    backstory="资深营销文案专家，创造过多个爆款内容",
    tools=[]
)

designer = Agent(
    role="创意设计师",
    goal="提供视觉创意建议",
    backstory="创意总监，擅长视觉传达和品牌设计"
)

# 创建Tasks
research = Task(
    description="调研20-30岁年轻人对AI工具的态度和需求",
    agent=market_researcher,
    expected_output="用户调研报告"
)

write_copy = Task(
    description="基于调研报告，写一篇针对年轻人的产品推广文案",
    agent=copywriter,
    context=[research],
    expected_output="1000字营销文案"
)

design_ideas = Task(
    description="为文案提供视觉创意建议",
    agent=designer,
    context=[research, write_copy],
    expected_output="视觉创意方案"
)

# 创建Crew
crew = Crew(
    agents=[market_researcher, copywriter, designer],
    tasks=[research, write_copy, design_ideas],
    process=Process.sequential
)

# 启动
result = crew.kickoff()

7.2 软件开发团队

# 软件开发Crew
architect = Agent(
    role="架构师",
    goal="设计高质量的系统架构",
    backstory="10年架构经验，擅长微服务和云原生架构"
)

backend_dev = Agent(
    role="后端开发",
    goal="实现高效可靠的后端代码",
    backstory="Python专家，精通FastAPI和数据库"
)

frontend_dev = Agent(
    role="前端开发",
    goal="实现美观易用的用户界面",
    backstory="React高手，注重用户体验"
)

qa_engineer = Agent(
    role="测试工程师",
    goal="确保代码质量和稳定性",
    backstory="资深QA，擅长自动化测试"
)

# Tasks
design_task = Task(
    description="设计一个电商系统的微服务架构",
    agent=architect,
    expected_output="架构设计文档"
)

backend_task = Task(
    description="实现商品、订单、支付相关的后端API",
    agent=backend_dev,
    context=[design_task],
    expected_output="可运行的Python代码"
)

frontend_task = Task(
    description="实现用户端和管理端的界面",
    agent=frontend_dev,
    context=[backend_task],
    expected_output="React组件"
)

test_task = Task(
    description="为系统编写单元测试和集成测试",
    agent=qa_engineer,
    context=[backend_task, frontend_task],
    expected_output="测试代码覆盖率>80%"
)

crew = Crew(
    agents=[architect, backend_dev, frontend_dev, qa_engineer],
    tasks=[design_task, backend_task, frontend_task, test_task],
    process=Process.sequential
)

7.3 新闻分析团队

# 新闻分析Crew
news_collector = Agent(
    role="新闻收集员",
    goal="收集最新最相关的新闻",
    tools=[search_tool, serper_tool]
)

fact_checker = Agent(
    role="事实核查员",
    goal="验证信息准确性"
)

analyst = Agent(
    role="分析师",
    goal="提供深度洞察"
)

writer = Agent(
    role="编辑",
    goal="产出高质量文章"
)

# 并行收集，顺序处理
collect_task_1 = Task(description="收集AI领域今日新闻", agent=news_collector)
collect_task_2 = Task(description="收集科技行业动态", agent=news_collector)
collect_task_3 = Task(description="收集政策相关消息", agent=news_collector)

check_task = Task(
    description="核查收集到的信息真实性",
    agent=fact_checker,
    context=[collect_task_1, collect_task_2, collect_task_3]
)

analyze_task = Task(
    description="分析新闻背后的趋势和影响",
    agent=analyst,
    context=[check_task]
)

write_task = Task(
    description="写一篇新闻分析文章",
    agent=writer,
    context=[analyze_task]
)

crew = Crew(
    agents=[news_collector, fact_checker, analyst, writer],
    tasks=[collect_task_1, collect_task_2, collect_task_3, check_task, analyze_task, write_task],
    process=Process.sequential
)

八、记忆功能

8.1 启用记忆

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # 启用记忆
    embedder={
        "provider": "openai",
        "model": "text-embedding-3-small"
    }
)

8.2 记忆类型

# CrewAI使用多种记忆
CREW_MEMORY_TYPES = [
    "short_term",   # 短期：当前任务上下文
    "long_term",    # 长期：历史经验
    "entity",        # 实体：关键概念和关系
    "context"       # 上下文：对话历史
]

九、优缺点分析

9.1 优点

优点	说明
上手简单	概念清晰，5分钟上手
Multi-Agent友好	原生支持多Agent协作
流程灵活	顺序/并行/层级
记忆功能	内置记忆系统

9.2 缺点

缺点	说明
新框架	生态待完善
定制能力	部分场景受限
调试困难	多Agent问题难追踪

9.3 使用建议

# 最佳实践
BEST_PRACTICES = {
    "agent定义": {
        "role": "具体明确",
        "goal": "可衡量",
        "backstory": "有代入感"
    },
    "task定义": {
        "description": "清晰具体",
        "expected_output": "明确格式"
    },
    "crew配置": {
        "process": "根据场景选择",
        "memory": "长流程建议开启",
        "verbose": "开发时开启"
    }
}

十、总结

适用场景

场景	推荐度
内容创作团队	⭐⭐⭐⭐⭐
软件开发团队	⭐⭐⭐⭐
新闻分析	⭐⭐⭐⭐
营销策划	⭐⭐⭐⭐⭐

框架评分

维度	评分
易用性	9/10
Multi-Agent	9/10
灵活性	7/10
文档	7/10
生态	6/10

下期预告

下一篇：Agent Framework对比总结——如何选择适合你的框架？

---

相关阅读

• Agent框架横评（三）：LlamaIndex我的RAG知识库专家
• Agent框架横评（二）：LangChain我的LLM应用瑞士军刀
• Multi-Agent协作让团队工作