Day 05 / 共 60 天 · 阶段1 入门与心智

Crew 组装：把员工和派工单编成团队

前两天分别认识了 Agent（员工）和 Task（派工单）。今天把它们装进一个容器——Crew。Crew 持有 agents + tasks + process（排班规矩），外加一堆团队级配置（记忆、缓存、回调、manager）。我们看它的核心字段，以及两种 process——sequential（流水线） 和 hierarchical（组长派活）——分别怎么把任务派发出去。源码在全仓最大的 crew.py（2400 行）。

📍 你在 60 天里的位置（阶段1：入门与心智 · 共 6 天）

D01 项目全景→ D02 装环境跑通→ D03 Agent→ D04 Task→ D05 Crew→ D06 kickoff 旅程→ 阶段2 Agent 深入

💡 先用一个类比兜住今天 Crew 是一个项目组。agents 是组员名单，tasks 是待办清单，process 是"怎么干"的规矩：sequential = 流水线，一张单子接一张，前一个的产出喂给后一个；hierarchical = 配一个组长（manager），由组长决定把哪张单子派给谁、还能追问返工。再加上团队公共资源：共享记忆（memory）、缓存（cache）、开跑前后的回调钩子。今天认识这个容器，明天（D06）看它按一下 kickoff 的完整旅程。

L01

痛点：一堆 agent 和 task 摆在那，谁来调度

🤔 痛点你造了 3 个 agent、5 个 task，然后呢？谁决定先干哪个、把上一个的结果传给下一个、任务失败了怎么办、多个任务能不能并行、团队要不要共享一份记忆？如果这些调度逻辑要你自己手写 for 循环+传参+异常处理，很快就乱成一团。

💡 本质：Crew = 团队容器 + 调度器Crew 一手持有资源（agents/tasks），一手持有调度策略（process），并提供统一入口 kickoff()。它是 FlowTrackable + BaseModel（crew.py:159）。

# crew.py:159
class Crew(FlowTrackable, BaseModel):
    """..."""

BaseModel又是 Pydantic 模型：字段自动校验、可序列化、可复制（crew.copy()）。

FlowTrackable混入类，让 Crew 能被 Flow（阶段7 的事件驱动编排）追踪——即"Crew 可以作为 Flow 里的一个步骤"。呼应 D01 说的两条主线可以嵌套。

大白话Agent 是人、Task 是事，Crew 是"把人和事装在一起、并规定怎么干"的项目组。你只管填名单和待办，调度交给它。

L02

三大核心字段：agents / tasks / process

整个 Crew 的骨架就这三个（crew.py:220）：

# crew.py:220
tasks: list[Task] = Field(default_factory=list)
agents: Annotated[
    list[BaseAgent],
    BeforeValidator(_resolve_agents),        # 校验时把各种 agent 引用解析成对象
] = Field(default_factory=list)
process: Process = Field(default=Process.sequential)   # ★默认流水线

tasks: list[Task]待办清单，有序。sequential 流程严格按这个顺序执行。

agents: list[BaseAgent]组员名单。注意类型是 BaseAgent（D03 的契约），所以适配的外部 agent 也能进来。

process★排班规矩，默认 Process.sequential。它的取值来自 D01 见过的 11 行枚举 process.py：只有 sequential 和 hierarchical（consensual 还是 TODO）。

default_factory=list小细节：用工厂函数而非 [] 当默认，避免"可变默认参数被多个实例共享"的经典 Python 坑。

📝 process 枚举本体（就这么小）

# process.py:4
class Process(str, Enum):
    sequential = "sequential"
    hierarchical = "hierarchical"
    # TODO: consensual = 'consensual'

继承 str 让它既是枚举又能当字符串用（process="sequential" 也行）。

L03

团队级配置字段群

核心三件之外，一批字段配置"整个团队的公共行为"（crew.py:218 起）：

# crew.py:218
name: str | None = Field(default="crew")
cache: bool = Field(default=True)                 # 团队级工具结果缓存（默认开）
verbose: bool = Field(default=False)              # 详细日志
memory: ... = Field(default=False, ...)           # ★共享记忆：True 用默认 Memory()
manager_llm: ... = Field(default=None, ...)       # hierarchical 时给自动 manager 用的模型
manager_agent: ... = Field(default=None, ...)     # ★自定义 manager（组长）
step_callback: ... = Field(default=None, ...)     # 每一步后的回调
task_callback: ... = Field(default=None, ...)     # 每个任务后的回调
before_kickoff_callbacks: list = Field(default_factory=list, ...)  # 开跑前钩子（D02 见过）
after_kickoff_callbacks: list = Field(default_factory=list, ...)   # 跑完后钩子

字段	作用
`memory=True`	开启团队共享记忆，agent 之间/多轮之间能记事（阶段6 深挖；注意它拖 lancedb，见 D01 惰性导入）
`cache=True`	相同工具调用复用结果，默认开，省钱
`manager_agent` / `manager_llm`	hierarchical 流程的组长（L05）
`before/after_kickoff_callbacks`	开跑前改 inputs、跑完后改结果（D02 的 prepare_kickoff 里被调用）
`step_callback` / `task_callback`	细粒度观测：每步/每任务后触发，做日志或干预

这些几乎都有默认值，最小用法（D01 那段）只填 agents 和 tasks 就能跑。memory 默认是 False——记忆是要显式开的，呼应 D01 "重能力默认关"的一贯设计。

图注：Crew = 核心三件（agents/tasks/process）+ 一圈团队级公共配置。最小用法只需前两件。

L04

sequential：最朴素的流水线

kickoff 时按 process 分流（D02 L05 见过）。sequential 的实现只有一行转发（crew.py:1475）：

# crew.py:1475
def _run_sequential_process(self) -> CrewOutput:
    """Executes tasks sequentially and returns the final output."""
    return self._execute_tasks(self.tasks)      # 直接把全部任务丢给主循环

直接调 _execute_taskssequential 没有额外准备——因为每个 task 自带 agent（D04 讲过），不需要"分配执行者"这一步。主循环按 tasks 顺序一个个跑。

前一个喂后一个L06 会看到：执行时前面任务的输出被当作后面任务的 context 传入。这就是"流水线"——上一道工序的产出是下一道的原料。

大白话sequential = 排好队一个接一个干，前面的成果自动往后传。90% 的场景用它就够。

L05

hierarchical：配一个组长来派活

hierarchical 在跑之前，多一步"创建组长"（crew.py:1479 / crew.py:1484）：

# crew.py:1479
def _run_hierarchical_process(self) -> CrewOutput:
    self._create_manager_agent()               # ★先造/配组长
    return self._execute_tasks(self.tasks)

# crew.py:1484
def _create_manager_agent(self) -> None:
    if self.manager_agent is not None:          # 你自带了组长
        self.manager_agent.allow_delegation = True
        manager = self.manager_agent
        if manager.tools:                       # ★边界：组长不该拿工具
            manager.tools = []
            raise Exception("Manager agent should not have tools")
    else:                                        # 没自带 → 用 manager_llm 自动造一个
        self.manager_llm = create_llm(self.manager_llm)
        i18n = get_i18n(prompt_file=self.prompt_file)
        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, verbose=self.verbose)
        self.manager_agent = manager
    manager.crew = self

两种来源组长要么是你传的 manager_agent，要么框架用 manager_llm 自动生成一个——所以 hierarchical 时这两个至少要给一个（否则 :699 附近会报错）。

allow_delegation=True组长的核心能力就是委派——把任务分给合适的组员。这个开关必须开。

role/goal/backstory 来自 i18n自动组长的人设从翻译文件取（hierarchical_manager_agent），所以支持多语言，且行为一致。

tools=AgentTools(...).tools()给组长的不是普通工具，而是"把活委派给某个组员"的委派工具。组长靠它调度。

manager.tools 非空则报错★边界：组长不该持有执行类工具——它的职责是"派活"不是"干活"。检测到就清空并抛异常。

💡 设计取舍①：为什么组长要"没有工具、只有委派权"？如果组长既能自己用工具干活、又能派活，它很容易"越权亲自下场"，团队就退化成一个大 agent，分工失效。源码用一条硬规则——manager 不许有 tools，检测到就 raise——把组长的角色锁死在"纯调度"。这是把"职责单一"从口头约定变成代码强制的典型：宁可报错也不允许角色越界。代价是灵活性下降（你不能让组长顺手干点活），回报是 hierarchical 的行为可预测、分工清晰。

L06

_execute_tasks：两种流程共用的派发主循环

不管哪种 process，最后都汇到同一个主循环 _execute_tasks（crew.py:1519）。看它的骨架：

# crew.py:1543（裁剪）
for task_index, task in enumerate(tasks):
    exec_data, task_outputs, last_sync_output = prepare_task_execution(
        self, task, task_index, start_index, task_outputs, last_sync_output)
    if exec_data.should_skip:                       # 断点恢复时跳过已完成的
        continue
    if isinstance(task, ConditionalTask):           # 条件任务：不满足就跳过
        skipped = self._handle_conditional_task(...)
        if skipped: task_outputs.append(skipped); continue

    if task.async_execution:                        # ★异步任务：丢线程池，先不等
        context = self._get_context(task, [last_sync_output] if last_sync_output else [])
        future = task.execute_async(agent=exec_data.agent, context=context, tools=exec_data.tools)
        futures.append((task, future, task_index))
    else:                                            # 同步任务
        if futures:                                  # 先把攒着的异步任务结果收回来
            task_outputs.extend(self._process_async_tasks(futures, was_replayed)); futures.clear()
        context = self._get_context(task, task_outputs)   # ★前面任务的输出当上下文
        task_output = task.execute_sync(agent=exec_data.agent, context=context, tools=exec_data.tools)
        task_outputs.append(task_output)
        self._process_task_result(task, task_output)
if futures:
    task_outputs.extend(self._process_async_tasks(futures, was_replayed))
return self._create_crew_output(task_outputs)        # ★汇成 CrewOutput

prepare_task_execution每个任务执行前先"备料"：确定用哪个 agent（sequential 用 task.agent，hierarchical 由 manager 定）、准备工具、判断要不要跳过。

_get_context(task, task_outputs)★流水线的灵魂：把已完成任务的输出组装成 context 传给当前任务。呼应 D04 的 context 字段。

task.execute_sync(...)调用 D04 讲过的执行入口，把活真正交给 agent。返回 TaskOutput 收集起来。

async_execution 分支异步任务不阻塞：先 execute_async 拿 Future 攒着，等遇到下一个同步任务或全部结束时统一收割（_process_async_tasks）。阶段3 D17 深挖。

ConditionalTask条件任务：根据前面结果判断要不要跑，跳过则记一个"跳过输出"。阶段3 D18 讲。

_create_crew_output★所有 TaskOutput 收齐 → 汇成一个 CrewOutput 返回。这就是 D02 见过的最终返回值的来源。

图注：主循环逐个处理 task，同步立即执行、异步攒着并发；前序输出经 context 流向后序任务。

L07

边界 + 今日小结

⚠️ 边界：hierarchical 忘了给 manager，会怎样？选了 process=Process.hierarchical，却既没给 manager_agent 也没给 manager_llm，源码在校验阶段就报错（crew.py:699 附近）："Attribute manager_llm or manager_agent is required when using hierarchical process."。另一个坑（L05 已见）：给的 manager_agent 带了 tools → 直接 raise Exception("Manager agent should not have tools")。还有 crew.py:708：如果你把 manager_agent 也塞进了 agents 列表里（重复），同样会报错——组长不能同时是普通组员。记住：hierarchical 三条铁律——必须有 manager、manager 不能带工具、manager 不能进 agents 名单。

💡 设计取舍②：两种 process 为什么共用 _execute_tasks，只在"谁来定 agent"上分叉？ sequential 和 hierarchical 看似差别很大，但源码没有为它们各写一套循环。区别被收敛到一个点：任务的执行者从哪来——sequential 用 task.agent（你预先指定），hierarchical 由 manager 动态委派。其余（上下文传递、异步收割、结果汇总、条件/断点处理）完全复用同一个 _execute_tasks。好处：调度逻辑只有一份，改一处两种流程都受益，bug 面小。代价是 _execute_tasks 里要兼容两种"agent 来源"，读起来分支多一点。这是"把差异下沉到最小的一个决策点、其余共享"的经典抽象手法。

👶 小白：我该用 sequential 还是 hierarchical？

👨‍🏫 老师：任务顺序你心里清楚、能一条条排好 → sequential（简单、可控、省一个 manager 的模型开销）。任务怎么分配需要"临场判断"、或者你希望有个组长统筹调度和返工 → hierarchical。新手先用 sequential 打通，理解了再上 hierarchical。阶段4（D20/D21）会把两者掰开揉碎。

                    🧠 今天你应该能回答
                    Crew 的三大核心字段？（agents / tasks / process）
process 有哪几种？默认哪种？（sequential 默认 / hierarchical；consensual 还是 TODO）
memory 默认开吗？（不开，False，要显式开——重能力默认关）
sequential 怎么把结果往后传？（前序 TaskOutput 经 _get_context 变成后序任务的 context）
hierarchical 的 manager 从哪来、有什么限制？（自带或 manager_llm 自动造；不能有工具、不能进 agents 名单、必须存在）
两种 process 共用什么？（_execute_tasks 主循环，只在"谁来定执行 agent"上分叉）

                

✋ 10 分钟动手

# 1. 看核心字段与两种 process 实现
sed -n '218,257p' lib/crewai/src/crewai/crew.py
sed -n '1475,1509p' lib/crewai/src/crewai/crew.py

# 2. 看派发主循环
sed -n '1543,1588p' lib/crewai/src/crewai/crew.py

# 3. 故意触发 hierarchical 的边界报错，加深印象
uv run python -c "
from crewai import Agent, Task, Crew, Process
a=Agent(role='R', goal='G', backstory='B', llm='gpt-4o')
t=Task(description='do X', expected_output='Y', agent=a)
# 没给 manager_llm/manager_agent，看它报什么
Crew(agents=[a], tasks=[t], process=Process.hierarchical).kickoff()
"

明日预告 · Day 06：阶段1 收官！明天把 D01~D05 全部串起来——完整走一遍 crew.kickoff() 的旅程：从 prepare_kickoff → 分流 process → _execute_tasks 派发 → task.execute_sync → agent.execute_task → executor 循环 → 打包 TaskOutput → 汇成 CrewOutput。一张大图看懂"按一下按钮，数据在五大主角间怎么流动"。

← Day 04 Task 是什么 Day 06 · 一次 kickoff 旅程 →