Task 就是一个 Pydantic BaseModel（task.py:114），所有字段都是声明：

# task.py:114
class Task(BaseModel):
    """Class that represents a task to be executed.

    Each task must have a description, an expected output and an
    agent responsible for execution.
    """
    __hash__ = object.__hash__
    ...
    description: str = Field(description="Description of the actual task.")
    expected_output: str = Field(
        description="Clear definition of expected output for the task."
    )
    agent: Annotated[BaseAgent | None, BeforeValidator(_resolve_agent)] = Field(
        description="Agent responsible for execution the task.", default=None
    )

Task 唯二强制要写的是 description 和 expected_output；agent 可以晚点由 Crew 指派（比如 hierarchical 流程里交给 manager）。这条底线由一个 after 校验器兜住（task.py:365）：

# task.py:365
@model_validator(mode="after")
def validate_required_fields(self) -> Self:
    if self.description is None:
        raise ValueError(
            "description must be provided either directly or through config"
        )
    if self.expected_output is None:
        raise ValueError(
            "expected_output must be provided either directly or through config"
        )
    return self

Task 的字段很多，但可以按"用途"分成几组。下面这张图把 task.py:141~298 里声明的字段做了归类——记住分组比记住每个字段更有用：

这里先看两个"依赖/执行"组的字段声明，注意它们的默认值很讲究（task.py:161、task.py:165）：

# task.py:161 —— context 默认不是 None，而是一个特殊哨兵 NOT_SPECIFIED
context: list[Task] | None | _NotSpecified = Field(
    description="Other tasks that will have their output used as context for this task.",
    default=NOT_SPECIFIED,
)
# task.py:165
async_execution: bool | None = Field(
    description="Whether the task should be executed asynchronously or not.",
    default=False,
)

每个 Task 有唯一 id，但它禁止用户设置——只能框架自动生成（task.py:222），并且有个 before 校验器专门拦截用户赋值（task.py:460）：

# task.py:222
id: uuid.UUID = Field(
    default_factory=uuid.uuid4,
    frozen=True,                       # ★冻结：造好后不能改
    description="Unique identifier for the object, not set by user.",
)

# task.py:460 —— 拦截"用户手动传 id"
@field_validator("id", mode="before")
@classmethod
def _deny_user_set_id(cls, v, info):
    if v and not (info.context or {}).get("from_checkpoint"):
        raise PydanticCustomError(
            "may_not_set_field", "This field is not to be set by the user.", {}
        )
    return v

那"两个任务是不是同一个任务"靠什么判断？靠 key ——它是内容指纹，用描述+期望输出算 md5（task.py:582）：

# task.py:582
@property
def key(self) -> str:
    description = self._original_description or self.description
    expected_output = self._original_expected_output or self.expected_output
    source = [description, expected_output]
    return md5("|".join(source).encode(), usedforsecurity=False).hexdigest()

你写 Task(...) 一行，Pydantic 会依次跑一串校验器（before 先、after 后）。它们像一条装配流水线，把松散的输入组装成合法的 Task。控制流如下：

其中最值得看的是 check_tools（工具兜底）和 check_output（互斥检查）（task.py:541、task.py:548）：

# task.py:541
@model_validator(mode="after")
def check_tools(self) -> Self:
    """Check if the tools are set."""
    if not self.tools and self.agent and self.agent.tools:
        self.tools = self.agent.tools        # 任务没配工具 → 继承 agent 的工具
    return self

# task.py:548
@model_validator(mode="after")
def check_output(self) -> Self:
    """Check if an output type is set."""
    output_types = [self.output_json, self.output_pydantic]
    if len([type for type in output_types if type]) > 1:
        raise PydanticCustomError(
            "output_type",
            "Only one output type can be set, either output_pydantic or output_json.",
            {},
        )
    return self

Task 最终要变成一段文字丢给模型。这个"拼装"发生在 prompt()（task.py:890）——它把 description、expected_output、markdown 指令等拼成一条提示：

# task.py:963（节选核心拼装）
tasks_slices = [description]
output = I18N_DEFAULT.slice("expected_output").format(
    expected_output=self.expected_output
)
tasks_slices = [description, output]

if self.markdown:
    markdown_instruction = """Your final answer MUST be formatted in Markdown syntax.
Follow these guidelines:
- Use # for headers
- Use ** for bold text
..."""
    tasks_slices.append(markdown_instruction)
return "\n".join(tasks_slices)

配方卡填好后，真正"开火"的入口是 execute_sync（task.py:572），它只记个开始时间，转手交给 _execute_core（task.py:762）这个大厨：

# task.py:572
def execute_sync(self, agent=None, context=None, tools=None) -> TaskOutput:
    """Execute the task synchronously."""
    self.start_time = datetime.datetime.now()
    return self._execute_core(agent, context, tools)

# task.py:762（_execute_core 骨架，删繁就简）
def _execute_core(self, agent, context, tools) -> TaskOutput:
    agent = agent or self.agent
    if not agent:
        raise Exception(f"The task '{self.description}' has no agent assigned...")
    self.prompt_context = context
    tools = tools or self.tools or []
    result = agent.execute_task(task=self, context=context, tools=tools)  # ★交给 Agent 干
    ...
    task_output = TaskOutput(name=..., raw=raw, pydantic=..., json_dict=..., ...)  # 包装结果
    if self._guardrails: ...   # 跑护栏（D14）
    self.output = task_output
    if self.callback: ...      # 完成回调
    if self.output_file: self._save_file(...)
    return task_output

✋ 10 分钟动手

P=lib/crewai/src/crewai
# 1. 通读 Task 全部字段声明
sed -n '141,300p' $P/task.py

# 2. 看校验器流水线（三要素/工具兜底/输出互斥）
sed -n '365,375p;541,558p' $P/task.py

# 3. 看 id 只读 + key 指纹
sed -n '222,226p;460,467p;582,588p' $P/task.py

# 4. 看执行入口
sed -n '572,580p' $P/task.py