CrewAI — 架构与原理
30 秒导读: CrewAI 是一个 Python 多智能体框架。它给你两种搭法:Crew——把几个有「角色」的 AI Agent 编成一支队伍,自治地协作完成一串 Task(像一个项目组);Flow——用装饰器把普通 Python 方法连成一张事件驱动的图,做精确、可控的工作流(像一条流水线)。两者能互相嵌套:Flow 里可以跑 Crew,Crew 里的 Agent 执行器本身底层就是一个 Flow。
1. 这是什么(零基础也能懂)
一句话定义: CrewAI 是一个让你用「角色化的 AI Agent」组队干活、或用「事件驱动的流程图」精确编排 LLM 调用的开源框架。
解决什么问题 / 给谁用。 假设你想让 AI 自动完成一个多步骤任务——比如「研究一个主题 → 写成报告 → 校对」。单次调一下大模型(LLM)往往不够:它需要分步骤、需要查资料(调工具)、需要不同「专长」的角色分工。CrewAI 给写应用的工程师提供两层抽象,把这套「让 LLM 反复思考+动手+协作」的脏活封装好。
它提供两种范式(README 开头即点明这是核心卖点):
| 范式 | 比喻 | 适合 | 控制粒度 |
|---|---|---|---|
| Crew(队伍) | 一个自治的项目组 | 开放式、需要协作/委派的任务 | 粗:你定角色和任务,过程交给 Agent |
| Flow(流程) | 一条事件驱动的流水线 | 确定性强、要精确控制走向的工作流 | 细:你画出方法之间的触发关系 |
它能做什么(功能):
- 定义有
role/goal/backstory的 Agent,自动生成系统提示词。 - 给 Agent 挂工具(Tool),Agent 在循环里自己决定何时调用。
- 用 Crew 把多个 Task 顺序执行(
sequential)或交给一个「经理」Agent 调度(hierarchical)。 - Agent 之间可委派(delegate)和互相提问。
- 用 Flow 写
@start/@listen/@router装饰的方法,构成事件驱动的有向图,支持并行、路由、持久化、人类反馈。 - 统一的 LLM 抽象(底层走 litellm,支持上百种模型)、记忆、知识库、事件总线。
用起来什么样(最小 Crew,示意,非源码):
from crewai import Agent, Task, Crew, Process
# 1) 定义一个有角色的 Agent
researcher = Agent(
role="Senior Researcher", # 它扮演谁
goal="Find the key facts about {topic}",
backstory="You are meticulous and cite sources.",
)
# 2) 定义一个任务,交给这个 Agent
task = Task(
description="Research {topic} and list 3 findings",
expected_output="A bullet list of 3 findings",
agent=researcher,
)
# 3) 组队并启动
crew = Crew(agents=[researcher], tasks=[task], process=Process.sequential)
result = crew.kickoff(inputs={"topic": "tidal energy"})
print(result.raw)
crew.kickoff(...) 是同步入口(crew.py:980 Crew.kickoff);它把任务一个个跑完,返回一个 CrewOutput。
一句话直觉/类比: 把 Crew 想成「你雇了一个项目组,只交代目标和任务清单,他们自己开会、分工、查资料」;把 Flow 想成「你自己画了一张流程图,规定 A 完成后触发 B,B 的结果决定走 C 还是 D」。Crew 重自治,Flow 重控制。
2. 顶层全景(它大概怎么转)
怎么读这张图: 从上到下是抽象层级,从「你写的代码」到「真正打 LLM」。Crew 和 Flow 是两个并列入口,最终都落到 Agent 的执行循环,再落到 LLM 抽象层。
你的代码
┌──────────┴───────────┐
Crew(自治队伍) Flow(事件驱动流程)
process=sequential/ @start / @listen /
hierarchical @router + or_/and_
│ │
│ 逐个执行 Task │ 方法完成 → 触发下游方法
▼ ▼
Task.execute_sync ───→ Agent.execute_task
│ 组装提示词 + 工具
▼
AgentExecutor(执行循环 = 主线)
├ ReAct 文本模式:解析 Thought/Action/Final Answer
└ 原生 function calling:LLM 直接吐 tool_calls
│ 反复「想→调工具→喂回结果」直到 AgentFinish
▼
LLM 抽象层(BaseLLM / LLM)
│
▼
litellm → 任意模型供应商
主要部 件一句话职责:
| 部件 | 干什么 | 在哪个文件(符号) |
|---|---|---|
Agent | 一个角色化的智能体,持有 LLM、工具、记忆;execute_task 是它干活的入口 | agent/core.py (Agent.execute_task) |
AgentExecutor | 核心循环:反复让 LLM 思考、调工具、喂回结果,直到拿到最终答案 | experimental/agent_executor.py(新默认);agents/crew_agent_executor.py(旧,已弃用) |
Task | 一个工作单元:描述 + 期望输出 + 指派的 Agent;_execute_core 把结果包成 TaskOutput | task.py (Task._execute_core) |
Crew | 编排器:按 sequential 或 hierarchical 把 Task 串起来跑 | crew.py (Crew.kickoff, _execute_tasks) |
Flow | 事件驱动引擎:根据方法间的 @listen/@router 关系调度执行 | flow/runtime/__init__.py (Flow.kickoff_async, _execute_listeners) |
Process | 枚举:sequential / hierarchical | process.py (Process) |
LLM / BaseLLM | 统一模型抽象,底层走 litellm | llm.py (LLM)、llms/base_llm.py (BaseLLM) |
crewai_event_bus | 全局事件总线,各阶段 emit 事件供监听器消费 | events/event_bus.py (CrewAIEventsBus) |
Memory | 统一记忆:LLM 分析后存储、自适应深度召回 | memory/unified_memory.py (Memory) |
主线走一遍(高层):
- 你调
crew.kickoff(inputs=...)(crew.py:980)。 - Crew 按
process选顺序或层级执行(crew.py:1037分支到_run_sequential_process/_run_hierarchical_process)。 - 对每个 Task 调
task.execute_sync(...),它进而调agent.execute_task(...)(crew.py:1586、task.py:790)。 - Agent 组装提示词,把活交给它的
agent_executor.invoke(...)(agent/core.py:905)。 - 执行器进入循环:让 LLM 回答 → 若是「要调工具」就执行工具、把结果喂回 → 重复,直到 LLM 给出最终答案(
AgentFinish)。 - 结果一路包成
TaskOutput→ 最后聚合成CrewOutput返回。
下面四章逐层展开。建议顺序: 先读 01-agent-loop(理解单个 Agent 怎么转,这是全框架的心脏)→ 02-crew-orchestration(多 Agent 怎么编排)→ 03-flows(另一套范式)→ 04-internals(横切的基础设施 + 巧妙之处 + 边界)。
阅读地图
- 01-agent-loop.md — Agent 的执行循环。ReAct 文本协议 vs 原生 function calling、解析器、最大迭代、上下文超限处理。框架的心 脏,先读这章。
- 02-crew-orchestration.md — Crew 编排。sequential / hierarchical、经理 Agent、Task 上下文传递、委派工具、异步任务、条件任务、guardrail。
- 03-flows.md — Flow 引擎。装饰器 DSL、事件触发模型、
or_/and_条件、路由、并行、状态与持久化。 - 04-internals.md — 横切设施:LLM 抽象层、统一记忆、事件总线;巧妙之处、边界与局限、横向对比、代码地图。