Day 19 / 共 60 天 · 阶段4 Crew 与流程

Crew 全字段:把"一支团队"拆成一张 Pydantic 表

前 18 天你已经把 Agent(会干活的人)和 Task(要干的活)拆透了。今天进入阶段4——Crew(团队)。和 Agent 一样,Crew 本身也只是一个 BaseModel 数据模型:它把"哪些 agent、哪些 task、用什么流程、要不要记忆/规划/缓存、出事回调谁"全部登记成字段。真正"跑"是 kickoff 触发的(Day 22)。今天先把 crew.pyCrew 类的每一类字段逐个读懂——这是后面 7 天的地图。

📍 你在 60 天里的位置(阶段4 Crew 与流程 · 共 8 天)
阶段3 Task D13-18 D19 Crew 全字段 D20 顺序流程 D21 层级+manager D22 kickoff 家族 D23 规划 D24 训练+replay D25 记忆开关 D26 事件系统
💡 先用一个类比兜住今天 Crew 就像一张「剧组开机登记表」:演员名单(agents)、拍摄计划(tasks)、拍法(process 是顺序拍还是有导演调度)、要不要留素材档案(memory)、要不要先开策划会(planning)、每拍完一条谁来签字(callbacks)。表填好了戏还没开拍——真正"喊 Action"是 kickoff()(Day 22)。今天就是把这张登记表上的每一栏读懂:它们分身份、核心、LLM、开关、回调、私有状态六类。
L01

痛点: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_controllerPrivateAttr 声明,用户不能传)。加上一串 @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 则是"表交上来后自动盖章检查"的规则。今天把登记栏按用途分成六摞,一摞一摞看。
数据结构:Crew 字段的六个抽屉 Crew (BaseModel) ① 身份 id/name/key fingerprint ② 核心 agents/tasks process ③ 五个 LLM manager/planning chat/function... ④ 开关 memory/cache planning/max_rpm ⑤ 回调/钩子 before/after_kickoff step/task_callback ⑥ 私有状态 PrivateAttr _memory/_rpm_controller _train/_inputs
图注:Crew 的字段可归入六个抽屉,前五个是用户可填的公开字段,第六个是框架私有草稿。
L02

身份字段: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? id 回答"是哪一个",key 回答"是不是同一种"。如果只有随机 id,两次用同样配置跑就没法复用缓存(id 每次都不同);如果只有内容 key,你就没法区分同配置的两个独立实例(日志会混)。源码两个都留id 做实例身份,key 做内容指纹——各司其职,是很常见的双标识设计。
📝 例子:key 什么时候变 你有一个 Crew,改了某个 agent 的 role("翻译" → "资深翻译")。那么 agent.key 变 → 拼进去的 source 变 → crew.key 的 md5 结果整个变了。而 crew.id 一动不动。所以缓存系统看 key 就知道"配置改了,旧缓存别用了"。
L03

核心三件套: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 走哪条执行路径。
💡 为什么 tasks 是 list(有序)而不是 set? 因为任务天然有先后——"先调研,再写稿,最后校对"。顺序流程直接按列表下标一个个执行(Day 20 会看到 for task_index, task in enumerate(tasks));就算是层级流程,manager 也需要一个稳定的任务清单去分派。用 set 会丢掉顺序,那"先后依赖"就无从表达了。
agentsBaseAgent 而非 Agent 作类型:面向接口而非实现。这样 LiteAgent(Day 12)、适配第三方框架的 agent 只要实现了 BaseAgent 协议就能进 crew。扩展性靠这一个类型标注就留出来了。
L04

五个 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_llmplanning=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 字段管全部?

👨‍🏫 老师:因为这几件事需求不一样:经理要"会调度、上下文大"的强模型;规划只跑一次,可以用便宜快的;对话要低延迟。分成多个字段,你就能按岗位选模型、按预算配——用一个强模型跑规划纯属烧钱。字段分开,是把"成本/能力的调度权"交给用户。

L05

开关字段: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 为什么设计成"bool 或 实例"的联合类型? 为了同时照顾"小白"和"高级用户"。小白只想开记忆:写 memory=True,框架自动配好一切(_ensure_memory_kind 把 True 转成默认配置)。高级用户想换存储后端/换 embedder:直接传一个配置好的 Memory(...) 实例。朴素做法是加一堆 memory_backend=/memory_embedder=/memory_enabled= 分散字段——参数爆炸且容易冲突。联合类型把"要不要 + 怎么配"收进一个字段,代价是类型复杂一点,但用户 API 干净得多。
L06

回调与钩子字段:在关键节点插手

让你在执行的关键点"插一脚"(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_kickoffnormalized = before_callback(normalized))。
after_kickoff_callbacks(列表)kickoff 后跑,可以改最终输出result = after_callback(result))。做后处理、格式转换。
_drop_unresolvable_callbacks校验器把列表里的 None 剔掉——防止反序列化时留下无法还原的空回调把执行搞崩。
⚠️ 边界:before/after 是列表,step/task 是单个 注意类型差异:before/after_kickoff_callbackslist(可挂多个,按顺序链式执行,前一个的返回喂给后一个);而 step_callback/task_callback单个可空函数。所以想在 kickoff 前挂两个预处理,要写 before_kickoff_callbacks=[f1, f2];而 step 级只能设一个。搞混类型会直接 Pydantic 校验报错。
L07

私有状态 + 实例化时的"体检"校验器

私有属性(用户不可传)与几个关键 @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 时"? 越早失败越好(fail fast)。如果层级流程缺 manager 这种错拖到 kickoff() 才报,你可能已经调了几十次 LLM、烧了钱、跑了几分钟才崩。放在 @model_validator(mode="after") 里,意味着 Crew(...) 那一行就报错——配置错在写代码时立刻暴露,成本几乎为零。把"配置合法性"这类静态检查前移到构造期,是 Pydantic 模型的核心价值之一。
L08

边界 + 今日小结

⚠️ 边界:至少要有 (agents+tasks) 或 config 还有一个校验器 check_configcrew.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_usageusage_metrics 两个字段是重复吗?

👨‍🏫 老师:几乎一样,都记 LLM 用量。usage_metrics 是老字段,token_usage 是新加的(crew.py:388),kickoff 结束时两个都会被填。这是字段演进期的常见现象——新旧并存一段时间保证向后兼容。读源码碰到"看起来重复的字段",先看注释和 deprecated 标记判断哪个是当下推荐的。

🧠 今天你应该能回答

  • Crew 是数据模型还是执行器?字段分哪两大类(Field vs PrivateAttr)?
  • idkey 有什么区别?各回答什么问题?
  • 五个 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)
"
明日预告 · Day 20:字段读完了,明天看 process 字段最常用的那个值——Process.sequential。它的执行入口是 _run_sequential_process → _execute_tasks,我们逐行读那个 for task in tasks 主循环:任务怎么按序执行、上一个的输出怎么变成下一个的 context、异步任务怎么攒着最后一起 join。
← Day 18 条件任务 Day 20 · 顺序流程 →