ConditionalTask 直接继承 Task，只多一个 condition（tasks/conditional_task.py:14、tasks/conditional_task.py:28）：

# tasks/conditional_task.py:14
class ConditionalTask(Task):
    """A task that can be conditionally executed based on the output of another task.

    Notes:
        - Cannot be the only task in your crew
        - Cannot be the first task since it needs context from the previous task
    """
    condition: SerializableCallable | None = Field(
        default=None,
        description="Function that determines whether the task should be executed "
                    "based on previous task output.",
    )

    def __init__(self, condition=None, **kwargs) -> None:
        super().__init__(**kwargs)      # 先按普通 Task 造好（继承全部字段/校验）
        self.condition = condition      # 再挂上开关函数

判断跑不跑的方法是 should_execute（tasks/conditional_task.py:41）——它把上一步产出喂给你的 condition 函数：

# tasks/conditional_task.py:41
def should_execute(self, context: TaskOutput) -> bool:
    """Determines whether the conditional task should be executed
       based on the provided context."""
    if self.condition is None:
        raise ValueError("No condition function set for conditional task")
    return bool(self.condition(context))    # ★把上一步产出丢进 condition，取布尔结果

关键设计：跳过不是让任务凭空消失，而是产出一个"空壳" TaskOutput（tasks/conditional_task.py:57）：

# tasks/conditional_task.py:57
def get_skipped_task_output(self) -> TaskOutput:
    """Generate a TaskOutput for when the conditional task is skipped."""
    return TaskOutput(
        description=self.description,
        raw="",                                       # ★空文本：表示"没产出"
        agent=self.agent.role if self.agent else "",
        output_format=OutputFormat.RAW,
    )

Crew 顺序循环里，碰到 ConditionalTask 会先走 _handle_conditional_task（crew.py:1590），它调 check_conditional_skip（crews/utils.py:186）：

# crew.py:1590
def _handle_conditional_task(self, task, task_outputs, futures, task_index, was_replayed):
    if futures:                                        # ★先把在跑的异步任务收割掉（D17）
        task_outputs.extend(self._process_async_tasks(futures, was_replayed))
        futures.clear()
    return check_conditional_skip(self, task, task_outputs, task_index, was_replayed)

# crews/utils.py:186
def check_conditional_skip(crew, task, task_outputs, task_index, was_replayed):
    previous_output = task_outputs[-1] if task_outputs else None    # ★只看"上一个"产出
    if previous_output is not None and not task.should_execute(previous_output):
        crew._logger.log("debug", f"Skipping conditional task: {task.description}", ...)
        skipped_task_output = task.get_skipped_task_output()         # 造空占位
        if not was_replayed:
            crew._store_execution_log(task, skipped_task_output, task_index)
        return skipped_task_output      # 返回空占位 = "跳过了"
    return None                         # 返回 None = "该执行，照常跑"

Crew 对 ConditionalTask 有三条 after 校验（构造 crew 时就检查）：

# crew.py:790 —— ① 不能当第一个任务
@model_validator(mode="after")
def validate_first_task(self) -> Crew:
    if self.tasks and isinstance(self.tasks[0], ConditionalTask):
        raise PydanticCustomError("invalid_first_task",
            "The first task cannot be a ConditionalTask.", {})
    return self

# crew.py:774 —— ② 不能是唯一/全部（至少要有一个非条件任务）
@model_validator(mode="after")
def validate_must_have_non_conditional_task(self) -> Crew:
    if not self.tasks: return self
    non_conditional_count = sum(
        1 for task in self.tasks if not isinstance(task, ConditionalTask))
    if non_conditional_count == 0:
        raise PydanticCustomError("only_conditional_tasks",
            "Crew must include at least one non-conditional task", {})
    return self

# crew.py:800 —— ③ 不能是异步
@model_validator(mode="after")
def validate_async_tasks_not_async(self) -> Crew:
    for task in self.tasks:
        if task.async_execution and isinstance(task, ConditionalTask):
            raise PydanticCustomError("invalid_async_conditional_task",
                "Conditional Task: {description}, cannot be executed asynchronously.",
                {"description": task.description})
    return self

🧠 今天你应该能回答

• ConditionalTask 比普通 Task 多了什么？为什么用子类而非加字段？
• should_execute 的参数是什么？拿它做什么判断？
• 跳过时为什么要造一个空 TaskOutput，而不是什么都不留？
• Crew 判断条件前为什么要先收割异步任务？
• 三条硬约束（不能第一个/不能全是/不能异步）背后的共同道理是什么？
• ConditionalTask 和 Flow 的 router 有何本质区别？

✋ 10 分钟动手

P=lib/crewai/src/crewai
# 1. ConditionalTask 全文（很短，通读）
sed -n '1,69p' $P/tasks/conditional_task.py

# 2. Crew 里的跳过决策
sed -n '1590,1605p' $P/crew.py
sed -n '186,217p' $P/crews/utils.py

# 3. 三条硬约束
sed -n '773,812p' $P/crew.py