📚 编程进阶教程站

第 4 章 · Task 深入：任务定义与上下文 ​

本章目标：掌握 description 与 expected_output 的写法如何决定输出质量，学会用 context 构建任务依赖链、用 async_execution 并行加速、用 human_input 与 callback 增加人工与程序控制点。

4.1 Task 的核心属性全景 ​

如果说 Agent 是"员工"，Task 就是交给他的"工单"。官方属性表中，日常最常用的是这些：

属性类型说明
descriptionstr（必填）任务内容描述，支持 {占位符} 插值
expected_outputstr（必填）完成标准的详细描述，直接写入提示词
agentAgent执行该任务的 agent
contextList[Task]额外注入哪些任务的输出作为上下文
toolsList[BaseTool]本任务专属工具（覆盖 agent 默认工具集）
human_inputbool完成前是否需要人工确认
output_filestr把输出写入指定文件
callbackCallable任务完成后的回调函数
async_executionbool是否异步并行执行
output_pydantic / output_jsonBaseModel结构化输出（第 12 章详解）

注意 description 和 expected_output 是必填项——CrewAI 用它们构造提示词，缺一不可。

4.2 description 与 expected_output 的写法 ​

任务文本的质量 = 输出质量的上限。对比：

python
from crewai import Task

# ❌ 反例：描述含糊，标准缺失
vague_task = Task(
    description="分析一下这份销售数据。",
    expected_output="一份分析报告。",
    agent=analyst,
)

# ✅ 正例：背景 + 动作 + 约束 + 标准
precise_task = Task(
    description=(
        "分析附件中的 2024 年 Q3 销售数据（{region} 区域）。"
        "1) 计算环比增长率最高的三个品类；"
        "2) 找出退货率超过 8% 的SKU并列出可能原因；"
        "3) 所有结论必须给出数据出处（表格行号）。"
    ),
    expected_output=(
        "Markdown 报告，包含："
        "## 增长Top3（表格：品类/环比/依据行号）、"
        "## 高退货SKU（表格+原因假设）、"
        "## 建议行动（不超过3条）。"
        "总长度不超过 600 字。"
    ),
    agent=analyst,
)

三条实战原则：

  1. 把"怎么算完成"写进 expected_output：格式（表格/列表/Markdown）、长度、必须包含的段落；
  2. description 里给动作分解（1/2/3 步骤），模型对编号指令的遵循度远高于一段散文；
  3. 占位符插值：{region} 等变量在 kickoff(inputs={...}) 时替换，让任务定义可复用。

4.3 用 context 构建任务依赖链 ​

默认情况下，顺序流程中前一个任务的输出自动流入下一个任务。但当依赖关系不是"紧邻上一个"时，就要显式声明 context：

python
research_ai = Task(description="调研 AI Agent 赛道动态。", expected_output="5 条要点", agent=researcher)
research_ops = Task(description="调研运维自动化市场。", expected_output="5 条要点", agent=researcher)

# 汇总任务同时依赖两个上游任务
merge_task = Task(
    description="合并两份调研，输出统一的赛道综述。",
    expected_output="不超过 500 字的综述，分'共性趋势'与'差异机会'两节。",
    agent=writer,
    context=[research_ai, research_ops],   # 显式声明双依赖
)

context=[...] 的语义是：这些任务全部完成后，它们的输出会被拼接注入本任务的上下文。这是构建 DAG（有向无环图）式工作流的基础工具。

4.4 并行任务：async_execution ​

互相独立的上游任务可以并行执行，显著缩短总时长：

python
t1 = Task(
    description="调研 A 方向",
    expected_output="3 条要点",
    agent=researcher,
    async_execution=True,    # 异步执行，不阻塞主链
)
t2 = Task(
    description="调研 B 方向",
    expected_output="3 条要点",
    agent=researcher,
    async_execution=True,    # 与 t1 并行
)
summary = Task(
    description="汇总 A、B 两方向调研结果。",
    expected_output="200 字综述",
    agent=writer,
    context=[t1, t2],        # 汇总任务会等待两个异步任务完成
)

规则：异步任务由"下一个引用它的同步任务"触发等待。上面 summary 是同步任务且 context 引用了 t1、t2，所以它成为汇合点。若所有任务都异步且无人汇合，Crew 会在收尾时等待全部完成。

4.5 人工确认与回调控制点 ​

python
import json

def notify(result):
    """任务完成回调：拿到 TaskOutput 对象"""
    print(f"[回调] 任务完成，输出长度 {len(result.raw)} 字符")
    with open("task_log.jsonl", "a") as f:
        f.write(json.dumps({"raw": result.raw[:200]}, ensure_ascii=False) + "\n")

review_task = Task(
    description="起草给客户的道歉信。",
    expected_output="150 字以内的正式信函，语气诚恳，不承诺具体赔偿金额。",
    agent=writer,
    human_input=True,        # 输出后暂停，等人在终端确认或给修改意见
    callback=notify,         # 完成后执行回调（含人工修改后的最终版）
    output_file="output/apology_letter.md",  # 最终输出落盘（目录自动创建）
)
  • human_input=True：任务给出答案后在终端暂停，你可以输入修改意见让 agent 迭代，回车确认后继续。适合对外发布前的最后一道人工闸门；
  • callback：拿到 TaskOutput，适合做日志、指标上报、下游触发；
  • output_file：自动创建目录并写入，是"留档审计"的最简单实现。

4.6 本章小结 ​

  • description 与 expected_output 是必填项且直接进入提示词：动作分步骤编号、完成标准写清格式与长度；
  • 默认输出自动流向下一任务；非紧邻依赖用 context=[...] 显式声明，可构建 DAG；
  • async_execution=True 让独立任务并行，由引用它们的同步任务充当汇合点；
  • human_input 是人工闸门，callback 是程序钩子，output_file 负责留档。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Task 的哪两个属性是必填的？

2. 汇总任务需要等待两个互不相关的调研任务完成，正确做法是？

3. human_input=True 的行为是？

4. 关于 callback 参数，正确的说法是？

🛠️ 动手实践 ​

  1. 把第 2 章的 crew 改造为三路并行：同时调研"技术趋势/商业模式的竞品/用户社区反馈"，再汇总成简报。
  2. 给你的最终输出任务加上 human_input=True 和 output_file="output/report.md"，体验人工修改闭环。
  3. 写一个 callback：把每次任务的 token 用量（TaskOutput 中可获取）追加写入 usage.csv。

任务会写了。下一章我们把 agents 和 tasks 组装成 Crew，玩转批量执行与结果解读。