Day 16 / 共 60 天 · 阶段3 Task 深入

context 任务依赖:让第二个任务吃到第一个的产出

D13 埋了个点:context 默认不是 None,而是一个神秘的 NOT_SPECIFIED。今天把它讲透。真实的 crew 里任务总是环环相扣——"研究员搜集资料 → 作家写稿",作家得看到研究员的产出。context 就是声明"我这个任务依赖哪些任务的产出"。今天看它的三态语义、_get_context 如何把上游产出拼成上下文、Crew 怎么拦截非法依赖,源码在 task.pycrew.pyutilities/formatter.py

📍 你在 60 天里的位置(阶段3:Task 深入 · 共 6 天)
阶段2 Agent D13 Task 模型 D14 输出与护栏 D15 结构化输出 D16 context 依赖 D17 异步任务 D18 条件任务 阶段4 Crew
L01

痛点:任务之间怎么传数据?

🤔 痛点Crew 里任务是一条链:研究 → 分析 → 写作。写作任务需要前面研究、分析的产出当素材。可 Task 之间没有共享变量,你怎么告诉"写作任务":请把研究员和分析师的结果当参考?手动复制粘贴产出显然不行。
💡 一句话本质:context = "我依赖哪些任务的产出"每个任务用 context 字段声明它的"上游依赖"。执行到它时,框架把这些上游任务的产出汇总成一段文字,作为额外上下文塞进它的提示里。你只声明依赖关系,数据搬运框架来做。
🔗 生活类比:工序交接单工厂流水线上,下一道工序开工前,要拿到上一道工序的"交接单"(半成品+说明)。context 就是这张交接单的声明:写作工序在单子上写"我要研究工序和分析工序的交接单",框架就把那两份产出递过来。
L02

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——三选一,对应三种截然不同的语义:

数据结构:context 的三种状态 = 三种依赖意图 NOT_SPECIFIED(默认) 你没写 context = "自动串联" 吃前面所有任务的产出 最常见,省心 None 你明确写 context=None = "我谁都不吃" 独立任务,无上文 刻意隔离 [taskA, taskB] 你给一个任务列表 = "只吃这几个" 精确指定依赖 精确控制
图注:正因为要区分"没写"和"明确写 None",才不能用 None 当默认值——于是引入第三种状态 NOT_SPECIFIED。
💡 关键:None 和"没写"必须能区分如果默认值是 None,那"用户没填"和"用户明确说不要上下文"就撞车了——框架无法知道到底该不该自动串联。引入 NOT_SPECIFIED 这个第三态后:NOT_SPECIFIED = 用户没管,我来默认串联;None = 用户明确要隔离,我一个都不给。语义清清楚楚。
L03

_NotSpecified:哨兵对象长什么样

NOT_SPECIFIED 是一个专门的哨兵对象(sentinel),定义在 utilities/constants.pyutilities/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 里表达"三态可选"的标准手法。
L04

_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 → 上下文 = 空(从零开始写)。
L05

拼接:多份产出用分隔线连成一段

"汇总"具体怎么做?看 utilities/formatter.pyutilities/formatter.py:16utilities/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 → 分隔线拼接 → 塞进下游 prompt 研究 TaskOutput .raw = "资料..." 分析 TaskOutput .raw = "结论..." DIVIDERS.join(raw...) "资料..." + 分隔线 + "结论..." context 字符串 → 作为写作任务的上文 分隔线 "\n\n----------\n\n" 让 LLM 能看清"这是两份不同来源的材料" 最终这段 context 通过 execute_sync(context=...) 传进 _execute_core → agent.execute_task
图注:只取每份产出的 raw(文本面孔),用固定分隔线拼成一段,作为下游任务的上文喂进去。
DIVIDERS.join(...)用固定分隔线 \n\n----------\n\n 把多份产出连起来——分隔线让 LLM 能分清"这是几份不同的材料",而不是糊成一团。
只取 output.raw汇总的是文本面孔(raw),不是 pydantic 对象。因为上下文最终要变成 prompt 里的文字,文本才是通用媒介。
task.output is not None只汇总已经执行完、有产出的任务(output 已回填);还没跑的任务没 output,跳过。
💡 设计取舍②:为什么上下文用"拼文本"而不是传结构化对象? 你可能觉得应该把上游的 pydantic 对象结构化地传给下游。但下游任务本质是给 LLM 的一段提示词——LLM 只吃文本。所以框架统一把上游产出降维成 raw 文本、用分隔线拼接。好处是简单通用(任何产出都能变文本、任何 LLM 都能读);代价是丢了结构(下游 LLM 得从文本里重新理解)。对"任务间传上下文"这个场景,文本拼接的简单性压倒了结构化的精确性——这与 Flow/state 那种需要精确结构的场景形成对比。
L06

边界:不许依赖"未来"的任务

依赖关系必须是"向后看"的——你只能依赖排在你前面的任务。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 时就报错 "写作任务依赖一个排在它后面的校对任务"——校对还没跑,哪来产出?这是逻辑上不可能满足的依赖。CrewAI 在 Crew(...) 构造时(after 校验器)就查出并 raise ValueError,而不是等跑到写作任务、发现上文是空的才莫名其妙。又是 fail-fast:把"结构上不可能"的错误挡在执行之前。另外 validate_async_task_cannot_include_sequential_async_tasks_in_contextcrew.py:814)还限制"异步任务的 context 里不能夹另一个未被同步任务隔开的异步任务"——这是 D17 异步的伏笔。
L07

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。三态在拷贝时都被正确保留。
💡 呼应 D13:这就是 key 的用武之地还记得 D13 说"id 是身份、key 是内容指纹,replay/复制时靠 key"吗?这里就是活例子:复制一批任务后,靠 key 在新旧任务之间建立对应,把依赖关系原样迁移。用 id 就不行(id 每个实例都不同,拷贝后对不上);用 key(内容相同则指纹相同)才能跨拷贝识别"这是同一个任务"。
L08

今日小结 + 动手

🧠 今天你应该能回答

  • 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
明日预告 · Day 17:到目前任务都是一个接一个跑。但有些任务互不依赖,能并行。明天讲 async_execution 异步任务execute_async 怎么开线程、Crew 怎么攒一批异步任务再 future.result() 收割、为什么"crew 最多以一个异步任务收尾",以及原生 async 的 aexecute_sync 那条路。
← Day 15 结构化输出 Day 17 · 异步任务 →