Crew 全字段:把"一支团队"拆成一张 Pydantic 表
前 18 天你已经把 Agent(会干活的人)和 Task(要干的活)拆透了。今天进入阶段4——Crew(团队)。和 Agent 一样,Crew 本身也只是一个 BaseModel 数据模型:它把"哪些 agent、哪些 task、用什么流程、要不要记忆/规划/缓存、出事回调谁"全部登记成字段。真正"跑"是 kickoff 触发的(Day 22)。今天先把 crew.py 里 Crew 类的每一类字段逐个读懂——这是后面 7 天的地图。
Crew 就像一张「剧组开机登记表」:演员名单(agents)、拍摄计划(tasks)、拍法(process 是顺序拍还是有导演调度)、要不要留素材档案(memory)、要不要先开策划会(planning)、每拍完一条谁来签字(callbacks)。表填好了戏还没开拍——真正"喊 Action"是 kickoff()(Day 22)。今天就是把这张登记表上的每一栏读懂:它们分身份、核心、LLM、开关、回调、私有状态六类。痛点:Crew 到底"装"了什么?
Crew(agents=[...], tasks=[...]) 一行就能跑,但心里发虚:它内部到底存了多少东西?为什么有 manager_llm 又有 manager_agent 又有 planning_llm——这么多 LLM 字段各管什么?memory=True 和传一个 Memory() 实例有啥区别?出错了 before_kickoff_callbacks 会不会执行?不把字段表看一遍,后面读流程代码就总在"这个属性哪来的"里打转。Crew 是一个 FlowTrackable + BaseModel 的纯数据模型(crew.py:159)。它的字段分两拨:公开字段(用户传的,如 agents/tasks/process/memory,Pydantic Field 声明)和 私有属性(框架运行时用的,如 _memory/_rpm_controller,PrivateAttr 声明,用户不能传)。加上一串 @model_validator 在实例化时做体检(层级流程必须有 manager、顺序流程每个 task 必须有 agent……)。今天就按"六类字段 + 校验器"把这张表读完。类的骨架(crew.py:159):
# crew.py:159
class Crew(FlowTrackable, BaseModel):
"""Represents a group of agents, defining how they should collaborate
and the tasks they should perform."""
entity_type: Literal["crew"] = "crew" # 实体类型标记
__hash__ = object.__hash__ # 用对象身份做 hash
# —— 私有属性(用户不可传,框架运行时填)——
_rpm_controller: RPMController = PrivateAttr()
_memory: Memory | MemoryScope | MemorySlice | None = PrivateAttr(default=None)
_train: bool | None = PrivateAttr(default=False)
_inputs: dict[str, Any] | None = PrivateAttr(default=None)
_kickoff_event_id: str | None = PrivateAttr(default=None)
# —— 公开字段(用户可传)——
name: str | None = Field(default="crew")
tasks: list[Task] = Field(default_factory=list)
agents: Annotated[list[BaseAgent], BeforeValidator(_resolve_agents)] = Field(default_factory=list)
process: Process = Field(default=Process.sequential)
... # 后面几十个字段今天逐类看
xxx: T = Field(...) 是用户能填的登记栏;_xxx: T = PrivateAttr(...) 是框架自己用的草稿栏(下划线开头、外人别碰)。@model_validator 则是"表交上来后自动盖章检查"的规则。今天把登记栏按用途分成六摞,一摞一摞看。身份字段:id / name / key / fingerprint
身份相关字段(crew.py:198 起)与两个计算属性:
# crew.py:198
entity_type: Literal["crew"] = "crew" # 固定为 "crew",序列化/事件里用它区分实体
name: str | None = Field(default="crew") # 团队名,日志/记忆命名空间会用到
id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True) # 随机 UUID,frozen 不可改
# crew.py:857 计算属性:内容指纹(不是随机的,由成员内容决定)
@property
def key(self) -> str:
source = [agent.key for agent in self.agents] + [task.key for task in self.tasks]
return md5("|".join(source).encode(), usedforsecurity=False).hexdigest()
# crew.py:864 安全指纹(来自 security_config)
@property
def fingerprint(self) -> Fingerprint:
return self.security_config.fingerprint
entity_typeLiteral["crew"] 恒等于 "crew"。事件系统、序列化时靠它一眼认出"这是个 crew 不是 agent"。id (frozen=True)随机 UUID,创建后不可改。它标识"这一个实例",两个内容一样的 crew 也有不同 id。key(计算属性)★和 id 相反:它由成员的 key 拼起来做 md5。内容相同→key 相同。用于缓存命中、判断"是不是同一批活"。fingerprint安全指纹,来自 security_config。审计/追溯时标记"这次工具调用是哪个 crew 发起的"(Day 08 见过 agent 指纹)。id 做实例身份,key 做内容指纹——各司其职,是很常见的双标识设计。Crew,改了某个 agent 的 role("翻译" → "资深翻译")。那么 agent.key 变 → 拼进去的 source 变 → crew.key 的 md5 结果整个变了。而 crew.id 一动不动。所以缓存系统看 key 就知道"配置改了,旧缓存别用了"。核心三件套:agents / tasks / process
没有它们仨就没有 crew(crew.py:220):
# crew.py:220
tasks: list[Task] = Field(default_factory=list)
agents: Annotated[
list[BaseAgent],
BeforeValidator(_resolve_agents), # 传进来先过一道"解析"
] = Field(default_factory=list)
process: Process = Field(default=Process.sequential) # 默认顺序流程
tasks: list[Task]要干的活,有序列表。顺序流程下就是按这个顺序执行(Day 20)。默认空表。agents: list[BaseAgent]干活的人。类型是 BaseAgent(不是具体 Agent),所以自定义 agent 也能塞。BeforeValidator(_resolve_agents)★Pydantic 前置校验器:在赋值前先跑 _resolve_agents,把 "@org/name" 引用之类解析成真正的 agent 对象。用户传字符串也能用。process: Process枚举,默认 sequential(顺序)。另一个值 hierarchical(层级,D21)。这一个字段决定 kickoff 走哪条执行路径。for task_index, task in enumerate(tasks));就算是层级流程,manager 也需要一个稳定的任务清单去分派。用 set 会丢掉顺序,那"先后依赖"就无从表达了。agents 用 BaseAgent 而非 Agent 作类型:面向接口而非实现。这样 LiteAgent(Day 12)、适配第三方框架的 agent 只要实现了 BaseAgent 协议就能进 crew。扩展性靠这一个类型标注就留出来了。五个 LLM 字段:各管一摊
初学最懵的就是这堆 LLM 字段(crew.py:248 起,节选):
# crew.py:248
manager_llm: Annotated[str | BaseLLM | None, BeforeValidator(_validate_llm_ref), ...] = \
Field(description="Language model that will run the agent.", default=None)
manager_agent: Annotated[BaseAgent | None, BeforeValidator(_resolve_agent)] = \
Field(description="Custom agent that will be used as manager.", default=None)
function_calling_llm: Annotated[str | LLM | None, ...] = Field(
default=None,
deprecated="function_calling_llm is deprecated ...") # 已弃用
# crew.py:322
planning_llm: Annotated[str | BaseLLM | None, ...] = Field(
default=None,
description="Language model that will run the AgentPlanner if planning is True.")
chat_llm: Annotated[str | BaseLLM | None, ...] = Field(
default=None, description="LLM used to handle chatting with the crew.")
| 字段 | 什么时候用 | 谁在读它 |
|---|---|---|
manager_llm | 层级流程下,自动造一个"经理 agent"用的模型 | D21 _create_manager_agent |
manager_agent | 你想自己指定经理(而非自动造) | D21 同上 |
planning_llm | planning=True 时,开策划会的规划 agent 用的模型 | D23 CrewPlanner |
chat_llm | 和 crew 对话式交互 / 记忆分析兜底用 | D25 _memory_llm |
function_calling_llm | (已弃用)曾经统一给所有 agent 做工具调用 | 已标 deprecated |
str | BaseLLM | None三种都收:传字符串(如 "gpt-4o")、传 LLM 实例、或不传(None)。灵活输入。BeforeValidator(_validate_llm_ref)赋值前先校验/规范化这个 LLM 引用,把字符串等统一成框架认识的形式。PlainSerializer(_serialize_llm_ref...)序列化成 JSON 时用它转成 dict——因为 LLM 对象本身没法直接 JSON 化。deprecated=...★function_calling_llm 带 deprecated 标记:还能用但会告警,未来会删。读到这种字段就知道"别再用它了"。👶 小白:为什么不用一个 llm 字段管全部?
👨🏫 老师:因为这几件事需求不一样:经理要"会调度、上下文大"的强模型;规划只跑一次,可以用便宜快的;对话要低延迟。分成多个字段,你就能按岗位选模型、按预算配——用一个强模型跑规划纯属烧钱。字段分开,是把"成本/能力的调度权"交给用户。
开关字段:memory / cache / verbose / planning / max_rpm
一堆布尔/配置开关,其中 memory 最有意思(crew.py:225):
# crew.py:218
cache: bool = Field(default=True) # 工具结果缓存,默认开
verbose: bool = Field(default=False) # 详细日志
# crew.py:225 memory 不是简单 bool!
memory: Annotated[
bool
| Annotated[Memory | MemoryScope | MemorySlice, Field(discriminator="memory_kind")]
| None,
BeforeValidator(_ensure_memory_kind),
] = Field(default=False, description=(
"Enable crew memory. Pass True for default Memory(), "
"or a Memory/MemoryScope/MemorySlice instance for custom configuration."))
# crew.py:300
max_rpm: int | None = Field(default=None) # 每分钟最大请求数(限流)
# crew.py:318
planning: bool | None = Field(default=False,
description="Plan the crew execution and add the plan to the crew.")
cache=True(默认开)同样的工具+参数不重复调,直接取缓存结果。省钱省时间。memory: bool | Memory | ...★联合类型:传 True → 框架给你造默认 Memory();传实例 → 用你的配置;False → 关闭。一个字段吃三种用法(D25 详解)。discriminator="memory_kind"Pydantic 判别式联合:靠 memory_kind 字段区分传进来的到底是 Memory 还是 MemoryScope 还是 MemorySlice,解析更快更准。max_rpm=None默认不限流。设了数字,_rpm_controller 就会替所有 agent 踩刹车,防止把 API 打爆。planning=False默认关。开了会在 kickoff 前先跑一轮"规划",给每个 task 追加执行步骤(D23)。memory=True,框架自动配好一切(_ensure_memory_kind 把 True 转成默认配置)。高级用户想换存储后端/换 embedder:直接传一个配置好的 Memory(...) 实例。朴素做法是加一堆 memory_backend=/memory_embedder=/memory_enabled= 分散字段——参数爆炸且容易冲突。联合类型把"要不要 + 怎么配"收进一个字段,代价是类型复杂一点,但用户 API 干净得多。回调与钩子字段:在关键节点插手
让你在执行的关键点"插一脚"(crew.py:270 起):
# crew.py:270
step_callback: SerializableCallable | None = Field(default=None,
description="Callback to be executed after each step for all agents execution.")
task_callback: SerializableCallable | None = Field(default=None,
description="Callback to be executed after each task for every agents execution.")
before_kickoff_callbacks: list[SerializableCallable] = Field(default_factory=list,
description="List of callbacks to be executed before crew kickoff. "
"It may be used to adjust inputs before the crew is executed.")
after_kickoff_callbacks: list[SerializableCallable] = Field(default_factory=list,
description="List of callbacks to be executed after crew kickoff. "
"It may be used to adjust the output of the crew.")
# crew.py:397 校验器:把 None 从回调列表里过滤掉
@field_validator("before_kickoff_callbacks", "after_kickoff_callbacks", mode="before")
@classmethod
def _drop_unresolvable_callbacks(cls, value: Any) -> Any:
if isinstance(value, list):
return [v for v in value if v is not None]
return value
step_callback每个 agent 每走一步(一圈 ReAct)后回调你。用来做进度条、监控。task_callback每个 task 完成后回调。粒度比 step 粗。before_kickoff_callbacks(列表)★kickoff 前跑,可以改 inputs——返回值会替换原 inputs(D22 会看到 prepare_kickoff 里 normalized = before_callback(normalized))。after_kickoff_callbacks(列表)kickoff 后跑,可以改最终输出(result = after_callback(result))。做后处理、格式转换。_drop_unresolvable_callbacks校验器把列表里的 None 剔掉——防止反序列化时留下无法还原的空回调把执行搞崩。before/after_kickoff_callbacks 是 list(可挂多个,按顺序链式执行,前一个的返回喂给后一个);而 step_callback/task_callback 是单个可空函数。所以想在 kickoff 前挂两个预处理,要写 before_kickoff_callbacks=[f1, f2];而 step 级只能设一个。搞混类型会直接 Pydantic 校验报错。私有状态 + 实例化时的"体检"校验器
私有属性(用户不可传)与几个关键 @model_validator:
# crew.py:603 实例化后:填私有属性、建限流器
@model_validator(mode="after")
def set_private_attrs(self) -> Crew:
if not getattr(self, "_cache_handler", None):
self._cache_handler = CacheHandler()
...
self._logger = Logger(verbose=self.verbose)
self._rpm_controller = RPMController(max_rpm=self.max_rpm, logger=self._logger)
return self
# crew.py:696 层级流程必须有 manager
@model_validator(mode="after")
def check_manager_llm(self) -> Self:
if self.process == Process.hierarchical:
if not self.manager_llm and not self.manager_agent:
raise PydanticCustomError("missing_manager_llm_or_manager_agent",
"Attribute `manager_llm` or `manager_agent` is required "
"when using hierarchical process.", {})
if (self.manager_agent is not None) and (self.agents.count(self.manager_agent) > 0):
raise PydanticCustomError("manager_agent_in_agents",
"Manager agent should not be included in agents list.", {})
return self
# crew.py:741 顺序流程每个 task 必须绑定 agent
@model_validator(mode="after")
def validate_tasks(self) -> Self:
if self.process == Process.sequential:
for task in self.tasks:
if task.agent is None:
raise PydanticCustomError("missing_agent_in_task",
"Sequential process error: Agent is missing in the task ...", {})
return self
mode="after"这些校验器在字段都填好之后跑(能同时看到 process 和 manager_llm 等多个字段做联合判断)。set_private_attrs实例化时建好私有工具:缓存处理器、日志器、限流控制器。用户没传但框架内部要用。check_manager_llm★层级流程必须有 manager_llm 或 manager_agent,否则没人当经理调度——直接在建对象时报错,不拖到 kickoff。manager 不能在 agents 里经理不能同时是普通干活的——否则调度关系乱套。也在这里拦下。validate_tasks★顺序流程每个 task 都得有 agent(没经理分派,只能自己指定谁干)。层级流程可以不指定(经理会分)。kickoff() 才报,你可能已经调了几十次 LLM、烧了钱、跑了几分钟才崩。放在 @model_validator(mode="after") 里,意味着 Crew(...) 那一行就报错——配置错在写代码时立刻暴露,成本几乎为零。把"配置合法性"这类静态检查前移到构造期,是 Pydantic 模型的核心价值之一。边界 + 今日小结
check_config(crew.py:720):if not self.config and not self.tasks and not self.agents: raise ...("Either 'agents' and 'tasks' need to be set or 'config'.")。也就是说你不能造一个三样全空的 crew——那是个啥也干不了的空壳。要么直接给 agents+tasks,要么给一个 config(YAML 配置,D55 讲)让框架从配置里 _setup_from_config() 生成。空壳会在构造期就被拦。👶 小白:token_usage 和 usage_metrics 两个字段是重复吗?
👨🏫 老师:几乎一样,都记 LLM 用量。usage_metrics 是老字段,token_usage 是新加的(crew.py:388),kickoff 结束时两个都会被填。这是字段演进期的常见现象——新旧并存一段时间保证向后兼容。读源码碰到"看起来重复的字段",先看注释和 deprecated 标记判断哪个是当下推荐的。
🧠 今天你应该能回答
- Crew 是数据模型还是执行器?字段分哪两大类(Field vs PrivateAttr)?
id和key有什么区别?各回答什么问题?- 五个 LLM 字段各管什么岗位?为什么不合并成一个?
memory为什么是联合类型?True和传实例有何不同?- before/after 回调是列表、step/task 是单个——差异在哪?
- 为什么层级流程缺 manager 会在"建对象"时就报错,而不是 kickoff 时?
✋ 10 分钟动手
P=lib/crewai/src/crewai
sed -n '159,240p' $P/crew.py # 类头 + 私有属性 + 核心字段
sed -n '248,335p' $P/crew.py # LLM 字段 + 开关字段
sed -n '696,760p' $P/crew.py # 三个体检校验器
# 亲手触发一次校验错误:层级流程不给 manager
python -c "
from crewai import Crew, Agent, Task, Process
a=Agent(role='写手', goal='写', backstory='x')
t=Task(description='写一句', expected_output='一句话', agent=a)
try:
Crew(agents=[a], tasks=[t], process=Process.hierarchical) # 缺 manager_llm
except Exception as e:
print('如期报错:', e)
"
process 字段最常用的那个值——Process.sequential。它的执行入口是 _run_sequential_process → _execute_tasks,我们逐行读那个 for task in tasks 主循环:任务怎么按序执行、上一个的输出怎么变成下一个的 context、异步任务怎么攒着最后一起 join。