入口极简（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)

# crew.py:1479  对比：层级流程只是多了一步"造经理"（D21）
def _run_hierarchical_process(self) -> CrewOutput:
    self._create_manager_agent()
    return self._execute_tasks(self.tasks)

process.py 整个文件就 11 行（process.py:4）：

# process.py:1
from enum import Enum

class Process(str, Enum):
    """Class representing the different processes that can be used to tackle tasks"""
    sequential = "sequential"        # 顺序（今天）
    hierarchical = "hierarchical"    # 层级（D21）
    # TODO: consensual = 'consensual'   # 共识——还没实现

kickoff 里靠它二选一（crew.py:1023，节选）：

# crew.py:1023  在 kickoff 内部
if self.process == Process.sequential:
    result = self._run_sequential_process()
elif self.process == Process.hierarchical:
    result = self._run_hierarchical_process()
else:
    raise NotImplementedError(
        f"The process '{self.process}' is not implemented yet.")

核心引擎（crew.py:1519，裁剪到主干）：

# crew.py:1519
def _execute_tasks(self, tasks, start_index=0, was_replayed=False) -> CrewOutput:
    custom_start = self._get_execution_start_index(tasks)   # 断点续跑用（D24）
    if custom_start is not None:
        start_index = custom_start

    task_outputs: list[TaskOutput] = []
    futures: list[tuple[Task, Future[TaskOutput], int]] = []   # 攒异步任务
    last_sync_output: TaskOutput | None = None

    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):     # 条件任务：判断要不要跳（L06）
            skipped = self._handle_conditional_task(task, task_outputs, futures, task_index, was_replayed)
            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:                                        # 同步任务：先 join 之前的异步，再当场执行
            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)
            self._store_execution_log(task, task_output, task_index, was_replayed)

    if futures:                                      # 收尾：还有没 join 的异步任务
        task_outputs.extend(self._process_async_tasks(futures, was_replayed))
    return self._create_crew_output(task_outputs)

传接力棒的关键在 _get_context（crew.py:1827）：

# crew.py:1827
@staticmethod
def _get_context(task: Task, task_outputs: list[TaskOutput]) -> str:
    if not task.context:
        return ""                          # 任务没声明要 context → 空
    return (
        aggregate_raw_outputs_from_task_outputs(task_outputs)   # 默认：把之前所有产出拼起来
        if task.context is NOT_SPECIFIED
        else aggregate_raw_outputs_from_tasks(task.context)     # 显式指定：只拼指定的那几个任务
    )

异步任务的收割在 _process_async_tasks（crew.py:1918）：

# crew.py:1918
def _process_async_tasks(self, futures, was_replayed=False) -> list[TaskOutput]:
    task_outputs = []
    for future_task, future, task_index in futures:
        task_output = future.result()          # ★阻塞：等这个异步任务真正跑完
        task_outputs.append(task_output)
        self._process_task_result(future_task, task_output)
        self._store_execution_log(future_task, task_output, task_index, was_replayed)
    return task_outputs

回看主循环里的关键分流（crew.py:1558）：

# crew.py:1558
if task.async_execution:
    context = self._get_context(task, [last_sync_output] if last_sync_output else [])
    future = task.execute_async(...)      # 启动，不等，塞 futures
    futures.append((task, future, task_index))
else:
    if futures:                            # ★碰到同步任务前，先把攒的异步 join 掉
        task_outputs.extend(self._process_async_tasks(futures, was_replayed))
        futures.clear()
    ...                                    # 再执行这个同步任务

条件任务在主循环里被特判（crew.py:1590）：

# crew.py:1590
def _handle_conditional_task(self, task: ConditionalTask, task_outputs,
                             futures, task_index, was_replayed) -> TaskOutput | None:
    if futures:                                   # 先 join 掉攒着的异步——因为
        task_outputs.extend(self._process_async_tasks(futures, was_replayed))
        futures.clear()                           # 条件判断要看前面的真实产出
    return check_conditional_skip(
        self, task, task_outputs, task_index, was_replayed)

循环跑完，_create_crew_output 收尾（crew.py:1880，节选）：

# crew.py:1880
def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput:
    if not task_outputs:
        raise ValueError("No task outputs available to create crew output.")
    valid_outputs = [t for t in task_outputs if t.raw]       # 过滤掉空产出（被跳过的条件任务）
    if not valid_outputs:
        raise ValueError("No valid task outputs available to create crew output.")
    final_task_output = valid_outputs[-1]                    # ★最后一个有效产出 = crew 结果
    ...
    self._drain_memory_writes()          # 等后台记忆写完（D25）
    crewai_event_bus.emit(self, CrewKickoffCompletedEvent(...))   # 发"完成"事件（D26）
    return CrewOutput(
        raw=final_task_output.raw,
        pydantic=final_task_output.pydantic,
        json_dict=final_task_output.json_dict,
        tasks_output=task_outputs,       # 全部任务产出也一并带上
        token_usage=self.token_usage)

每个同步任务完成还会写执行日志（crew.py:1445）：

# crew.py:1445  _store_execution_log（供 replay 用，D24）
log = {
    "task": task,
    "output": {"description": ..., "raw": output.raw, "agent": output.agent, ...},
    "task_index": task_index,
    "inputs": inputs,
    "was_replayed": was_replayed,
}
self._task_output_handler.update(task_index, log)   # 存进磁盘，replay 时读回