CrewAI 快速上手：构建多 Agent 协作系统

来源：docs.crewai.com | github.com/crewAIInc/crewAI | 整理时间：2026-04-06

什么是 CrewAI？

CrewAI 是一个多 Agent 协作框架，让你用 Python 快速编排多个 AI Agent 共同完成复杂任务。核心思想：像管理团队一样管理 Agent——定义角色、分配任务、设定流程。

GitHub Stars：30K+（截至 2026-04） 最新版本：v1.13.0（2026-04-02） 许可证：MIT

与其他框架的定位差异

核心概念

四大组件

Agent 属性

from crewai import Agent

researcher = Agent(
    role="市场研究员",
    goal="深入研究指定市场的趋势和机会",
    backstory="你是一位有10年经验的市场分析专家，擅长发现隐藏的商业机会。",
    tools=[search_tool, web_scraper],
    llm="gpt-4o",               # 指定模型
    verbose=True,
    allow_delegation=False,      # 是否可以委派任务
    max_iter=15,                 # 最大迭代次数
    memory=True,                 # 启用记忆
)

Task 属性

from crewai import Task

research_task = Task(
    description="研究 {topic} 领域的最新趋势，分析前5个竞争者",
    expected_output="一份包含市场趋势、竞争分析和机会评估的报告",
    agent=researcher,
    output_file="output/research.md",  # 输出文件
)

快速上手

安装

pip install crewai crewai-tools

最小示例

from crewai import Agent, Task, Crew, Process

# 1. 定义 Agent
researcher = Agent(
    role="研究员",
    goal="收集和分析信息",
    backstory="你是一位严谨的研究员，擅长收集和分析数据。",
    verbose=True,
)

writer = Agent(
    role="技术作者",
    goal="将研究结果写成清晰的文章",
    backstory="你是一位技术写作专家，能把复杂概念变得易懂。",
    verbose=True,
)

# 2. 定义 Task
research_task = Task(
    description="研究 {topic} 的最新进展",
    expected_output="包含关键发现的研究摘要",
    agent=researcher,
)

write_task = Task(
    description="基于研究结果撰写一篇文章",
    expected_output="一篇结构清晰的中文技术文章",
    agent=writer,
)

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

# 4. 启动
result = crew.kickoff(inputs={"topic": "MCP 协议"})
print(result)

使用 CLI 脚手架

# 创建项目
crewai create crew my-crew
cd my-crew

# 项目结构
# my-crew/
# ├── src/my_crew/
# │   ├── crew.py         # Crew 定义
# │   ├── main.py         # 入口文件
# │   └── tools/          # 自定义工具
# ├── knowledge/           # 知识库文件
# └── pyproject.toml

工具系统

内置工具（30+）

自定义工具

方式一：装饰器

from crewai.tools import tool

@tool("查询库存")
def check_inventory(product_name: str) -> str:
    """查询指定产品的库存数量"""
    stock = query_database(product_name)
    return f"{product_name} 当前库存: {stock} 件"

方式二：类继承

from crewai.tools import BaseTool
from pydantic import BaseModel

class InventoryInput(BaseModel):
    product_name: str

class InventoryTool(BaseTool):
    name: str = "查询库存"
    description: str = "查询指定产品的库存数量"

    def _run(self, product_name: str) -> str:
        stock = query_database(product_name)
        return f"{product_name} 当前库存: {stock} 件"

方式三：集成 MCP Server

from crewai.tools import MCPToolAdapter

# 直接连接 MCP Server 使用其工具
mcp_tool = MCPToolAdapter(
    server_url="http://localhost:3000",
    tool_name="query_database",
)

记忆与知识系统

六大记忆组件

知识源

支持多种文件格式作为知识库：

from crewai.knowledge.source import (
    PDFKnowledgeSource,
    CSVKnowledgeSource,
    JSONKnowledgeSource,
    StringKnowledgeSource,
)

pdf_source = PDFKnowledgeSource(file_paths=["docs/report.pdf"])
csv_source = CSVKnowledgeSource(file_paths=["data/sales.csv"])

crew = Crew(
    agents=[analyst],
    tasks=[analyze_task],
    knowledge_sources=[pdf_source, csv_source],
)

嵌入提供商

支持 10+ 套入模型：OpenAI（默认）、Google、Ollama、FastEmbed、Cohere、Voyage AI 等。

Flow 编排

Flow 用于编排多阶段工作流（Crew 级别的协调）：

from crewai.flow.flow import Flow, listen, start, router

class ResearchFlow(Flow):
    @start()
    def initial_research(self):
        result = research_crew.kickoff(inputs={"topic": self.state.topic})
        return result

    @listen(initial_research)
    def analyze_findings(self, research_result):
        result = analysis_crew.kickoff(inputs={"research": research_result})
        return result

    @router(analyze_findings)
    def route_by_sentiment(self, analysis):
        if analysis.positive:
            return "write_report"
        return "deep_dive"

    @listen("write_report")
    def write_report(self, analysis):
        return report_crew.kickoff(inputs={"analysis": analysis})

    @listen("deep_dive")
    def deep_dive(self, analysis):
        return research_crew.kickoff(inputs={"topic": analysis.gaps})

flow = ResearchFlow()
result = flow.kickoff(inputs={"topic": "AI Agent 市场"})

流程模式

Sequential（顺序）

Crew(process=Process.sequential)
# Task 1 → Task 2 → Task 3（按顺序执行）

Hierarchical（层级）

Crew(
    process=Process.hierarchical,
    manager_llm="gpt-4o",   # Manager Agent 自动创建
)
# Manager 分配任务给 Agent，Agent 可互相委派

框架选型指南

CrewAI vs LangGraph

什么时候选 CrewAI？

• 需要快速搭建多 Agent 原型
• Agent 角色分工明确（研究员、写手、审核员…）
• 不需要精确控制每一步的执行流程
• 团队更熟悉 Python 且偏好声明式 API

什么时候选 LangGraph？

• 工作流有复杂的条件分支和循环
• 需要精确的状态管理和检查点
• 需要人工审批节点（breakpoint）
• 对执行过程有严格的可观测性要求

相关链接

• CrewAI 官方文档：https://docs.crewai.com
• GitHub 仓库：https://github.com/crewAIInc/crewAI
• CrewAI Enterprise：https://crewai.com
• LangGraph 实战
• Agent 框架全景
• OpenManus 实战