字段本身很朴素（task.py:165），就一个布尔开关，默认同步：

# task.py:165
async_execution: bool | None = Field(
    description="Whether the task should be executed asynchronously or not.",
    default=False,
)

真正的差异在执行入口——Task 有三个：同步 execute_sync（D13 讲过）、线程异步 execute_async、原生异步 aexecute_sync：

execute_async（task.py:596）的实现——它当场返回一个空的 Future（取餐号），真正的活丢给一个后台线程：

# task.py:596
def execute_async(self, agent=None, context=None, tools=None) -> Future[TaskOutput]:
    """Execute the task asynchronously."""
    future: Future[TaskOutput] = Future()          # ① 造一个空的"结果占位凭据"
    ctx = contextvars.copy_context()               # ② 拷贝当前上下文变量
    threading.Thread(
        daemon=True,                               # ③ 守护线程（主程序退出它也退）
        target=ctx.run,                            # ④ 在拷贝的上下文里运行
        args=(self._execute_task_async, agent, context, tools, future),
    ).start()                                      # ⑤ 立刻启动、立刻返回
    return future                                  # ⑥ 把取餐号交给调用方（此时活还没跑完）

后台线程实际跑的是 _execute_task_async（task.py:612）。它跑完把结果塞进 Future，出错就把异常塞进去：

# task.py:612
def _execute_task_async(self, agent, context, tools, future: Future[TaskOutput]) -> None:
    """Execute the task asynchronously with context handling."""
    try:
        self.start_time = datetime.datetime.now()
        result = self._execute_core(agent, context, tools)   # ★还是走同一个大厨房
        future.set_result(result)                            # 成功：结果装进 Future
    except Exception as e:
        future.set_exception(e)                              # 失败：异常也装进 Future

Crew 的顺序执行循环（crew.py:1543）里，异步任务和同步任务走不同分支。核心策略是"异步任务先攒着，遇到同步任务前必须先把攒的收割掉"：

# crew.py:1543（顺序执行循环，节选异步相关）
futures: list[tuple[Task, Future[TaskOutput], int]] = []
for task_index, task in enumerate(tasks):
    ...
    if task.async_execution:
        context = self._get_context(task, [last_sync_output] if last_sync_output else [])
        future = task.execute_async(agent=..., context=context, tools=...)
        futures.append((task, future, task_index))       # ① 异步：派出去，攒进 futures
    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=..., context=context, tools=...)  # ③ 同步：原地等
        task_outputs.append(task_output)
        ...
if futures:                                               # ④ 循环结束，收割剩余异步
    task_outputs.extend(self._process_async_tasks(futures, was_replayed))

"收割"就是 _process_async_tasks（crew.py:1918）——挨个 future.result() 阻塞等结果：

# 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 结尾最多只能有一个连续的异步任务。校验器 validate_end_with_at_most_one_async_task（crew.py:753）：

# crew.py:753
@model_validator(mode="after")
def validate_end_with_at_most_one_async_task(self) -> Self:
    """Validates that the crew ends with at most one asynchronous task."""
    final_async_task_count = 0
    for task in reversed(self.tasks):          # 从最后一个任务往前数
        if task.async_execution:
            final_async_task_count += 1
        else:
            break                              # 一碰到同步任务就停
    if final_async_task_count > 1:
        raise PydanticCustomError(
            "async_task_count",
            "The crew must end with at most one asynchronous task.", {})
    return self

除了"开线程"的 execute_async，Task 还有一条原生 async/await 的路：aexecute_sync（task.py:627）→ _aexecute_core（task.py:637）：

# task.py:627
async def aexecute_sync(self, agent=None, context=None, tools=None) -> TaskOutput:
    """Execute the task asynchronously using native async/await."""
    self.start_time = datetime.datetime.now()
    return await self._aexecute_core(agent, context, tools)

# task.py:637（_aexecute_core 骨架）
async def _aexecute_core(self, agent, context, tools) -> TaskOutput:
    ...
    result = await agent.aexecute_task(task=self, context=context, tools=tools)  # await 而非开线程
    ...
    pydantic_output, json_output = await self._aexport_output(result)  # 连解析也用 async 版
    ...

✋ 10 分钟动手

P=lib/crewai/src/crewai
# 1. 字段 + 三个执行入口
sed -n '165,168p;572,635p' $P/task.py

# 2. 后台线程真身
sed -n '612,626p' $P/task.py

# 3. Crew 攒批 + 收割
sed -n '1543,1588p' $P/crew.py
sed -n '1918,1931p' $P/crew.py

# 4. "最多一个异步收尾"校验
sed -n '753,771p' $P/crew.py