context 任务依赖:让第二个任务吃到第一个的产出
D13 埋了个点:context 默认不是 None,而是一个神秘的 NOT_SPECIFIED。今天把它讲透。真实的 crew 里任务总是环环相扣——"研究员搜集资料 → 作家写稿",作家得看到研究员的产出。context 就是声明"我这个任务依赖哪些任务的产出"。今天看它的三态语义、_get_context 如何把上游产出拼成上下文、Crew 怎么拦截非法依赖,源码在 task.py、crew.py、utilities/formatter.py。
痛点:任务之间怎么传数据?
context 字段声明它的"上游依赖"。执行到它时,框架把这些上游任务的产出汇总成一段文字,作为额外上下文塞进它的提示里。你只声明依赖关系,数据搬运框架来做。context 就是这张交接单的声明:写作工序在单子上写"我要研究工序和分析工序的交接单",框架就把那两份产出递过来。NOT_SPECIFIED:一个字段的三种状态
回看 context 字段声明(task.py:161),它的类型和默认值都很讲究:
# 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,
)
类型是 list[Task] | None | _NotSpecified——三选一,对应三种截然不同的语义:
None,那"用户没填"和"用户明确说不要上下文"就撞车了——框架无法知道到底该不该自动串联。引入 NOT_SPECIFIED 这个第三态后:NOT_SPECIFIED = 用户没管,我来默认串联;None = 用户明确要隔离,我一个都不给。语义清清楚楚。_NotSpecified:哨兵对象长什么样
NOT_SPECIFIED 是一个专门的哨兵对象(sentinel),定义在 utilities/constants.py(utilities/constants.py:46):
# utilities/constants.py:46
class _NotSpecified:
"""A sentinel type for a value that has not been specified.
- TODO: Consider moving this class and NOT_SPECIFIED to types.py
"""
def __repr__(self) -> str:
return "NOT_SPECIFIED"
...
# 之后(:77 附近)
NOT_SPECIFIED: Final[...] = _NotSpecified()
# "Unlike `None`, which might be a valid value from the user, `NOT_SPECIFIED`
# ..."(源码注释点明了它存在的理由)
class _NotSpecified一个专门的空类,唯一目的就是"造一个独一无二、不等于任何其它值的东西"。NOT_SPECIFIED = _NotSpecified()全局唯一实例(单例)。判断时用 is NOT_SPECIFIED(身份比较),绝不会和 None、空列表、False 混淆。__repr__ = "NOT_SPECIFIED"打印时显示成可读的名字,调试友好。None / "" / -1?
很多代码用 None 或魔法值(-1、"")表示"没设置"。但这些值本身可能是合法输入——用户可能真的想传 None(隔离)、空列表(依赖空集)。哨兵对象 NOT_SPECIFIED 是一个用户永远不会自己创建的独一无二的东西,所以"看到它 = 一定是默认值、用户没碰过"。代价是多定义一个类、判断要用 is 而非 ==;收益是把"未指定"这个状态和所有合法值彻底隔开。这是 Python 里表达"三态可选"的标准手法。_get_context:按三态分流
执行任务前,Crew 用 _get_context 算出要喂给这个任务的上下文字符串(crew.py:1827)——它正好按三态分三支:
# crew.py:1827
@staticmethod
def _get_context(task: Task, task_outputs: list[TaskOutput]) -> str:
if not task.context: # ① None 或 空列表 → 无上下文
return ""
return (
aggregate_raw_outputs_from_task_outputs(task_outputs) # ② NOT_SPECIFIED → 用"目前为止所有产出"
if task.context is NOT_SPECIFIED
else aggregate_raw_outputs_from_tasks(task.context) # ③ 指定列表 → 只用这些任务的产出
)
if not task.context处理 None(也顺带处理空列表):直接返回空串,这个任务没有任何上文。注意 NOT_SPECIFIED 是真值对象,不会走进这一支。task.context is NOT_SPECIFIED★用 is 身份判断(L03 说过)。默认态:把 task_outputs(到目前为止执行过的所有任务产出)全汇总当上下文——这就是"不写 context 就自动串联"的实现。else 分支用户给了明确列表:只汇总这几个任务的产出(aggregate_raw_outputs_from_tasks)。task_outputs = [研究产出, 分析产出]。• 写作没写 context(NOT_SPECIFIED)→ 上下文 = 研究产出 + 分析产出(全串)。
• 写作写了
context=[研究任务] → 上下文 = 仅研究产出(不含分析)。• 写作写了
context=None → 上下文 = 空(从零开始写)。拼接:多份产出用分隔线连成一段
"汇总"具体怎么做?看 utilities/formatter.py(utilities/formatter.py:16、utilities/formatter.py:29):
# utilities/formatter.py:13
DIVIDERS: Final[str] = "\n\n----------\n\n"
# utilities/formatter.py:16
def aggregate_raw_outputs_from_task_outputs(task_outputs: list[TaskOutput]) -> str:
"""Generate string context from the task outputs."""
return DIVIDERS.join(output.raw for output in task_outputs) # ★把各 raw 用分隔线连起来
# utilities/formatter.py:29
def aggregate_raw_outputs_from_tasks(tasks: list[Task] | _NotSpecified) -> str:
task_outputs = (
[task.output for task in tasks if task.output is not None] # 取每个任务已回填的 output
if isinstance(tasks, list)
else []
)
return aggregate_raw_outputs_from_task_outputs(task_outputs)
raw(文本面孔),用固定分隔线拼成一段,作为下游任务的上文喂进去。DIVIDERS.join(...)用固定分隔线 \n\n----------\n\n 把多份产出连起来——分隔线让 LLM 能分清"这是几份不同的材料",而不是糊成一团。只取 output.raw汇总的是文本面孔(raw),不是 pydantic 对象。因为上下文最终要变成 prompt 里的文字,文本才是通用媒介。task.output is not None只汇总已经执行完、有产出的任务(output 已回填);还没跑的任务没 output,跳过。边界:不许依赖"未来"的任务
依赖关系必须是"向后看"的——你只能依赖排在你前面的任务。Crew 有个校验器专门拦截"依赖未来任务"(crew.py:838):
# crew.py:838
@model_validator(mode="after")
def validate_context_no_future_tasks(self) -> Self:
"""Validates that a task's context does not include future tasks."""
task_indices = {id(task): i for i, task in enumerate(self.tasks)} # 每个任务的位置
for task in self.tasks:
if isinstance(task.context, list):
for context_task in task.context:
if id(context_task) not in task_indices:
continue
if task_indices[id(context_task)] > task_indices[id(task)]: # 依赖者在后
raise ValueError(
f"Task '{task.description}' has a context dependency "
f"on a future task '{context_task.description}', "
f"which is not allowed.")
return self
task_indices = {id(task): i}先建一张"任务 → 它在列表里第几位"的表(用 id() 对象身份当 key)。context_task 位置 > task 位置如果某任务依赖的上游,位置反而排在它后面——这是"还没产出就想用"的悖论,直接报错。Crew(...) 构造时(after 校验器)就查出并 raise ValueError,而不是等跑到写作任务、发现上文是空的才莫名其妙。又是 fail-fast:把"结构上不可能"的错误挡在执行之前。另外 validate_async_task_cannot_include_sequential_async_tasks_in_context(crew.py:814)还限制"异步任务的 context 里不能夹另一个未被同步任务隔开的异步任务"——这是 D17 异步的伏笔。copy:复制任务时如何"重连"依赖
Crew 常要深拷贝任务(比如 kickoff_for_each 多份输入各跑一遍)。复制时有个难题:拷贝出来的任务,它的 context 应该指向拷贝后的上游任务,而不是原来的。copy() 用 D13 讲的 key(内容指纹)来重连(task.py:1095):
# task.py:1095(copy 内部)
cloned_context = (
self.context
if self.context is NOT_SPECIFIED # 默认态:原样保留
else [task_mapping[context_task.key] for context_task in self.context] # ★按 key 查新任务
if isinstance(self.context, list)
else None # None:还是 None
)
is NOT_SPECIFIED → 原样默认态不涉及具体任务引用,直接保留(新任务照样"自动串联")。task_mapping[context_task.key]★核心:用上游任务的 key(内容指纹)去 task_mapping(旧 key → 新任务的映射)里查出对应的拷贝任务。这样依赖关系在拷贝后被正确"重新接线",不会误指回原任务。else Nonecontext 原本是 None(隔离),拷贝后还是 None。三态在拷贝时都被正确保留。key 在新旧任务之间建立对应,把依赖关系原样迁移。用 id 就不行(id 每个实例都不同,拷贝后对不上);用 key(内容相同则指纹相同)才能跨拷贝识别"这是同一个任务"。今日小结 + 动手
🧠 今天你应该能回答
context解决什么问题?它声明的是什么?- 为什么用
NOT_SPECIFIED而不是None当默认值?三态各是什么语义? - 哨兵对象为什么用
is比较而不是==? _get_context如何按三态分流?默认态汇总的是什么?- 上下文为什么用"拼 raw 文本 + 分隔线"而不是传结构化对象?
- "依赖未来任务"会怎样?为什么在构造时就报错?
- copy 复制任务时,靠什么把 context 依赖重新接线?(key)
✋ 10 分钟动手
P=lib/crewai/src/crewai
# 1. context 字段 + 哨兵定义
sed -n '161,164p' $P/task.py
sed -n '46,82p' $P/utilities/constants.py
# 2. _get_context 三态分流
sed -n '1827,1835p' $P/crew.py
# 3. 拼接实现
sed -n '13,46p' $P/utilities/formatter.py
# 4. 未来任务校验 + copy 重连
sed -n '838,854p' $P/crew.py
sed -n '1095,1101p' $P/task.py
execute_async 怎么开线程、Crew 怎么攒一批异步任务再 future.result() 收割、为什么"crew 最多以一个异步任务收尾",以及原生 async 的 aexecute_sync 那条路。