Day 04 / 共 60 天 · 阶段1 入门与心智
Task 是什么:一张会自己执行的"派工单"
昨天认识了"员工"(Agent),今天认识"派工单"(Task)。Task 描述要交付什么、由谁交付、依赖哪些前置结果、输出成什么格式。它不只是数据——还带着执行方法:execute_sync / execute_async 会把自己交给 agent 去干,再把结果包成标准的 TaskOutput。源码在单文件 task.py(1463 行)。
📍 你在 60 天里的位置(阶段1:入门与心智 · 共 6 天)
D01 项目全景→
D02 装环境跑通→
D03 Agent→
D04 Task→
D05 Crew→
D06 kickoff 旅程→
阶段2 Agent 深入
💡 先用一个类比兜住今天
Task 就是一张派工单:description 写"这活是干嘛的";expected_output 写"交付验收标准长什么样"(这一栏极其重要,模型全靠它判断"我做完了没");agent 写"派给哪个员工";context 写"要先拿到哪几张单子的结果才能开工";output_pydantic/output_json 写"结果要以什么格式装订上交"。这张单子还很智能——你喊一声
execute_sync(),它自己就去找 agent 把活干了。L01
痛点:一句"帮我分析一下"根本没法验收
🤔 痛点你给模型说"帮我分析下科技行业",它可能回你三句话,也可能回三千字,格式还每次都不一样。做自动化流水线时,下一步没法稳定地消费上一步的输出:你不知道它给的是纯文本、还是能解析的 JSON、还是漏了关键字段。你需要把"要什么"和"验收标准"写死。
💡 本质:Task = 把"活 + 验收标准 + 交付格式"结构化Task 强制你写清
description(干什么)和 expected_output(长什么样才算完),可选地指定 output_pydantic/output_json(交付成结构化数据)。这样每一步的产出都可预期、可验收、可被下一步消费。Task 是一个 Pydantic 模型(task.py:114):# task.py:114
class Task(BaseModel):
# ...几十个字段,但只有两个是必填...
大白话Agent 是"人",Task 是"事"。把"事"写成一张有验收标准的派工单,团队协作才不会各说各话。
L02
两个必填:description 与 expected_output
Task 字段虽多,真正必填的只有两个(task.py:146):
# task.py:146
description: str = Field(description="Description of the actual task.")
expected_output: str = Field(
description="Clear definition of expected output for the task.")
description这活是什么。会成为送给模型的核心指令。可以带 {占位符},D02 讲过会被 inputs 插值替换。expected_output★验收标准。模型靠它判断"我该产出什么、到什么程度算完成"。写得越具体("一段带置信分的分析"),输出越稳。这是新手最爱偷懒、却最影响质量的字段。💡 为什么把 expected_output 单列为必填,而不是塞进 description?把"做什么"和"要什么样"分成两栏,一是逼你想清验收标准(很多失败的 agent 输出,根源就是没定义清楚"完成"长啥样);二是框架能把 expected_output 拼进 prompt 的固定位置,还能在有 guardrail/结构化输出时用它做校验参照。这跟 D03 把人设拆成"三件套"是同一个思路:用结构化的必填栏,替代一大坨自由文本。
📝 真实值
description="Analyze {sector} sector data for the past {timeframe}"、expected_output="Detailed market analysis with a confidence score"。插值后 description 变成 "Analyze tech sector data for the past 1W",加上 expected_output 一起送给模型。L03
agent 与 context:派给谁、依赖谁
两个关系字段决定了这张单子怎么融入团队(task.py:157):
# task.py:157
agent: Annotated[
BaseAgent | None,
BeforeValidator(_resolve_agent),
] = Field(description="Agent responsible for execution the task.", default=None)
# task.py:161
context: list[Task] | None | _NotSpecified = Field(
description="Other tasks that will have their output used as context for this task.",
default=NOT_SPECIFIED)
agent这张单子派给谁。可以为 None——那样必须放进一个支持自动分配的 Crew(比如 hierarchical 由 manager 派活)里执行,否则单独执行会报错(L07 详谈)。BeforeValidator(_resolve_agent)和 D03 的 llm 一样的套路:允许你传各种"agent 引用",校验阶段解析成真正的 agent 对象。context: list[Task]★任务依赖!把"任务 A"放进"任务 B 的 context",B 执行时就能拿到 A 的输出当上下文。这是把多任务串成有依赖的流水线的关键(阶段3 D16 深挖)。默认值 NOT_SPECIFIED★注意默认不是 None,而是一个特殊哨兵 NOT_SPECIFIED。这是为了区分"你没设 context(用默认:上文全部任务)"和"你显式设 context=None(我就是不要任何上下文)"两种意图——None 是有含义的值,不能当"未设置"。💡 设计取舍①:为什么 context 默认值用哨兵 NOT_SPECIFIED,而不是 None?
很多 API 用
None 表示"没传"。但这里 None 本身是一个合法且有意义的值——"我明确不要任何前置上下文"。如果用 None 当默认,框架就无法分辨"用户压根没管 context(应走默认行为:把前面任务的输出作为上下文)"和"用户特意要求无上下文"。所以源码引入第三态哨兵 NOT_SPECIFIED(类型里还专门写了 _NotSpecified)。代价是多一个内部类型概念,回报是能精确表达"未设置 / 设为空 / 设为具体值"三种语义——这是 Python API 设计里处理"可选且 None 有意义"参数的标准手法。L04
输出形态字段群:纯文本 / JSON / Pydantic / 文件
一批字段决定"结果长成什么样、存到哪"(task.py:165 起):
# task.py:165
async_execution: bool | None = Field(default=False, ...) # 是否异步执行(不阻塞后面)
output_json: type[BaseModel] | None = Field(default=None,..)# 结果解析成 JSON(给个模型类当模板)
output_pydantic: type[BaseModel] | None = ... # 结果解析成 Pydantic 对象
response_model: type[BaseModel] | None = ... # 用 provider 原生结构化输出
output_file: str | None = Field(default=None, ...) # 结果写到文件
create_directory: bool | None = Field(default=True, ...) # 文件目录不存在就自动建
output: TaskOutput | None = Field(default=None, ...) # ★执行后回填的结果对象
tools: list[BaseTool] | None = Field(default_factory=list,.)# 本任务限定可用的工具
| 字段 | 设了它会怎样 |
|---|---|
output_pydantic=MyModel | 结果被解析成 MyModel 实例,放进 TaskOutput.pydantic |
output_json=MyModel | 结果被解析成 dict,放进 TaskOutput.json_dict |
async_execution=True | 任务在后台线程跑,Crew 可以先派后面的活(阶段3 D17) |
output_file="report.md" | 结果同时写入文件,目录不存在自动创建 |
tools=[...] | 本任务只允许用这些工具,覆盖 agent 的默认工具集 |
context=[taskA] | (L03)拿 taskA 的输出当本任务上下文 |
大白话如果你要把结果喂给下一段程序,别用纯文本——设
output_pydantic 让它交付成结构化对象,下游 result["confidence"] 直接取值,稳。L05
Task 会自己执行:execute_sync → _execute_core
Task 不只是数据,它带执行方法。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 —— 核心执行(裁剪)
def _execute_core(self, agent, context, tools) -> TaskOutput:
agent = agent or self.agent
self.agent = 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 []
self.processed_by_agents.add(agent.role)
crewai_event_bus.emit(self, TaskStartedEvent(context=context, task=self)) # 发"任务开始"事件
result = agent.execute_task(task=self, context=context, tools=tools) # ★把活真正交给 agent
self._post_agent_execution(agent)
# ...根据 output_pydantic/output_json 决定 raw/pydantic/json_output(见 L06)...
agent = agent or self.agent优先用传入的 agent(Crew 调度时会传),否则用任务自带的 self.agent。if not agent: raise★边界:一个 task 最终没有任何 agent 可用 → 直接抛异常,明确告诉你"这活没人干"。emit(TaskStartedEvent)又见事件总线——任务开始也发事件,监听器/遥测据此记录。agent.execute_task(...)★交接棒!Task 把自己、上下文、工具打包,调用 D03 见过的 Agent.execute_task。真正的"思考+用工具"循环在 agent 一侧(阶段2)。start_time记录开始时间,配合 end_time 能算 execution_duration(:591)。图注:Task 负责"接活、检查、发事件、交接、收结果打包";真正的 LLM 循环在 agent 一侧。
L06
结果如何包成 TaskOutput
agent 返回结果后,_execute_core 根据你设的输出形态字段,决定怎么装配(task.py:798):
# task.py:798
if isinstance(result, BaseModel): # agent 直接返回了结构化对象
raw = result.model_dump_json()
if self.output_pydantic:
pydantic_output = result; json_output = None
elif self.output_json:
pydantic_output = None; json_output = result.model_dump()
else:
pydantic_output = None; json_output = None
elif not self._guardrails and not self._guardrail:
raw = result
pydantic_output, json_output = self._export_output(result) # 从文本里尝试解析结构化
else:
raw = result; pydantic_output, json_output = None, None
# task.py:816
task_output = TaskOutput(
name=self.name or self.description,
description=self.description,
expected_output=self.expected_output,
raw=raw,
pydantic=pydantic_output,
json_dict=json_output,
agent=agent.role,
output_format=self._get_output_format(),
messages=agent.last_messages)
三条分支结果可能是:① 已是结构化对象;② 纯文本且无护栏(尝试 _export_output 解析);③ 有 guardrail(先留原始,后续护栏处理,D14 讲)。raw / pydantic / json_dict★对应 D02 见过的 CrewOutput 三视图。TaskOutput 是单个任务的结果,CrewOutput 里的 tasks_output[] 就是一串 TaskOutput。agent=agent.role记录这个结果是哪个角色产出的,便于追溯。messages=agent.last_messages把 agent 这次对话的消息也存进来,调试/审计时能看到完整过程。💡 层级关系
Agent.execute_task 返回原始结果 → Task._execute_core 包成 TaskOutput → Crew 收集所有 TaskOutput 汇成 CrewOutput。三层各管一段,这就是 D06 要串起来的完整旅程。图注:单个任务的输出是 TaskOutput(含 raw/pydantic/json 多视图);整场 kickoff 的输出 CrewOutput 里装着它们的列表。
💡 设计取舍②:为什么让 Task 自己带执行方法(execute_sync),而不是让 Crew 统一去跑?
朴素做法是:Task 只是纯数据,所有执行逻辑都写在 Crew 里。CrewAI 却把
execute_sync/execute_async/_execute_core 放在 Task 自己身上。好处:① Task 能脱离 Crew 单独执行(task.execute_sync(agent=a) 直接就能跑,便于单元测试和调试一个任务);② 同步/异步、结果打包成 TaskOutput、发任务事件这些"和单个任务强相关"的逻辑,就近放在 Task 里最内聚,Crew 只需在主循环里调一下 task.execute_sync(...),不必关心一个任务内部怎么打包结果。代价是"执行"这件事分散在 Crew 和 Task 两处,读整条链要跳着看(D06 会把它串起来)。这是"高内聚——把和某对象强相关的行为放到该对象上"的体现。L07
边界 + 今日小结
⚠️ 边界:agent=None 的 Task 什么时候能跑、什么时候崩?
L05 里
if not agent: raise Exception("...has no agent assigned...")。这意味着:如果你单独 task.execute_sync() 一个没指定 agent 的任务,会直接崩。但同样一个 agent=None 的 task 放进 hierarchical 流程的 Crew 里却能跑——因为 Crew 会用 manager_agent 来分配执行者,调用时把 agent 传进来(回看 L05 agent = agent or self.agent 的第一个 agent 就是 Crew 传的)。结论:sequential 流程里每个 task 通常要显式给 agent;hierarchical 里可以留空交给 manager 派。搞混这点,是新手"任务没 agent 报错"的高频原因。D05/阶段4 会讲清两种流程的区别。👶 小白:我想让输出是干净的 JSON,设 output_json 还是 output_pydantic?
👨🏫 老师:想在 Python 里当对象用(有类型提示、字段校验)就 output_pydantic=你的Model;只想要一个 dict/JSON 就 output_json=你的Model(Model 当模板告诉它字段)。两者都会让 TaskOutput 带上结构化视图。还有更"原生"的 response_model,用模型厂商的结构化输出特性——阶段3 D15 专讲三者区别。
🧠 今天你应该能回答
- Task 哪两个字段必填?为什么分开写?(description + expected_output;逼你定义验收标准、便于框架拼装/校验)
context=[taskA]干嘛用?(把 A 的输出当本任务上下文,串成依赖流水线)- context 默认为什么是 NOT_SPECIFIED 而不是 None?(区分"未设置/设空/设值"三态,None 本身有意义)
execute_sync内部把活交给了谁?(agent.execute_task)- TaskOutput 和 CrewOutput 什么关系?(TaskOutput 是单任务结果,CrewOutput.tasks_output 是它们的列表)
- agent=None 的 task 什么时候能跑?(放进 hierarchical Crew 由 manager 派;单独执行会报错)
✋ 10 分钟动手
# 1. 看必填字段与关系字段
sed -n '144,213p' lib/crewai/src/crewai/task.py
# 2. 看执行入口与核心
sed -n '572,580p;762,826p' lib/crewai/src/crewai/task.py
# 3. 亲手让输出结构化
uv run python -c "
from pydantic import BaseModel
from crewai import Agent, Task, Crew
class Report(BaseModel):
summary: str
confidence: float
a=Agent(role='Analyst', goal='analyze', backstory='veteran', llm='gpt-4o')
t=Task(description='Analyze tech sector', expected_output='summary + confidence',
agent=a, output_pydantic=Report)
out=Crew(agents=[a], tasks=[t]).kickoff()
print(out.pydantic) # → Report(summary=..., confidence=...)
"
明日预告 · Day 05:员工(Agent)和派工单(Task)都有了,明天把它们编成"团队"。打开全仓最大的
crew.py(2400 行),看 Crew 模型的核心字段(agents/tasks/process/memory/manager_agent……)以及它如何按 process 把 task 依次派发。