CrewAI 采用流程驱动的编排模型——Crew 类通过 Process 枚举确定执行模式,编排 Agent 实例执行 Task 对象。核心设计理念:约定优于配置——role/goal/backstory 是所有框架中最简洁的 Agent 定义方式。
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 # 输出验证
Task 按列表顺序执行,每个 Task 的输出通过 context 传递给下一个。最简单直接的模式。
Manager Agent 使用委托工具(DelegateWorkTool、AskQuestionTool)将任务分配给工作者 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 成员:
DelegateWorkTool — "Delegate a specific task to one of the following coworkers: {coworker_names}"AskQuestionTool — "Ask a specific question to one of the following coworkers: {coworker_names}"CrewAI 的系统提示词构建是所有框架中最简单的——直接拼接 role/goal/backstory:
You are {role}.
{backstory}
Your personal goal is: {goal}
没有动态提示词注入、没有模板变量、没有条件片段。这使其对非技术用户最友好,但也意味着提示词灵活性最低。
工具转换为 function-calling 格式。委托工具描述中动态注入同事名称:{coworker_names}。
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,提交到并发线程池。后续同步任务会等待所有挂起的异步任务完成。
crew.kickoff(
checkpoint_kickoff_event_id=previous_event_id
)
支持从检查点恢复——如果 Task 已有输出则跳过。
超越基础 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"
@start() — 流程入口点@listen() — 响应方法输出@router() — 条件分支@or_() / @and_() — 组合多触发器Flow 的装饰器模式比 LangGraph 的 StateGraph 更直观,比 Google ADK 的 SequentialAgent 更灵活。
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" # 用户级别
search() — 语义搜索save() — 存储新记忆analyze() — AI 驱动的记忆分析reset() — 清除记忆启用记忆后,Agent 自动获得内置记忆工具:
save_memory — 存储信息search_memory — 检索相关记忆4 层统一记忆是仅次于 Agno 的最完整记忆方案,但缺少 Agno 的 Cultural Knowledge 创新。
CrewAI 拥有独特的任务级 Guardrail 系统:
Task(
guardrail=lambda output: (True, output), # (通过, 输出)
...
)
两种类型:
Guardrail 失败时,Task 可以带着反馈信息重试——这是其他框架没有的输出质量控制机制。
crew.train())crewai/crew.py — Crew 编排器(2305 行)crewai/agent/core.py — Agent 定义(1898 行)crewai/task.py — Task 定义(1468 行)crewai/flow/flow.py — Flow 编排(3636 行)crewai/memory/unified_memory.py — 记忆系统(1062 行)crewai/process.py — Process 枚举crewai/agents/crew_agent_executor.py — Agent 执行循环crewai/tools/ — 工具系统crewai/a2a/ — A2A 协议支持crewai/utilities/i18n.py — 国际化@start、@listen、@router)比 LangGraph 的 StateGraph 更直观,比 Google ADK 的 SequentialAgent 更灵活。Unified Memory 4 层方案是仅次于 Agno 的最完整记忆系统,但缺少 Cultural Knowledge 创新。Guardrail 是独有亮点——LLM 级别的输出质量控制,其他框架都没有。