← 返回总览

CrewAI

🔬 DEEP
🧩 Agent Frameworks Python ⭐ 30k+ | GitHub → | 更新于 2026-05-17 18:59
Process-DrivenRole-Based AgentsFlow OrchestrationDelegationGuardrailsi18n

🏗️ 核心架构

CrewAI 采用流程驱动的编排模型——Crew 类通过 Process 枚举确定执行模式,编排 Agent 实例执行 Task 对象。核心设计理念:约定优于配置——role/goal/backstory 是所有框架中最简洁的 Agent 定义方式。

核心三件套:Crew + Agent + Task

class Crew(FlowTrackable, BaseModel):
    agents: list[Agent]      # 团队成员
    tasks: list[Task]        # 工作项
    process: Process          # sequential 或 hierarchical
    manager_agent: Agent      # hierarchical 模式的管理者
    manager_llm: LLM          # 自动创建管理者(无 manager_agent 时)

class Agent(BaseAgent):
    role: str                 # "Researcher"
    goal: str                 # "Find information"
    backstory: str            # "You are an experienced researcher..."
    tools: list[Tool]         # 可用工具
    allow_delegation: bool    # 是否可委托其他 Agent

class Task(BaseModel):
    description: str          # 任务描述
    agent: Agent              # 负责的 Agent
    async_execution: bool     # 是否并行执行
    context: list[Task]       # 依赖其他任务的输出
    output_json/pydantic      # 结构化输出
    guardrail                 # 输出验证

Process 类型

Sequential(顺序)

Task 按列表顺序执行,每个 Task 的输出通过 context 传递给下一个。最简单直接的模式。

Hierarchical(层级)

Manager Agent 使用委托工具(DelegateWorkToolAskQuestionTool)将任务分配给工作者 Agent。Manager 决定谁做什么。

class Process(str, Enum):
    sequential = "sequential"      # 按序执行
    hierarchical = "hierarchical"  # 管理者委托
    # TODO: consensual = 'consensual'  (未实现)

👔 层级管理与委托

Hierarchical 模式下,如果没有提供 manager_agent,CrewAI 会自动创建:

def _create_manager_agent(self):
    if self.manager_agent is None:
        manager = Agent(
            role=i18n.retrieve("hierarchical_manager_agent", "role"),
            goal=i18n.retrieve("hierarchical_manager_agent", "goal"),
            backstory=i18n.retrieve("hierarchical_manager_agent", "backstory"),
            tools=AgentTools(agents=self.agents).tools(),  # 委托工具
            allow_delegation=True,
            llm=self.manager_llm,
        )

Manager 获得特殊的委托工具,可以分配工作给任何 Crew 成员:

📝 提示词工程(源码级分析)

CrewAI 的系统提示词构建是所有框架中最简单的——直接拼接 role/goal/backstory:

You are {role}.
{backstory}

Your personal goal is: {goal}

没有动态提示词注入、没有模板变量、没有条件片段。这使其对非技术用户最友好,但也意味着提示词灵活性最低。

工具描述

工具转换为 function-calling 格式。委托工具描述中动态注入同事名称:{coworker_names}

i18n 国际化系统

CrewAI 是唯一内置国际化系统的框架:utilities/i18n.py 支持多语言提示词模板,prompt_file 参数允许自定义模板。

♻️ 主循环

Crew.kickoff()
  → _process_handler()
    → Sequential: _run_sequential_process()
      → _execute_tasks(tasks)
        → for each task:
            → task.execute(agent, context, tools)
              → agent.execute_task(task, context)
                → agent_llm.call(tools=tools, ...)
                → (tool_calls → execute → LLM call)*  → ReAct 循环
                → TaskOutput
            → append to task_outputs
            → if async_execution: 提交到线程池

    → Hierarchical: _run_hierarchical_process()
      → _create_manager_agent()
      → _execute_tasks(tasks)  → manager 决定委托

异步任务执行

Task 可标记 async_execution=True,提交到并发线程池。后续同步任务会等待所有挂起的异步任务完成。

Checkpoint/Resume

crew.kickoff(
    checkpoint_kickoff_event_id=previous_event_id
)

支持从检查点恢复——如果 Task 已有输出则跳过。

🌊 Flow 系统(高级编排)

超越基础 Crew 模式,CrewAI 提供了 Flow 系统(3636 行)用于复杂多步骤编排——这是 CrewAI 最强大的特性:

class Flow(FlowTrackable, BaseModel):
    # 装饰器驱动的流程定义
    @start()
    def generate_topic(self):
        return {"topic": "AI"}

    @listen("generate_topic")
    def write_article(self, topic):
        return {"article": "..."}

    @router("generate_topic")
    def route(self, topic):
        if topic == "AI":
            return "tech_path"
        return "general_path"

Flow 特性一览

Flow 的装饰器模式比 LangGraph 的 StateGraph 更直观,比 Google ADK 的 SequentialAgent 更灵活。

🧠 记忆系统(Unified Memory)

class UnifiedMemory(BaseModel):
    short_term_memory    # Crew 级,近期交互
    long_term_memory     # 跨 Crew 持久化
    entity_memory        # 人物/实体知识
    user_memory          # 用户偏好

记忆作用域

class MemoryScope(str, Enum):
    crew = "crew"        # Crew 级别
    task = "task"        # Task 级别
    agent = "agent"      # Agent 级别
    user = "user"        # 用户级别

记忆操作

记忆工具

启用记忆后,Agent 自动获得内置记忆工具:

4 层统一记忆是仅次于 Agno 的最完整记忆方案,但缺少 Agno 的 Cultural Knowledge 创新。

🛡️ Guardrails(任务护栏)

CrewAI 拥有独特的任务级 Guardrail 系统:

Task(
    guardrail=lambda output: (True, output),  # (通过, 输出)
    ...
)

两种类型:

Guardrail 失败时,Task 可以带着反馈信息重试——这是其他框架没有的输出质量控制机制。

🆕 独特特性

  1. Process-Driven 编排 — sequential/hierarchical 作为一等公民,开箱即用
  2. Role-Goal-Backstory 模式 — 最简洁直观的 Agent 定义方式
  3. Flow 系统 — 装饰器驱动的编排,超越基础 Crew
  4. 委托工具 — Agent 可委托工作、向同事提问
  5. Task Guardrails — 基于 LLM 的输出验证
  6. Training — Crew 可"训练"以改进表现(crew.train()
  7. Unified Memory — 4 层记忆 + 语义搜索
  8. A2A Protocol — 实验性 Agent-to-Agent 支持
  9. Human Feedback — Crew 和 Flow 均内置人机协作
  10. i18n — 唯一内置国际化的框架

🔑 核心类

CrewAgentTaskProcessFlowUnifiedMemoryConditionalTaskCrewOutputTaskOutputAgentToolsCacheHandlerGuardrail

📂 关键文件

⚖️ 架构权衡

✅ 优势
  • 最低学习门槛——role/goal/backstory 三行定义 Agent
  • Flow 装饰器比 LangGraph StateGraph 更直观
  • Guardrail 是独特的输出质量控制
  • i18n 支持非英语场景
  • 4 层统一记忆覆盖面广
⚠️ 代价
  • 提示词灵活性最低——无动态注入/模板/条件
  • Process 只有两种模式,复杂路由需降级到 Flow
  • 核心文件庞大(crew.py 2305 行、agent/core.py 1898 行)
  • consensual 模式未实现
  • 记忆缺少 Agno 的 Cultural Knowledge 创新
💡 洞察:CrewAI 的设计哲学是"约定优于配置"——role/goal/backstory 是所有框架中最简洁的 Agent 定义,让非技术用户零门槛上手。但简洁有代价:提示词工程是最不灵活的(无动态提示、无模板变量、无条件片段)。Process 枚举是双刃剑——sequential 和 hierarchical 覆盖 90% 场景,但无法表达复杂路由,必须降级到 Flow 系统。Flow 本身是 CrewAI 最强大的特性:装饰器编排(@start@listen@router)比 LangGraph 的 StateGraph 更直观,比 Google ADK 的 SequentialAgent 更灵活。Unified Memory 4 层方案是仅次于 Agno 的最完整记忆系统,但缺少 Cultural Knowledge 创新。Guardrail 是独有亮点——LLM 级别的输出质量控制,其他框架都没有。
自动研究系统生成 · 源码级深度分析 · 2026-05-17 18:59