CrewAI 项目分析：多 Agent 协作框架

什么是 CrewAI

CrewAI 是另一个非常受欢迎的多智能体协作框架，在 GitHub 上拥有超过 45,000 颗星。与 MetaGPT 的”软件公司”理念不同，CrewAI 的设计哲学更加灵活：它让 AI Agents 像真实的团队成员一样协作，每个 Agent 可以有自己的专业技能、工具和任务。

简单来说，CrewAI 的核心概念是：

• Crew（团队）：多个 Agent 组成的协作单位
• Agent（智能体）：具有特定角色的独立个体
• Task（任务）：需要完成的具体工作
• Process（流程）：Agent 之间的协作方式

CrewAI 的优势在于它的简洁性和灵活性。你可以快速定义一个 Agent，给它分配工具，然后让它和其他 Agent 协作完成任务。无论你是想做一个研究助手、内容生成器，还是复杂的自动化工作流，CrewAI 都能提供良好的支持。

---

核心概念

1. Agent（智能体）

在 CrewAI 中，每个 Agent 都有以下核心属性：

from crewai import Agent

# 定义一个研究员 Agent
researcher = Agent(
    role="高级研究员",
    goal="收集并分析最准确、最相关的信息",
    backstory="""
        你是一位经验丰富的研究员，
        擅长从多个来源收集信息，
        并能够识别关键见解。
        你以严谨和深度著称。
    """,
    verbose=True,
    allow_delegation=False,  # 是否允许委托任务
    tools=[serper_tool, scraper_tool]  # 配备的工具
)

属性	说明
role	角色的名称，如”研究员”、“作家”
goal	角色的目标，应该是具体的、可衡量的
backstory	角色的背景故事，影响行为风格
verbose	是否输出详细日志
allow_delegation	是否允许委托任务给其他 Agent
tools	角色可以使用的工具列表

2. Task（任务）

每个任务定义了具体的工作内容：

from crewai import Task

# 定义研究任务
research_task = Task(
    description="调查 AI 在医疗领域的最新应用",
    agent=researcher,  # 指定执行者
    expected_output="""一份详细的报告，包含：
        1. 主要应用场景
        2. 关键技术
        3. 发展趋势
        4. 参考文献""",
    async_execution=False,  # 是否异步执行
    callback=None  # 完成后的回调函数
)

3. Crew（团队）

Crew 是 Agent 和 Task 的集合：

from crewai import Crew

# 创建团队
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process="sequential",  # 或 "hierarchical"
    verbose=True
)

# 运行项目
result = crew.kickoff()

4. Process（协作流程）

CrewAI 支持两种主要的协作流程：

顺序流程（Sequential）

任务按顺序执行，前一个任务的输出作为下一个任务的输入：

任务1 → 任务2 → 任务3 → 最终结果

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process="sequential"
)

层级流程（Hierarchical）

有一个管理者 Agent 负责协调其他 Agent：

    [Manager]
    /    |    \
[T1]   [T2]   [T3]

from crewai import Agent

manager = Agent(
    role="项目经理",
    goal="确保项目按时高质量完成",
    backstory="经验丰富的项目经理，擅长协调团队",
    allow_delegation=True
)

crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[...],
    process="hierarchical"
)

---

工具系统

1. 内置工具

CrewAI 提供了丰富的内置工具：

from crewai.tools import SerperDevTool, ScrapeWebsiteTool, FileReadTool

# 搜索工具
search_tool = SerperDevTool(
    description="搜索互联网获取最新信息"
)

# 网页抓取工具
scrape_tool = ScrapeWebsiteTool(
    description="抓取特定网页的内容"
)

# 文件读取工具
read_tool = FileReadTool(
    file_path="./data/input.txt"
)

2. 自定义工具

你可以创建自己的工具：

from crewai.tools import BaseTool
from pydantic import Field

class CustomCalculatorTool(BaseTool):
    name: str = "calculator"
    description: str = "执行数学计算"

    def _run(self, expression: str = Field(description="数学表达式")) -> str:
        try:
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"计算错误: {str(e)}"

calculator = CustomCalculatorTool()

# 在 Agent 中使用
agent = Agent(
    role="数学助手",
    tools=[calculator]
)

3. LangChain 工具

CrewAI 与 LangChain 兼容，可以直接使用 LangChain 的工具：

from langchain.tools import DuckDuckGoSearchRun

# 包装成 CrewAI 工具
search = DuckDuckGoSearchRun()

agent = Agent(
    role="搜索专家",
    tools=[search]
)

---

代码示例

1. 基础示例：研究 + 写作

这是一个经典的使用场景：研究员收集信息，作家撰写内容。

from crewai import Agent, Task, Crew, Process
from crewai.tools import SerperDevTool, FileWriteTool

# 1. 定义工具
search_tool = SerperDevTool(description="搜索互联网")
write_tool = FileWriteTool(file_path="./output/article.md")

# 2. 创建 Agent
researcher = Agent(
    role="AI 研究员",
    goal="收集关于目标主题的全面信息",
    backstory="""你是一位专业的 AI 研究员，
    擅长从多个来源收集信息并整理成结构化报告。
    你注重信息的准确性和来源的可信度。""",
    verbose=True,
    tools=[search_tool]
)

writer = Agent(
    role="技术作家",
    goal="将复杂的技术内容转化为易懂的文章",
    backstory="""你是一位经验丰富的技术作家，
    擅长用通俗易懂的语言解释复杂概念。
    你的文章逻辑清晰，例子丰富。""",
    verbose=True,
    tools=[write_tool]
)

# 3. 创建任务
research_task = Task(
    description="调查 Claude Code 的最新功能和应用场景",
    agent=researcher,
    expected_output="""一份详细的研究报告，包含：
        - Claude Code 简介
        - 核心功能
        - 实际应用案例
        - 最佳实践"""
)

writing_task = Task(
    description="基于研究报告撰写一篇博客文章",
    agent=writer,
    expected_output="""一篇结构完整、内容丰富的博客文章，
        字数在 1500-2000 字之间""",
    context=[research_task]  # 使用研究任务的输出
)

# 4. 创建团队并运行
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential
)

result = crew.kickoff()
print(result)

2. 层级流程示例：项目管理

在层级流程中，Manager Agent 负责分配任务：

from crewai import Agent, Task, Crew, Process

# Manager Agent
manager = Agent(
    role="项目经理",
    goal="协调团队，确保任务高效完成",
    backstory="""你是一位经验丰富的项目经理，
    擅长任务分解、进度跟踪和资源协调。""",
    allow_delegation=True
)

# 执行 Agent
analyst = Agent(
    role="数据分析师",
    goal="提供准确的数据分析",
    backstory="你是数据分析专家，擅长使用 SQL 和 Python 进行分析。",
    tools=[sql_tool, python_tool]
)

developer = Agent(
    role="开发工程师",
    goal="实现高质量的代码",
    backstory="你是全栈开发工程师，擅长快速实现功能。",
    tools=[code_tool, git_tool]
)

# 任务
analysis_task = Task(
    description="分析销售数据，识别关键趋势",
    agent=analyst
)

development_task = Task(
    description="根据分析结果开发数据可视化功能",
    agent=developer
)

# 层级团队
crew = Crew(
    agents=[manager, analyst, developer],
    tasks=[analysis_task, development_task],
    process=Process.hierarchical
)

result = crew.kickoff()

3. 异步执行示例

对于可以并行执行的任务：

from crewai import Task

# 这些任务可以并行执行
task1 = Task(description="搜索 AI 新闻", agent=researcher)
task2 = Task(description="搜索 ML 新闻", agent=researcher)
task3 = Task(description="搜索 Web3 新闻", agent=researcher)

# 创建团队（异步执行）
crew = Crew(
    agents=[researcher],
    tasks=[task1, task2, task3],
    process=Process.sequential  # 或者 hierarchical
)

# 运行
results = crew.kickoff()

4. 带条件的任务

from crewai import Task
from crewai.tasks.task_output import TaskOutput

def check_quality(output: TaskOutput) -> bool:
    """检查输出质量"""
    return len(output.raw) > 500  # 至少 500 字

quality_task = Task(
    description="撰写产品介绍",
    agent=writer,
    expected_output="至少 500 字的产品介绍",
    conditions=[check_quality]  # 质量检查
)

5. 完整的配置示例

import os
from crewai import Agent, Task, Crew

# 设置 API Key
os.environ["OPENAI_API_KEY"] = "your-key"
os.environ["SERPER_API_KEY"] = "your-key"

# 1. 创建 Agent
researcher = Agent(
    role="市场研究员",
    goal="收集竞争对手的详细信息",
    backstory="""你拥有 10 年市场研究经验，
    擅长竞争对手分析和趋势预测。
    你总是基于数据做出判断。""",
    verbose=True,
    memory=True,  # 开启记忆
    max_iter=5  # 最大迭代次数
)

# 2. 定义工具
from crewai.tools import SerperDevTool
search = SerperDevTool()

# 3. 创建任务
task = Task(
    description="调查主要竞争对手的产品策略",
    agent=researcher,
    tools=[search],
    expected_output="包含竞品分析、市场定位的详细报告"
)

# 4. 创建团队
crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
    verbose=2  # 详细程度：0-2
)

# 5. 运行
result = crew.kickoff(inputs={"topic": "智能手机市场"})

---

与 MetaGPT 对比

1. 核心差异

方面	MetaGPT	CrewAI
设计理念	模拟软件公司	灵活的团队协作
角色系统	固定角色（SOP）	自定义角色
工作流	严格的流程	可配置流程
适用场景	完整项目开发	多种任务类型
学习曲线	较陡	较平缓

2. 选择建议

选择 MetaGPT 的场景：

• 需要完整的软件工程流程
• 需求明确，需要产出完整代码
• 需要严格的 QA 和测试
• 项目结构清晰，流程标准化

选择 CrewAI 的场景：

• 需要灵活的 Agent 协作
• 研究、数据处理、内容生成
• 快速原型开发
• 非软件开发任务

3. 代码复杂度对比

# MetaGPT：完整的项目开发
company = SoftwareCompany()
company.add_member(ProjectManager())
company.add_member(Architect())
company.add_member(Engineer())
company.add_member(QAEngineer())
await company.run_project("开发一个博客系统")

# CrewAI：灵活的任务协作
crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential
)
crew.kickoff()

4. 产出质量对比

• MetaGPT：产出更完整（有代码、测试、文档），但资源消耗大
• CrewAI：产出更灵活，可能需要人工后续整理

---

实际应用案例

案例 1：内容创作管道

# 自动内容创作工作流
researcher = Agent(role="研究员", tools=[search_tool])
writer = Agent(role="作家", tools=[write_tool])
seo_specialist = Agent(role="SEO 专家", tools=[])

research_task = Task(description="研究主题", agent=researcher)
writing_task = Task(description="撰写初稿", agent=writer, context=[research_task])
seo_task = Task(description="优化 SEO", agent=seo_specialist, context=[writing_task])

crew = Crew(
    agents=[researcher, writer, seo_specialist],
    tasks=[research_task, writing_task, seo_task],
    process=Process.sequential
)

案例 2：数据分析管道

# 数据分析自动化
collector = Agent(role="数据收集员", tools=[sql_tool, api_tool])
cleaner = Agent(role="数据清洗员", tools=[python_tool])
analyst = Agent(role="数据分析师", tools=[python_tool])
visualizer = Agent(role="可视化专家", tools=[chart_tool])

# ... 定义任务 ...
crew = Crew(agents=[...], tasks=[...], process=Process.sequential)

案例 3：客服系统

# 智能客服
classifier = Agent(role="问题分类员", tools=[classifier_tool])
technical_support = Agent(role="技术支持", tools=[kb_tool, search_tool])
billing_support = Agent(role="Billing 支持", tools=[billing_tool])
human_handoff = Agent(role="人工转接", tools=[ticket_tool])

# 分类后分配给不同的专家
crew = Crew(agents=[...], tasks=[...], process=Process.hierarchical)

---

部署选项

1. 本地部署

# 安装
pip install crewai

# 或安装最新版本
pip install git+https://github.com/crewAIInc/crewAI.git

2. Docker 部署

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
CMD ["python", "main.py"]

# 构建和运行
docker build -t crewai-app .
docker run -e OPENAI_API_KEY=your-key crewai-app

3. 云端部署

CrewAI 支持部署到多种云平台：

# Vercel (通过 Flask/FastAPI)
from flask import Flask
app = Flask(__name__)

@app.route("/run", methods=["POST"])
def run_crew():
    data = request.json
    result = crew.kickoff(inputs=data)
    return {"result": result}

4. API 服务化

from crewai import Crew
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/api/crew", methods=["POST"])
def run_crew():
    inputs = request.json
    result = crew.kickoff(inputs=inputs)
    return jsonify({"result": result})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

---

最佳实践

1. Agent 设计原则

# 好的 Agent 定义
researcher = Agent(
    role="AI 研究员",
    goal="收集并整理 AI 领域的最新研究成果",  # 具体、可衡量
    backstory="""你是斯坦福大学 AI 实验室的研究员，
    专注于机器学习领域。""",  # 有背景故事
    verbose=True,
    tools=[search, scraper]
)

# 不好的 Agent 定义
agent = Agent(
    role="助手",  # 太笼统
    goal="帮助用户"  # 太模糊
)

2. 任务设计原则

# 好的任务定义
task = Task(
    description="""调查 Claude Code 的以下方面：
    1. 核心功能
    2. 实际应用案例
    3. 与竞品对比""",  # 具体、可执行
    agent=researcher,
    expected_output="""包含以上三点的详细报告，
    每个部分至少 300 字""",  # 明确的输出期望
    context=[previous_task]  # 适当的上下文
)

3. 流程选择

场景	推荐流程
线性依赖（研究→写作→编辑）	Sequential
需要协调多个子任务	Hierarchical
独立可并行执行	Async

4. 调试技巧

# 开启详细日志
crew = Crew(
    agents=[...],
    tasks=[...],
    verbose=2  # 0=quiet, 1=normal, 2=detailed
)

# 单步执行查看中间结果
for task in tasks:
    result = task.execute()
    print(f"Task {task.description}: {result}")

---

常见问题

Q: CrewAI 和 LangChain Agents 有什么区别？

A: CrewAI 建立在 LangChain 之上，专注于多 Agent 协作。LangChain Agents 更底层，适合单 Agent 场景。

Q: 如何选择顺序流程还是层级流程？

A: 任务有明确依赖关系时用顺序流程；需要协调多个子任务时用层级流程。

Q: Agent 的 memory 是什么？

A: memory 让 Agent 能够记住之前的对话和上下文，适合长对话场景。

Q: 如何处理 API 错误？

from crewai.utilities import RetryPolicy

crew = Crew(
    agents=[...],
    tasks=[...],
    retry_policy=RetryPolicy(
        max_attempts=3,
        delay=1
    )
)

---

总结

CrewAI 是一个强大而灵活的多智能体协作框架，它的核心特点是：

• 简洁易用：API 设计直观，学习曲线平缓
• 灵活协作：支持多种工作流模式
• 工具丰富：内置工具 + LangChain 兼容
• 可扩展：易于添加自定义 Agent 和工具

与 MetaGPT 相比，CrewAI 更适合那些不需要完整软件工程流程的场景，比如研究分析、内容创作、数据处理等。如果你需要快速构建一个 AI 协作系统，CrewAI 是一个很好的选择。

当然，CrewAI 也有它的局限性：对于需要完整代码、测试、文档的软件项目，MetaGPT 可能更合适。选择哪个框架，取决于你的具体需求。

“在 AI 时代，单打独斗已经不够了。CrewAI 教会我们，真正的 AI 力量来自于协作——让每个智能体发挥所长，共同完成超越个体极限的任务。“

---

参考链接

• GitHub: CrewAI
• 官方文档
• CrewAI 示例