4.5 Crew 的持久化与恢复

4.5 Crew 的持久化与恢复

本节约处在第四章最后一站。长任务最怕两件事：跑半小时崩在第 28 步，全重来；以及想复盘"上次到底怎么跑的"却什么都没留。持久化解决前者，恢复解决"从断点续跑"。本章前四节让系统有状态、有人、可观测，这一节让它"中断可续、过程可查"。

把持久化的两个层面画清楚，避免把

把持久化的两个层面画清楚，避免把"缓存"和"断点恢复"混为一谈：

第一层：中间产出落库。用 step_callback 把每步 output 写进你自己的存储（文件、数据库、对象存储都行），框架本身不替你管长期归档：

import json, os

STORE = "runs"

def persist_step(step_output):
    os.makedirs(STORE, exist_ok=True)
    task = getattr(step_output, "task", None)
    name = (task.description[:20] if task else "step")
    payload = {"task": name, "output": str(step_output)}
    # 用递增序号，保证顺序可回放
    idx = len(os.listdir(STORE))
    with open(f"{STORE}/{idx:03d}.json", "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False)

# crew = Crew(..., step_callback=persist_step)
​

这样即使本次运行崩了，磁盘上已经

这样即使本次运行崩了，磁盘上已经有前面各步的产出。复盘时按文件名顺序读，就能完整还原"它跑到哪、产出了什么"。

第二层：断点恢复。CrewAI 没有现成的"自动续跑"开关，但你可以用已落库的产出人工或半自动地重建后续 Crew。核心思路是：把最后好的一步产出，作为后续任务的 context 起点，跳过已完成的步骤。

from crewai import Agent, Task, Crew, Process

# 假设步骤A已跑完，产出存在 done_a.json
last_output = json.load(open("runs/000.json", encoding="utf-8"))["output"]

writer = Agent(role="写手", goal="成文", backstory="连贯", verbose=True)

# 不再重跑 A，直接从 B 开始，context 接上 A 的产出
t_b = Task(description="基于已有资料写报告", expected_output="报告", agent=writer,
           context=[])  # 用 inputs 直接喂 last_output
crew = Crew(agents=[writer], tasks=[t_b], process=Process.sequential)
# crew.kickoff(inputs={"material": last_output})
​

这里的关键是"把已完成的产出当外部输入注入"，而非指望框架记住。我们主张：恢复逻辑由你的编排层负责，框架只保证单步可重复——只要每步 expected_output 钉死、产出落了库，重建后续 Crew 就是机械活。

关于 cache：它确实能避免相同任务重算，但缓存是进程内/按输入命中的，不算持久化。进程一关、或换台机器，缓存就没了。所以别把 cache=True 当成恢复手段，它只省重复计算，不保中断续跑。

还有产出格式的稳定性直接影响恢复可行性：如果某步产出是"自由文本"，你重建后续 Crew 时喂不进去。所以 3.3 强调的"产出结构稳定"在这里显出价值——落库的东西能被程序可靠地再消费，恢复才成立。

一个工程建议：给每次运行一个 run_id，所有中间产出和最终 result 都挂在它下面。这样并发跑多个主题也不会串，复盘时能按 run_id 拉出完整时间线。

收尾提醒：持久化的本质是把"运行

收尾提醒：持久化的本质是把"运行态"变成"可查、可续的数据"。两层都要做——产出落库让你看得见，断点恢复让你崩得起。第四章到此讲完记忆、人在环、动态、可观测、持久化五项进阶能力。下一章我们不再加新零件，而是讲怎么把这些零件用出纪律：设计原则、测试、部署、安全。

把 Crew 的状态存下来，崩了能续

长任务最怕跑到一半进程挂掉、全重来。CrewAI 支持把 Crew 状态持久化，配合 CrewOutput 的保存与重放，实现断点续跑。下面演示把结果落盘、并在重启后读取：

from crewai import Agent, Task, Crew, Process

a = Agent(role="A", goal="g", backstory="b", verbose=True)
t = Task(description="长任务", expected_output="o", agent=a)
crew = Crew(agents=[a], tasks=[t], process=Process.sequential)

# 运行并保存产出
result = crew.kickoff()
crew_output = result  # CrewOutput 对象
with open("crew_state.json", "w", encoding="utf-8") as f:
    f.write(str(crew_output))   # 简化演示：实际可用 to_dict 持久化
print("已保存产出")
​

⚠️ 常见坑：只保存最终文本、不保存中间 Task 产出，恢复时无法判断卡在哪一步。我们坚持"每步产出都落盘"，恢复时从最后一个成功步骤之后继续，而不是从头再来。另一个坑是持久化文件无版本号，代码升级后旧格式读不出——加 schema_version 字段。

💡 关键直觉：持久化是给长链路买保险。多智能体任务动辄几十次 LLM 调用，任意一次外部故障都可能是"从头再来"的灾难。把状态当检查点，系统才具备"可恢复性"，这也是它能否上生产的分水岭。

# 恢复时先校验版本再读取，避免格式不兼容
schema_version = "v1"
assert schema_version == "v1", "状态格式不匹配，需迁移"
​

持久化的粒度选择

持久化到什么粒度，取决于任务多长。短任务（几步）只存最终产出即可；长任务（几十步、跨小时）要存每一步产出与当前指针，才能精确续跑。我们按"单次 kickoff 预计 LLM 调用次数"决定：超过 10 次就按步落盘，否则只存结果。

💡 关键直觉：持久化不是越细越好——每步都写盘有 IO 开销，且要处理并发写。粒度刚好覆盖"崩了能续、又不至于太慢"那条线，就是最优。