CrewAI 教程 2026：构建协同工作的多智能体系统

复杂任务通常需要专业分工与协作。一个研究项目需要研究员、分析师和写作者；一个软件项目需要架构师、编码员和评审员。CrewAI 让您能够构建专业 AI 智能体团队，像人类团队一样协作。

CrewAI 的独特之处

CrewAI 超越了单一智能体框架，提供：

基于角色的智能体：每个智能体拥有专业知识、目标和背景故事

任务委派：任务自动路由到合适的智能体

顺序/并行执行：控制工作流依赖关系

记忆共享：智能体基于彼此的输出构建

内置工具：网络搜索、文件 I/O、代码执行

安装

bash
pip install crewai crewai-tools

核心概念

智能体

每个智能体具有角色、目标、背景故事和可选工具：

python
from crewai import Agent
from crewai_tools import SerperDevTool, FileReadTool
研究智能体
researcher = Agent(
    role='高级研究分析师',
    goal='查找关于主题的准确、最新信息并综合见解',
    backstory="""你是一位拥有 15 年跨行业信息查找与验证经验的研究分析师。
    以严谨、引用来源、区分事实与推测而闻名。""",
    tools=[SerperDevTool()],  # 网络搜索能力
    llm='gpt-4o',
    verbose=True,
    memory=True
)
写作智能体
writer = Agent(
    role='专业内容写作者',
    goal='将研究转化为引人入胜、结构良好的内容',
    backstory="""你是一位专注于科技和商业领域的专业写作者。
    擅长将复杂主题变得易于理解，创作既能提供信息又能说服
    成熟受众的内容。""",
    llm='claude-3-5-sonnet',
    verbose=True
)
评审智能体
critic = Agent(
    role='内容质量评审员',
    goal='确保所有内容的准确性、质量和完整性',
    backstory="""你拥有严谨的编辑背景。你能在发布前发现事实错误、
    逻辑漏洞、无依据的主张和质量问题。""",
    llm='gpt-4o',
    verbose=True
)

任务

python
from crewai import Task
研究任务
research_task = Task(
    description="""研究 [主题] 在 2026 年的当前状况。
    
    查找：
    1. 最新进展和新闻（过去 3 个月）
    2. 关键参与者及其立场
    3. 统计数据与市场规模
    4. 来自 3 个不同视角的专家意见
    5. 潜在挑战与机遇
    
    引用所有来源并附上 URL。""",
    agent=researcher,
    expected_output="一份带有引用的综合研究报告，800-1000 字"
)
写作任务（依赖研究）
writing_task = Task(
    description="""使用提供的研究成果，撰写一篇关于 [主题] 的文章。
    
    要求：
    - 目标受众：[描述受众]
    - 语气：专业但易懂
    - 长度：1500-2000 字
    - 结构：引言、4 个主要部分、结论
    - 包含：具体示例、数据点、可操作见解
    - SEO 关键词：[目标关键词]""",
    agent=writer,
    expected_output="一篇完整的、可发布的文章",
    context=[research_task]  # 接收研究输出作为上下文
)
评审任务
review_task = Task(
    description="""评审文章并提供反馈：
    
    检查：
    1. 事实准确性（对照研究验证）
    2. 逻辑流程与结构
    3. 无依据的主张
    4. 语法与清晰度问题
    5. SEO 优化
    
    输出：改进后的文章及变更日志。""",
    agent=critic,
    expected_output="最终润色后的文章及变更日志",
    context=[research_task, writing_task]
)

团队

python
from crewai import Crew, Process
组建团队
article_crew = Crew(
    agents=[researcher, writer, critic],
    tasks=[research_task, writing_task, review_task],
    process=Process.sequential,  # 每个任务等待前一个完成
    verbose=True,
    memory=True  # 智能体间共享记忆
)
运行团队
result = article_crew.kickoff(inputs={
    'topic': 'AI 驱动的客户服务自动化',
    'audience': '中端市场 SaaS 创始人',
    'keyword': 'AI 客户服务 2026'
})
print(result.raw)  # 最终输出

真实案例：竞争情报团队

python
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
web_search = SerperDevTool()
web_scraper = ScrapeWebsiteTool()
竞争情报专用智能体
data_collector = Agent(
    role='竞争情报数据收集员',
    goal='收集关于竞争对手公司的全面数据',
    backstory='擅长查找公司公开信息：融资、招聘、产品变更、公关。',
    tools=[web_search, web_scraper],
    llm='gpt-4o'
)
strategic_analyst = Agent(
    role='战略业务分析师',
    goal='分析竞争格局并识别战略影响',
    backstory='前麦肯锡顾问，专精于科技市场分析。',
    llm='claude-3-5-sonnet'
)
report_writer = Agent(
    role='高管报告写作者',
    goal='创建简洁、可执行的高管报告',
    backstory='专精于撰写 C 级高管可用的、推动决策的报告。',
    llm='gpt-4o'
)
任务
collect_task = Task(
    description="""针对每个竞争对手：{competitors}
    收集：
    - 近期产品发布或更新
    - 定价变化
    - 新合作伙伴或集成
    - 招聘信号（战略领域的职位发布）
    - 营销定位变化
    - 融资新闻
    时间范围：过去 90 天""",
    agent=data_collector,
    expected_output="每个竞争对手的结构化数据及来源"
)
analyze_task = Task(
    description="""分析收集到的竞争数据。
    我们公司背景：{our_company_context}
    
    提供：
    1. 竞争威胁（短期 vs 长期）
    2. 差异化机会
    3. 市场定位空白
    4. 建议的战略应对""",
    agent=strategic_analyst,
    context=[collect_task]
)
report_task = Task(
    description="创建一页纸的高管竞争情报报告。包括：执行摘要、关键发现表格、前 3 项行动项和展望。",
    agent=report_writer,
    context=[collect_task, analyze_task],
    output_file="competitive_report.md"  # 自动保存输出
)
运行
comp_intel_crew = Crew(
    agents=[data_collector, strategic_analyst, report_writer],
    tasks=[collect_task, analyze_task, report_task],
    process=Process.sequential
)
result = comp_intel_crew.kickoff(inputs={
    'competitors': ['公司 A', '公司 B', '公司 C'],
    'our_company_context': '面向中端市场零售公司的 B2B SaaS'
})

并行团队执行

python
层级流程：管理者委派任务给智能体
manager = Agent(
    role='项目经理',
    goal='协调团队高效完成项目',
    llm='gpt-4o',
    allow_delegation=True  # 可以将任务分配给其他智能体
)
parallel_crew = Crew(
    agents=[manager, researcher, writer, seo_analyst],
    tasks=[main_task],  # 管理者分解任务并委派
    process=Process.hierarchical,
    manager_llm='gpt-4o'
)

添加自定义工具

python
from crewai_tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field
class DatabaseQueryInput(BaseModel):
    query: str = Field(description="要执行的 SQL 查询")
class DatabaseQueryTool(BaseTool):
    name: str = "数据库查询工具"
    description: str = "对公司数据库执行 SQL 查询"
    args_schema: Type[BaseModel] = DatabaseQueryInput
    
    def _run(self, query: str) -> str:
        # 连接到实际数据库
        conn = get_database_connection()
        result = conn.execute(query).fetchall()
        return str(result)
赋予数据分析智能体
data_analyst = Agent(
    role='数据分析师',
    tools=[DatabaseQueryTool()],
    ...
)

衡量团队性能

python
import time
start = time.time()
result = crew.kickoff(inputs=inputs)
elapsed = time.time() - start
print(f"执行时间：{elapsed:.0f} 秒")
print(f"Token 使用量：{result.token_usage}")
print(f"完成任务数：{len(result.tasks_output)}")

何时使用多智能体 vs 单智能体

使用多智能体的情况使用单智能体的情况

任务需要不同专业知识简单、专注的任务 质量评审至关重要速度更重要 存在并行工作流单一领域 研究 + 综合 + 写作一步转换

结论

CrewAI 擅长人类通常以团队形式完成的知识工作。上述竞争情报示例已可用于生产——公司每周运行它以跟踪竞争对手动态。从 2-3 个智能体在明确定义的任务上开始，再逐步增加复杂性。