3.2 构造与编排 Tasks

3.2 构造与编排 Tasks

本节约处在第三章第二站，也是构建阶段最该花心思的地方。Agent 再强，Task 编排错了，整条链就喂错料。我们说：编排是把"业务动作"翻译成"有向任务图"的过程，图的边用 context 表达，图的顺序用列表顺序配合 Process 决定。

先把一个典型的"扇出—汇聚"任务

先把一个典型的"扇出—汇聚"任务图画出来，建立全局视图：

（注意：手写 SVG 时属性要对位，把 viewBox 写进 <svg> 开标签而非单独成行，否则图不会渲染。下面所有图都用规范写法。）

回到正题。最基础的串行编排，把"

回到正题。最基础的串行编排，把"研究→撰写"用 context 连起来：

from crewai import Agent, Task

researcher = Agent(role="研究员", goal="找事实", backstory="严谨", verbose=True)
writer = Agent(role="写手", goal="成文", backstory="连贯", verbose=True)

t_research = Task(
    description="搜集 {topic} 的三个关键事实",
    expected_output="三条事实，各带来源链接",
    agent=researcher,
)
t_write = Task(
    description="基于上述事实写一段说明",  # 不重复说搜集什么
    expected_output="两百字说明，先结论后论据",
    agent=writer,
    context=[t_research],  # 消费上一个任务产出
)
​

关键点：t_write 的 description 不重写"搜集"，它靠 context=[t_research] 拿到上游结果。框架在跑 t_write 前，会把 t_research 的产出拼进它的上下文。这就是顺序模式下数据通路的真相。

分支再汇聚，用 async_execution 让无依赖分支并行，最后靠汇聚任务的 context 等齐：

from crewai import Agent, Task, Crew, Process

researcher = Agent(role="研究员", goal="找事实", backstory="严谨", verbose=True)

# 三个互不依赖的采集任务，标异步并行
t_news = Task(description="搜 {topic} 新闻", expected_output="要点", agent=researcher, async_execution=True)
t_data = Task(description="查 {topic} 数据", expected_output="数字", agent=researcher, async_execution=True)
t_policy = Task(description="找 {topic} 政策", expected_output="要点", agent=researcher, async_execution=True)

# 汇聚任务依赖三个分支，框架等齐后启动
t_merge = Task(
    description="综合新闻、数据、政策写一份简报",
    expected_output="简报全文",
    agent=researcher,
    context=[t_news, t_data, t_policy],
)

crew = Crew(agents=[researcher], tasks=[t_news, t_data, t_policy, t_merge], process=Process.sequential)
​

这里有个反直觉处：虽然 process=sequential，但三个分支因为 async_execution=True 会在前序空档并行跑，只有 t_merge 严格等它们。总耗时从"三段串行"压到"最长一段 + 汇聚"。我们建议：只要任务间真无数据依赖，就大胆标异步，这是顺序模式里唯一的并行加速口。

占位符与 kickoff 的契约也要在这里写清。Task 描述里的 {topic} 必须在 crew.kickoff(inputs={"topic": ...}) 中出现，否则启动即报错：

# crew.kickoff(inputs={"topic": "储能", "year": "2025"})
​

编排时还有一个常见错误：把 agent 写在 Task 上，又在层级模式里期待经理重派——两者意志会打架。我们的规则是：顺序模式显式写 agent 最稳；层级模式可以省略 agent 让经理分派，但那样 context 的约束力也变弱。选了模式就别混用两种控制哲学。

收尾提醒：编排的本质是"画业务动

收尾提醒：编排的本质是"画业务动作的有向图"。动笔前先在一张纸上把"谁产出什么、被谁消费"列清楚，比直接写代码少返工三倍。下一站把这些 Task 和 Agent 装进 Crew 启动，你会看到图在 kickoff 一刻真正流动起来。

把任务串成一条数据链

Task 编排的核心是 context：上游 Task 的产出自动成为下游 Task 的输入。下面构造"调研→写稿→校对"三步链，注意每步的 expected_output 都钉死格式：

from crewai import Agent, Task

researcher = Agent(role="研究员", goal="找事实", backstory="严谨", verbose=True)
writer = Agent(role="写手", goal="写稿", backstory="清楚", verbose=True)
proof = Agent(role="校对", goal="挑错", backstory="较真", verbose=True)

t_research = Task(description="调研 {topic} 的三个事实",
                  expected_output="三条带出处的事实", agent=researcher)
t_write = Task(description="据事实写 600 字稿",
               expected_output="含标题与正文的稿件", agent=writer, context=[t_research])
t_proof = Task(description="核对稿件事实",
               expected_output="每条事实的可核实标注", agent=proof, context=[t_write])

inputs = {"topic": "储能"}
print(t_write.context[0].description)
print(inputs)
​

⚠️ 常见坑：context 传的是 Task 对象，不是字符串结果。有人写成 context=["一些文字"]，CrewAI 无法把它识别为上游依赖，链路就断了。永远传 Task 对象本身。

异步任务能提速：互不依赖的 Task 设 async_execution=True，Crew 会并发跑它们。

t_x = Task(description="并行任务X", expected_output="x", agent=writer, async_execution=True)
t_y = Task(description="并行任务Y", expected_output="y", agent=proof, async_execution=True)
print(t_x.async_execution, t_y.async_execution)
​

💡 关键直觉：编排 Task 的本质是在画一张有向无环图，节点是 Task，边是 context。你写的是图的形状，CrewAI 负责按图调度。图越清晰（依赖声明越准），系统越可预测。

# 验证整条链的依赖是否闭合：每个 context 都要有对应 Task
assert all(c in [t_research, t_write, t_proof] for c in t_write.context + t_proof.context)
​