剥掉流式/检查点分支后的主干（crew.py:966）：

# crew.py:966
def kickoff(self, inputs=None, input_files=None, from_checkpoint=None) -> CrewOutput | ...:
    restored = apply_checkpoint(self, from_checkpoint)   # 检查点恢复（D24）
    if restored is not None:
        return restored.kickoff(inputs=inputs, input_files=input_files)
    get_env_context()
    if self.stream:                       # 流式：另起线程边跑边吐 chunk（略）
        ...
    baggage_ctx = baggage.set_baggage("crew_context", CrewContext(id=str(self.id), key=self.key))
    token = attach(baggage_ctx)           # 把 crew 身份挂到当前上下文（tracing 用）
    runtime_scope = crewai_event_bus._enter_runtime_scope()
    try:
        inputs = prepare_kickoff(self, inputs, input_files)   # ★备战（L03）
        if self.process == Process.sequential:
            result = self._run_sequential_process()           # 顺序（D20）
        elif self.process == Process.hierarchical:
            result = self._run_hierarchical_process()         # 层级（D21）
        else:
            raise NotImplementedError(f"The process '{self.process}' is not implemented yet.")
        for after_callback in self.after_kickoff_callbacks:   # after 回调改结果
            result = after_callback(result)
        result = self._post_kickoff(result)
        self.usage_metrics = self.calculate_usage_metrics()   # 统计 token
        return result
    except Exception as e:
        crewai_event_bus.emit(self, CrewKickoffFailedEvent(...))   # 失败发事件（D26）
        raise
    finally:
        self._drain_memory_writes()       # ★收尾（L07）
        clear_files(self.id)
        detach(token)
        crewai_event_bus._exit_runtime_scope(runtime_scope)

真正的"开机准备"在这（crews/utils.py:249，裁剪主干）：

# crews/utils.py:249
def prepare_kickoff(crew, inputs, input_files=None) -> dict | None:
    resuming = crew.checkpoint_kickoff_event_id is not None
    if not resuming and get_current_parent_id() is None:
        reset_emission_counter()      # 事件序号归零（D26）
        reset_last_event_id()

    normalized = dict(inputs) if inputs is not None else None
    for before_callback in crew.before_kickoff_callbacks:   # ① before 回调改 inputs
        if normalized is None: normalized = {}
        normalized = before_callback(normalized)

    if not resuming:
        started_event = CrewKickoffStartedEvent(crew_name=crew.name, inputs=normalized)
        crew._kickoff_event_id = started_event.event_id
        crewai_event_bus.emit(crew, started_event)          # ② 发"开始"事件

    crew._task_output_handler.reset()                       # ③ 清上次的任务产出存档
    if normalized is not None:
        crew._inputs = normalized
        crew._interpolate_inputs(normalized)                # ④ 把 {topic} 替换成真实值
    crew._set_tasks_callbacks()

    agents_to_setup = list(crew.agents)                     # ⑤ 建齐所有 agent（含 task 里的）
    for task in crew.tasks:
        if task.agent is not None and id(task.agent) not in seen_agent_ids:
            agents_to_setup.append(task.agent)
    setup_agents(crew, agents_to_setup, crew.embedder, crew.function_calling_llm, crew.step_callback)

    if crew.planning:                                       # ⑥ 开了规划就先规划（D23）
        crew._handle_crew_planning()
    return normalized

批量执行（crew.py:1061）：

# crew.py:1061
def kickoff_for_each(self, inputs: list[dict], input_files=None) -> list[CrewOutput | ...]:
    results = []
    total_usage_metrics = UsageMetrics()
    for input_data in inputs:                 # 对每一条输入
        crew = self.copy()                    # ★关键：复制一份全新的 crew
        output = crew.kickoff(inputs=input_data, input_files=input_files)
        if not self.stream and crew.usage_metrics:
            total_usage_metrics.add_usage_metrics(crew.usage_metrics)   # 累加 token 用量
        results.append(output)
    if not self.stream:
        self.usage_metrics = total_usage_metrics
    self._task_output_handler.reset()
    return results

最"偷懒"但实用的异步（crew.py:1097，非流式主干）：

# crew.py:1097
async def kickoff_async(self, inputs=None, input_files=None, from_checkpoint=None):
    restored = apply_checkpoint(self, from_checkpoint)
    if restored is not None:
        return await restored.kickoff_async(inputs=inputs, input_files=input_files)
    inputs = inputs or {}
    if self.stream:
        ...                       # 流式异步（略）
    return await asyncio.to_thread(self.kickoff, inputs, input_files)   # ★核心就这一句

为并发而生的原生异步（crew.py:1177，主干）：

# crew.py:1177
async def akickoff(self, inputs=None, input_files=None, from_checkpoint=None):
    """Native async kickoff method using async task execution throughout.
    Unlike kickoff_async which wraps sync kickoff in a thread, this method
    uses native async/await for all operations including task execution,
    memory operations, and knowledge queries."""
    ...
    try:
        inputs = prepare_kickoff(self, inputs, input_files)   # 同样的备战
        if self.process == Process.sequential:
            result = await self._arun_sequential_process()    # ★异步版流程
        elif self.process == Process.hierarchical:
            result = await self._arun_hierarchical_process()
        ...
    finally:
        self._drain_memory_writes()
        ...

# crew.py:1298  异步版顺序流程
async def _arun_sequential_process(self) -> CrewOutput:
    return await self._aexecute_tasks(self.tasks)

# crew.py:1151  批量原生异步（真并发）
async def kickoff_for_each_async(self, inputs, input_files=None):
    async def kickoff_fn(crew, input_data):
        return await crew.kickoff_async(inputs=input_data, input_files=input_files)
    return await run_for_each_async(self, inputs, kickoff_fn)

收尾里最有意思的是 _drain_memory_writes（crew.py:1848）：

# crew.py:1848
def _drain_memory_writes(self) -> None:
    """Block until all pending background memory saves have completed.
    ... Must run before CrewKickoffCompletedEvent is emitted: listeners
    (e.g. telemetry sessions) tear down on that event, and any
    MemorySaveCompletedEvent emitted after teardown is lost ..."""
    seen: set[int] = set()
    candidates = [self._memory, self.memory,
                  getattr(self.manager_agent, "memory", None),
                  *(getattr(agent, "memory", None) for agent in self.agents)]
    for mem in candidates:
        if mem is None or isinstance(mem, bool):
            continue
        backing = getattr(mem, "_memory", None) or mem
        if id(backing) in seen:      # 同一个底层 memory 只 drain 一次
            continue
        seen.add(id(backing))
        drain = getattr(backing, "drain_writes", None)
        if callable(drain):
            drain()                   # ★阻塞等这个记忆池的后台写入全部落盘