📚 编程进阶教程站

第 5 章 · Crew 组队与执行流程 ​

本章目标：掌握 Crew 对象的组装与执行 API（kickoff / kickoff_for_each / 异步变体），读懂 CrewOutput 的完整结构，学会用 inputs 模板变量实现批量任务和用 replay 从指定任务重跑。

5.1 组装一个规范的 Crew ​

python
import os
from crewai import Agent, Task, Crew, Process, LLM

deepseek = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7,
)

researcher = Agent(role="研究员", goal="产出结构化调研要点",
                   backstory="十年行业研究经验。", llm=deepseek)
writer = Agent(role="作者", goal="把要点扩写成流畅文章",
               backstory="资深科技专栏作者。", llm=deepseek)

t1 = Task(description="调研「{topic}」并总结 5 条要点。",
          expected_output="编号列表，每条不超过 30 字。", agent=researcher)
t2 = Task(description="基于调研写 300 字短文。",
          expected_output="Markdown 短文，含标题与三个小节。", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[t1, t2],
    process=Process.sequential,   # 顺序执行（默认）
    verbose=True,
)

组装规则：

  • agents 与 tasks 是两个列表；每个 task 通过 agent= 字段绑定执行者，未绑定的任务在启动时会报错；
  • process=Process.sequential 按任务定义顺序执行；分层模式 Process.hierarchical 在第 9 章展开；
  • output_log_file="crew_log.json" 可以把完整执行日志落盘（支持 .txt 与 .json），生产排障利器。

5.2 kickoff：单次执行与模板插值 ​

python
result = crew.kickoff(inputs={"topic": "大模型推理优化"})
print(result.raw)          # 最终任务（t2）的输出文本

inputs 字典会替换所有任务文本中的 {topic} 占位符。同一个 crew 定义，换一个 inputs 就能跑不同主题——"定义与数据分离" 是 Crew 工程化的第一步。

没有占位符时可以传 kickoff() 或 kickoff(inputs={})。

5.3 kickoff_for_each：批量执行 ​

要给 10 个主题各生成一篇短文？不需要循环里反复 kickoff，官方提供批量 API：

python
topics = [
    {"topic": "大模型推理优化"},
    {"topic": "多智能体协作"},
    {"topic": "RAG 检索增强"},
]

# 顺序批量：每个输入跑一遍完整 crew
results = crew.kickoff_for_each(inputs=topics)

for r in results:
    print("----")
    print(r.raw[:100])

高并发场景用原生异步版本（官方推荐用于高吞吐负载）：

python
import asyncio

async def main():
    results = await crew.akickoff_for_each(inputs=topics)   # 原生异步批量
    for r in results:
        print(r.raw[:80])

asyncio.run(main())

同理，单次执行也有异步版 await crew.akickoff(inputs={...})。

5.4 读懂 CrewOutput ​

kickoff 返回的 CrewOutput 是一个结构化对象：

python
result = crew.kickoff(inputs={"topic": "AI 安全"})

print(result.raw)            # 最终输出原文（默认格式）
print(result.tasks_output)   # 每个任务的 TaskOutput 列表
print(result.token_usage)    # token 用量统计

# 逐个任务回看中间产物——调试神器
for i, t_out in enumerate(result.tasks_output, 1):
    print(f"任务{i}: {t_out.raw[:60]}...")

# crew 对象上还有累计用量指标
print(crew.usage_metrics)
属性说明
raw最终输出字符串（最后一个任务的输出）
pydantic结构化 Pydantic 对象（需配置 output_pydantic，见第 12 章）
json_dictJSON 字典输出（需配置 output_json）
tasks_outputList[TaskOutput]，每个任务的原始/结构化输出
token_usage本次运行的 token 统计

str(result) 的取值优先级是 pydantic > json > raw。

5.5 replay：从指定任务重跑 ​

调试长流程时最痛的场景：第 5 个任务失败了，前 4 个任务的结果明明还在，难道全部重跑？CrewAI 的 replay 机制会把最近一次 kickoff 的任务输出保存在本地，允许你从指定任务继续：

bash
# CLI 方式：从指定 task_id 重跑（task_id 可从日志中获取）
crewai replay -t <task_id>

也可以在代码中调用 crew.replay(task_id=...)。适用场景：

  • 上游任务输出良好，只想重试失败的任务；
  • 调整了某个中间任务的提示词，想验证它对下游的影响而不必重跑更早的任务。

TIP

replay 依赖本地保存的最近一次运行记录，因此只对"刚刚跑过的 crew"有效。跨会话的断点恢复请使用 checkpoint 机制（第 22 章部署部分介绍）。

5.6 本章小结 ​

  • Crew = agents + tasks + process；每个 task 必须绑定 agent，sequential 按定义顺序执行；
  • kickoff(inputs={...}) 用模板插值实现"一套定义、多套数据"；
  • 批量用 kickoff_for_each，高并发用原生异步 akickoff / akickoff_for_each；
  • CrewOutput.raw / tasks_output / token_usage 是结果解读三件套，crew.usage_metrics 给累计账单；
  • crewai replay -t <task_id> 从指定任务重跑，长流程调试不必从头再来。

🧪 随堂测验

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

1. crew.kickoff() 返回的 result.raw 是什么？

2. 要给 50 个客户各跑一遍同一套 crew，最合适的 API 是？

3. 第 5 个任务失败但前 4 个结果良好，想从任务 5 继续执行，应该用？

4. 查看本次运行花了多少 token，正确的方式是？

🛠️ 动手实践 ​

  1. 把第 2 章的 crew 改造成批量版：准备 3 个不同主题的输入列表，用 kickoff_for_each 跑完并打印每篇短文的前 80 字。
  2. 编写代码遍历 result.tasks_output，把每个任务的输出分别保存为 output/task_1.md、output/task_2.md。
  3. 故意把最后一个任务的 expected_output 改成互相矛盾的要求制造失败，然后用 replay 从该任务重跑修复，记录 task_id 的获取方式。

单个 crew 已经玩转。下一章我们深入 LLM 配置，掌握三方模型接入的全部参数。