知识模块
🤖 Agent 知识模块
十、多 Agent 协作
CrewAI 框架

CrewAI 框架

CrewAI 是一个基于角色扮演的多智能体协作框架,通过定义明确的角色、任务和协作流程,让多个 Agent 像团队一样协作完成任务。它强调"角色化设计"和"流程编排",非常适合内容创作、研究任务和自动化工作流。


一、核心原理

1.1 CrewAI 设计哲学

CrewAI 的核心思想是将 Agent 视为"团队成员",每个成员有明确的角色和职责:

┌─────────────────────────────────────────────────────────────┐
│                    CrewAI 设计哲学                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   传统 Agent:                                              │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                   单一 Agent                        │  │
│   │                                                    │  │
│   │   输入 ──→ 处理 ──→ 输出                           │  │
│   │                                                    │  │
│   │   问题:能力有限,难以处理复杂任务                  │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│   CrewAI 模式:                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    Crew (团队)                      │  │
│   │                                                    │  │
│   │   ┌─────────┐  ┌─────────┐  ┌─────────┐          │  │
│   │   │ Agent 1 │  │ Agent 2 │  │ Agent 3 │          │  │
│   │   │ (角色A) │  │ (角色B) │  │ (角色C) │          │  │
│   │   └────┬────┘  └────┬────┘  └────┬────┘          │  │
│   │        │            │            │                 │  │
│   │        └────────────┼────────────┘                 │  │
│   │                     ↓                              │  │
│   │              ┌─────────────┐                       │  │
│   │              │   Process   │ ← 流程编排            │  │
│   │              │ (协作流程)  │                       │  │
│   │              └─────────────┘                       │  │
│   │                                                    │  │
│   │   优势:角色明确,流程可控,易于调试               │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 核心概念

概念英文描述示例
AgentAgent具有特定角色和能力的智能体研究员、撰稿人、编辑
TaskTask需要完成的具体任务"研究AI发展趋势"
CrewCrewAgent 和 Task 的组织单位一个内容创作团队
ProcessProcess任务执行流程Sequential、Hierarchical
ToolToolAgent 可使用的工具搜索、数据库查询

1.3 四层架构

┌─────────────────────────────────────────────────────────────┐
│                    CrewAI 四层架构                           │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   Layer 4: Crew Layer (团队层)                              │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │    Crew = Agents + Tasks + Process                  │  │
│   │                                                      │  │
│   │    • 组织 Agent 和 Task                            │  │
│   │    • 定义协作流程                                   │  │
│   │    • 管理执行上下文                                 │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                           ↑                                 │
│   Layer 3: Process Layer (流程层)                          │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │    Sequential / Hierarchical                        │  │
│   │                                                      │  │
│   │    • 定义任务执行顺序                               │  │
│   │    • 协调 Agent 间协作                              │  │
│   │    • 管理任务状态                                   │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                           ↑                                 │
│   Layer 2: Task Layer (任务层)                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │    Task = Description + Agent + Output             │  │
│   │                                                      │  │
│   │    • 定义任务目标                                   │  │
│   │    • 指定负责 Agent                                 │  │
│   │    • 管理任务输出                                   │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                           ↑                                 │
│   Layer 1: Agent Layer (智能体层)                          │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │    Agent = Role + Goal + Tools + Backstory         │  │
│   │                                                      │  │
│   │    • 定义角色身份                                   │  │
│   │    • 设定目标方向                                   │  │
│   │    • 配置能力工具                                   │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

二、核心组件详解

2.1 Agent 组件

Agent 是 CrewAI 的核心执行单元:

from crewai import Agent
 
# 创建 Agent
researcher = Agent(
    # 基础属性
    role="高级研究员",                    # 角色名称
    goal="发现前沿技术趋势和洞察",        # 目标
    backstory="""                        # 背景故事(塑造角色性格)
        你是一位经验丰富的研究员,在科技行业工作了15年。
        你擅长从海量信息中提取关键洞察,你的分析报告被业界广泛认可。
        你注重事实和数据,总是给出可靠的结论。
    """,
    
    # 能力配置
    verbose=True,                        # 详细日志
    allow_delegation=False,              # 是否允许委派任务
    llm="gpt-4",                         # 使用的 LLM
    
    # 工具配置
    tools=[search_tool, scrape_tool],    # 可用工具
    
    # 行为约束
    max_iter=15,                         # 最大迭代次数
    max_rpm=100,                         # 每分钟最大请求数
    memory=True,                         # 是否启用记忆
)

Agent 属性详解

属性类型说明示例
rolestr角色名称,定义身份"数据分析师"
goalstr目标,定义方向"从数据中发现价值"
backstorystr背景故事,塑造性格"10年经验的分析师..."
toolslist可用工具列表[search_tool]
llmstr使用的语言模型"gpt-4"
verbosebool是否输出详细日志True
allow_delegationbool是否允许委派False
max_iterint最大迭代次数15
memorybool是否启用记忆True

2.2 Task 组件

Task 定义了需要完成的工作:

from crewai import Task
 
# 创建任务
research_task = Task(
    description="""                      # 任务描述
        研究 {topic} 领域的最新发展趋势。
        需要关注:
        1. 主要技术突破
        2. 市场动态和投资情况
        3. 未来3年的发展预测
        
        输出一份详细的研究报告。
    """,
    expected_output="""                  # 期望输出格式
        一份结构化的研究报告,包含:
        - 摘要
        - 主要发现(3-5条)
        - 详细分析
        - 参考资料
    """,
    agent=researcher,                    # 负责的 Agent
    tools=[search_tool],                 # 任务专用工具
    
    # 输出配置
    output_file="report.md",             # 输出文件
    async_execution=False,               # 是否异步执行
    
    # 上下文
    context=[other_task],                # 依赖的其他任务
)

2.3 Crew 组件

Crew 是 Agent 和 Task 的组织单位:

from crewai import Crew, Process
 
# 创建 Crew
crew = Crew(
    agents=[researcher, writer, editor],  # Agent 列表
    tasks=[research_task, write_task, edit_task],  # Task 列表
    
    # 流程配置
    process=Process.sequential,           # 执行流程
    
    # 运行配置
    verbose=True,                         # 详细日志
    memory=True,                          # 启用记忆
    cache=True,                           # 启用缓存
    
    # 输出配置
    output_log_file="crew_log.txt",       # 日志文件
    share_crew_output=True,               # 共享输出
)

2.4 Process 组件

Process 定义了任务的执行方式:

┌─────────────────────────────────────────────────────────────┐
│                    Process 执行模式                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. Sequential Process(顺序执行)                          │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │   Task 1 ──→ Task 2 ──→ Task 3 ──→ Output          │  │
│   │   (Agent A)  (Agent B)  (Agent C)                   │  │
│   │                                                      │  │
│   │   特点:                                            │  │
│   │   • 任务按顺序依次执行                              │  │
│   │   • 前一任务输出作为后一任务输入                    │  │
│   │   • 适合有依赖关系的流水线任务                      │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│  2. Hierarchical Process(层次执行)                        │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                                                      │  │
│   │              ┌─────────────┐                        │  │
│   │              │   Manager   │ ← 自动创建的管理Agent  │  │
│   │              │   Agent     │                        │  │
│   │              └──────┬──────┘                        │  │
│   │                     │                               │  │
│   │         ┌───────────┼───────────┐                  │  │
│   │         ↓           ↓           ↓                  │  │
│   │   ┌──────────┐ ┌──────────┐ ┌──────────┐          │  │
│   │   │ Worker   │ │ Worker   │ │ Worker   │          │  │
│   │   │ Agent A  │ │ Agent B  │ │ Agent C  │          │  │
│   │   └──────────┘ └──────────┘ └──────────┘          │  │
│   │                                                      │  │
│   │   特点:                                            │  │
│   │   • 自动创建 Manager Agent 进行协调                │  │
│   │   • Manager 负责任务分配和结果汇总                 │  │
│   │   • 适合复杂任务自动分解                           │  │
│   │                                                      │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

三、代码实现

3.1 基础示例:内容创作团队

"""
CrewAI 基础示例:内容创作团队
展示顺序协作模式的完整实现
"""
 
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
 
# 初始化工具
search_tool = SerperDevTool()      # 搜索工具
file_read_tool = FileReadTool()    # 文件读取工具
 
# ==================== 定义 Agent ====================
 
# Agent 1: 研究员
researcher = Agent(
    role="高级研究员",
    goal="发现前沿技术趋势和洞察",
    backstory="""
        你是一位经验丰富的科技研究员,在硅谷工作了15年。
        你擅长从海量信息中提取关键洞察,你的分析报告被业界广泛认可。
        你注重事实和数据,总是给出可靠的结论。
    """,
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm="gpt-4"
)
 
# Agent 2: 撰稿人
writer = Agent(
    role="技术撰稿人",
    goal="创作引人入胜的技术文章",
    backstory="""
        你是一位技术作家,曾为多家知名科技媒体撰稿。
        你擅长将复杂的技术概念转化为通俗易懂的文章。
        你的文章风格清晰、有趣,深受读者喜爱。
    """,
    verbose=True,
    allow_delegation=False,
    tools=[file_read_tool],
    llm="gpt-4"
)
 
# Agent 3: 编辑
editor = Agent(
    role="资深编辑",
    goal="确保文章质量和一致性",
    backstory="""
        你是一位资深编辑,有10年的编辑经验。
        你注重文章的结构、逻辑和表达,追求完美。
        你会仔细检查每一篇文章,确保其符合出版标准。
    """,
    verbose=True,
    allow_delegation=True,  # 允许委派给研究员或撰稿人
    llm="gpt-4"
)
 
# ==================== 定义 Task ====================
 
# Task 1: 研究
research_task = Task(
    description="""
        研究 {topic} 领域的最新发展趋势。
        
        需要关注:
        1. 主要技术突破和创新
        2. 市场动态和投资情况
        3. 主要参与者和竞争格局
        4. 未来3年的发展预测
        
        提供详细的研究报告,包含数据和事实支撑。
    """,
    expected_output="""
        一份结构化的研究报告,包含:
        - 执行摘要(200字以内)
        - 主要发现(5-7条)
        - 详细分析(每个发现的数据支撑)
        - 参考资料列表
    """,
    agent=researcher,
    tools=[search_tool]
)
 
# Task 2: 写作
writing_task = Task(
    description="""
        根据研究报告,撰写一篇关于 {topic} 的深度文章。
        
        要求:
        1. 标题吸引人
        2. 开篇引人入胜
        3. 内容结构清晰
        4. 技术深度适中
        5. 结尾有洞察
        
        文章长度:2000-3000字
    """,
    expected_output="""
        一篇完整的技术文章,包含:
        - 标题
        - 导语
        - 正文(分章节)
        - 总结
    """,
    agent=writer,
    context=[research_task]  # 依赖研究任务
)
 
# Task 3: 编辑
editing_task = Task(
    description="""
        审校和优化文章,确保质量符合出版标准。
        
        检查要点:
        1. 逻辑连贯性
        2. 表达准确性
        3. 错别字和语法错误
        4. 格式规范
        
        如发现问题,可以委派给撰稿人修改。
    """,
    expected_output="""
        最终版文章,包含:
        - 修改说明
        - 最终文章内容
        - 质量评分(1-10)
    """,
    agent=editor,
    context=[writing_task]  # 依赖写作任务
)
 
# ==================== 创建 Crew ====================
 
content_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True,
    memory=True  # 启用记忆,共享上下文
)
 
# ==================== 执行 ====================
 
def run_content_pipeline(topic: str):
    """运行内容创作流水线"""
    result = content_crew.kickoff(inputs={"topic": topic})
    
    print("\n" + "="*50)
    print("最终文章:")
    print("="*50)
    print(result)
    
    return result
 
if __name__ == "__main__":
    run_content_pipeline("AI Agent 技术发展趋势")

3.2 进阶示例:层次协作模式

"""
CrewAI 进阶示例:层次协作模式
展示 Manager Agent 自动协调任务
"""
 
from crewai import Agent, Task, Crew, Process
 
# 定义工作 Agent
agents = [
    Agent(
        role="数据分析师",
        goal="分析数据并提供洞察",
        backstory="专业的数据分析师,擅长Python和数据可视化",
        tools=[python_repl_tool],
    ),
    Agent(
        role="报告撰写员",
        goal="撰写专业的分析报告",
        backstory="资深分析师,擅长撰写清晰的报告",
    ),
    Agent(
        role="质量控制员",
        goal="确保输出质量符合标准",
        backstory="有严格标准的质量专家",
    ),
]
 
# 定义任务(不指定 Agent,由 Manager 分配)
tasks = [
    Task(
        description="分析销售数据,找出增长最快的产品类别",
        expected_output="包含数据洞察的分析报告",
    ),
    Task(
        description="根据分析结果撰写商业报告",
        expected_output="结构化的商业报告文档",
    ),
    Task(
        description="审核报告质量,确保无错误",
        expected_output="质量审核报告和最终版本",
    ),
]
 
# 创建层次协作 Crew
hierarchical_crew = Crew(
    agents=agents,
    tasks=tasks,
    process=Process.hierarchical,  # 层次执行
    manager_llm="gpt-4",           # Manager 使用的 LLM
    verbose=True,
)
 
# 执行
result = hierarchical_crew.kickoff()

3.3 工具集成

"""
CrewAI 工具集成示例
展示如何创建自定义工具
"""
 
from crewai_tools import tool
from typing import Optional
 
# 使用装饰器定义工具
@tool("Database Query Tool")
def query_database(query: str, database: Optional[str] = None) -> str:
    """
    执行数据库查询
    
    Args:
        query: SQL 查询语句
        database: 数据库名称(可选)
    
    Returns:
        查询结果
    """
    # 实际实现中这里会连接真实数据库
    mock_result = {
        "sales": [
            {"product": "A", "revenue": 10000},
            {"product": "B", "revenue": 15000},
        ]
    }
    return str(mock_result)
 
@tool("Code Execution Tool")
def execute_code(code: str, language: str = "python") -> str:
    """
    执行代码片段
    
    Args:
        code: 要执行的代码
        language: 编程语言
    
    Returns:
        执行结果
    """
    try:
        # 实际实现中这里会使用安全的代码执行环境
        exec_globals = {}
        exec(code, exec_globals)
        return "代码执行成功"
    except Exception as e:
        return f"执行错误: {str(e)}"
 
# 在 Agent 中使用工具
analyst = Agent(
    role="数据分析师",
    goal="分析数据并提供洞察",
    backstory="专业的数据分析师",
    tools=[query_database, execute_code],  # 使用自定义工具
)

四、适用场景

4.1 最佳适用场景

场景描述推荐配置
内容创作研究、写作、编辑流水线Sequential + 3+ Agent
市场研究数据收集、分析、报告生成Hierarchical
自动化工作流多步骤业务流程Sequential
代码审查代码分析、改进建议、验证Sequential

4.2 场景实现示例

┌─────────────────────────────────────────────────────────────┐
│                    典型应用场景                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  场景1: 新闻内容生产                                        │
│   ┌─────────┐    ┌─────────┐    ┌─────────┐               │
│   │ 选题    │ ─→ │ 采访    │ ─→ │ 撰写    │               │
│   │ Agent   │    │ Agent   │    │ Agent   │               │
│   └─────────┘    └─────────┘    └────┬────┘               │
│                                       │                     │
│                                       ↓                     │
│                               ┌─────────────┐              │
│                               │ 编辑发布    │              │
│                               │ Agent       │              │
│                               └─────────────┘              │
│                                                             │
│  场景2: 数据分析报告                                        │
│   ┌──────────────────────────────────────────────────┐    │
│   │                  Manager Agent                    │    │
│   │               (自动任务分配)                      │    │
│   └─────────────────────┬────────────────────────────┘    │
│              ┌──────────┼──────────┐                      │
│              ↓          ↓          ↓                      │
│        ┌────────┐ ┌────────┐ ┌────────┐                  │
│        │数据收集│ │数据分析│ │报告撰写│                  │
│        │Agent   │ │Agent   │ │Agent   │                  │
│        └────────┘ └────────┘ └────────┘                  │
│                                                             │
│  场景3: 代码审查流水线                                      │
│   ┌─────────┐    ┌─────────┐    ┌─────────┐               │
│   │ 代码    │ ─→ │ 安全    │ ─→ │ 性能    │               │
│   │ 审查    │    │ 检查    │    │ 分析    │               │
│   │ Agent   │    │ Agent   │    │ Agent   │               │
│   └─────────┘    └─────────┘    └─────────┘               │
│         │              │              │                    │
│         └──────────────┼──────────────┘                    │
│                        ↓                                    │
│               ┌─────────────┐                              │
│               │ 综合报告    │                              │
│               │ Agent       │                              │
│               └─────────────┘                              │
│                                                             │
└─────────────────────────────────────────────────────────────┘

4.3 不适用场景

场景原因推荐替代
实时交互响应延迟较高单 Agent
高度灵活对话流程相对固定AutoGen
大规模部署资源消耗较大自定义架构

五、优缺点分析

5.1 优点

优点描述
角色清晰Agent 角色定义明确,职责清晰
流程可控Process 机制确保任务执行有序
易于理解概念直观,学习曲线平缓
工具丰富内置多种工具,支持自定义
记忆系统支持短期和长期记忆

5.2 缺点

缺点描述缓解方案
灵活性有限流程定义相对固定使用 Hierarchical 模式
资源消耗多 Agent 并行内存占用大优化 Agent 数量
通信开销Agent 间通信有延迟减少不必要的协作

六、面试问答

Q1: CrewAI 的核心概念是什么?

回答: CrewAI 的核心概念包括:

  • Agent:具有特定角色和能力的智能体
  • Task:需要完成的具体任务
  • Crew:Agent 和 Task 的组织单位
  • Process:任务执行流程(Sequential 或 Hierarchical)

Q2: Sequential 和 Hierarchical Process 有什么区别?

回答

对比项SequentialHierarchical
执行方式按顺序执行Manager 自动分配
任务依赖明确指定自动识别
适用场景流水线任务复杂任务分解
灵活性较低较高

Q3: 如何选择使用 CrewAI 还是 AutoGen?

回答

  • 选择 CrewAI:角色明确、流程固定、需要严格控制
  • 选择 AutoGen:需要灵活对话、代码执行、研究探索

Q4: CrewAI 如何处理 Agent 间的协作?

回答

  1. 上下文传递:通过 context 参数传递任务依赖
  2. 记忆系统:启用 memory=True 共享上下文
  3. 任务委派:启用 allow_delegation=True 允许委派
  4. 输出传递:前一任务输出自动传递给后一任务

七、小结

概念一句话总结面试关键词
CrewAI基于角色扮演的多智能体协作框架Agent、Task、Crew、Process
Sequential任务按顺序依次执行流水线、上下文传递
HierarchicalManager Agent 自动协调自动分配、灵活调度
适用场景内容创作、工作流自动化角色明确、流程固定

一句话总结:CrewAI 通过角色化的 Agent 设计和流程化的任务编排,实现了结构清晰的多智能体协作,非常适合内容创作和自动化工作流场景。


最后更新:2026年3月19日