检查点 (Checkpointing)

检查点功能会在运行期间保存执行状态的快照，从而使 Crew、Flow 或 Agent 能够在失败后恢复，或者分叉到另一个分支中。

说明

检查点的工作原理：事件、存储和继承。

教程

5 分钟快速上手：运行、中断、恢复。

操作指南

针对常见工作流程的任务导向型配方。

参考

CheckpointConfig、事件、提供程序和 CLI。

说明

什么是检查点

检查点捕获了 CrewAI 重建运行状态所需的一切：Crew、Flow 或 Agent 的完整状态（配置、Agent 记忆和知识源、任务进度、中间输出、内部状态和属性），以及 kickoff 输入、截止到该点的事件历史，以及将检查点与所属运行关联的谱系 ID。恢复操作会重建该状态并继续执行。已完成的任务会被跳过，记忆和知识会被重新加载，下游工作将基于原始运行产生的相同输出继续进行。分叉 (Forking) 操作会在新的谱系下执行相同的还原，因此新分支和原始运行可以并行写入检查点而互不覆盖。

检查点何时写入

检查点是事件驱动的。运行时会订阅您通过 on_events 选择的事件，并在每次触发时写入检查点。默认的 task_completed 会为每个完成的任务生成一个检查点，这是粒度与磁盘使用之间的合理权衡。更高频率的事件（如 llm_call_completed）也可用于细粒度恢复，但会写入更多文件。

存储

CrewAI 自带两个提供程序

JsonProvider 每个检查点写入一个文件。易于阅读和检查。
SqliteProvider 写入单个 SQLite 数据库。更适合高频检查点。

当设置了 max_checkpoints 时，两者都会修剪最旧的检查点。

自动检查点写入（事件驱动）是尽力而为的：写入失败会被记录，运行将继续。手动调用 state.checkpoint() 和 state.acheckpoint() 会在失败时重新抛出异常。

继承模型

Crew、Flow 和 Agent 都接受 checkpoint 参数。子项会从父项继承，除非它们设置了自己的值或传入 False 以选择退出。在 Crew 上启用一次检查点后，每个 Agent 都会参与，您也可以选择排除某个特定的 Agent。

教程：恢复失败的 Crew

本教程大约需要 5 分钟。您将运行一个包含两个任务的 Crew，在中途将其停止，并从保存的检查点恢复。

启用检查点并创建 Crew

from crewai import Agent, Crew, Task

researcher = Agent(role="Researcher", goal="Research", backstory="Expert")
writer = Agent(role="Writer", goal="Write", backstory="Expert")

crew = Crew(
    agents=[researcher, writer],
    tasks=[
        Task(description="Research AI trends", agent=researcher, expected_output="bullets"),
        Task(description="Write a summary", agent=writer, expected_output="paragraph"),
    ],
    checkpoint=True,
)

运行并在此第一个任务后中断

result = crew.kickoff()

第一个任务完成后按下 Ctrl+C。查看 ./.checkpoints/ 目录 — 一个名为 <timestamp>_<uuid>.json 的文件就是检查点。

从检查点恢复

from crewai import CheckpointConfig

result = crew.kickoff(
    from_checkpoint=CheckpointConfig(
        restore_from="./.checkpoints/<timestamp>_<uuid>.json",
    ),
)

研究任务被跳过，写入任务将针对已保存的研究输出进行，最后 Crew 完成任务。

操作指南

使用默认值启用检查点

crew = Crew(agents=[...], tasks=[...], checkpoint=True)

在每次 task_completed 时写入 ./.checkpoints/。

自定义存储和频率

from crewai import Crew, CheckpointConfig

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=CheckpointConfig(
        location="./my_checkpoints",
        on_events=["task_completed", "crew_kickoff_completed"],
        max_checkpoints=5,
    ),
)

选择存储提供程序

from crewai import Crew, CheckpointConfig
from crewai.state import JsonProvider

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=CheckpointConfig(
        location="./my_checkpoints",
        provider=JsonProvider(),
        max_checkpoints=5,
    ),
)

SQLite 启用了 WAL 日志模式以支持并发读取。高频检查点建议优先使用此模式。

排除一个 Agent

crew = Crew(
    agents=[
        Agent(role="Researcher", ...),
        Agent(role="Writer", ..., checkpoint=False),
    ],
    tasks=[...],
    checkpoint=True,
)

分叉到新分支

fork() 会在全新的谱系下恢复检查点，因此新运行不会与原始运行冲突。

config = CheckpointConfig(restore_from="./my_checkpoints/<file>.json")
crew = Crew.fork(config, branch="experiment-a")
result = crew.kickoff(inputs={"strategy": "aggressive"})

branch 标签是可选的；如果省略，将自动生成一个。

为 Crew、Flow 或 Agent 设置检查点

Crew
Flow
Agent

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task, review_task],
    checkpoint=CheckpointConfig(location="./crew_cp"),
)

默认触发器：task_completed。

from crewai.flow.flow import Flow, start, listen
from crewai import CheckpointConfig

class MyFlow(Flow):
    @start()
    def step_one(self):
        return "data"

    @listen(step_one)
    def step_two(self, data):
        return process(data)

flow = MyFlow(
    checkpoint=CheckpointConfig(
        location="./flow_cp",
        on_events=["method_execution_finished"],
    ),
)
result = flow.kickoff()

agent = Agent(
    role="Researcher",
    goal="Research topics",
    backstory="Expert researcher",
    checkpoint=CheckpointConfig(
        location="./agent_cp",
        on_events=["lite_agent_execution_completed"],
    ),
)
result = agent.kickoff(messages=[{"role": "user", "content": "Research AI trends"}])

手动写入检查点

在任何事件上注册处理器并调用 state.checkpoint()。

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.llm_events import LLMCallCompletedEvent

if TYPE_CHECKING:
    from crewai.state.runtime import RuntimeState


@crewai_event_bus.on(LLMCallCompletedEvent)
def on_llm_done(source: Any, event: LLMCallCompletedEvent, state: RuntimeState) -> None:
    path = state.checkpoint("./my_checkpoints")
    print(f"Saved checkpoint: {path}")

当处理器包含三个参数时，state 参数会自动提供。有关完整的事件目录，请参阅事件监听器。

通过 CLI 浏览、恢复和分叉

crewai checkpoint
crewai checkpoint --location ./my_checkpoints
crewai checkpoint --location ./.checkpoints.db

左侧面板按分支分组检查点；分叉嵌套在父项下。选择检查点会打开详细信息面板，其中包含元数据、实体状态和任务进度。恢复 (Resume) 可继续运行；分叉 (Fork) 可启动一个新分支。

详细信息面板暴露了两个可编辑区域

输入 (Inputs) — 原始 kickoff 输入，已预填且可编辑。
任务输出 (Task outputs) — 已完成任务的输出。编辑输出并点击 Fork 会使下游任务失效，以便它们针对修改后的上下文重新运行。

对于“假设分析”非常有用：分叉、调整、观察。

无需 TUI 即可检查检查点

crewai checkpoint list ./my_checkpoints
crewai checkpoint info ./my_checkpoints/<file>.json
crewai checkpoint info ./.checkpoints.db

参考

`CheckpointConfig`

location

str

默认值:"\"./.checkpoints\""

存储目标。JsonProvider 为目录路径，SqliteProvider 为数据库文件路径。

on_events

list[CheckpointEventType | Literal["*"]]

默认值:"[\"task_completed\"]"

触发检查点的事件类型。CheckpointEventType 是一个 Literal — 您的类型检查器将自动补全并拒绝不支持的值。有关完整列表，请参阅事件类型。

provider

BaseProvider

默认值:"JsonProvider()"

存储后端。可以是 JsonProvider 或 SqliteProvider。

max_checkpoints

int | None

默认值:"None"

保留的最大检查点数。每次写入后都会修剪最旧的。

restore_from

Path | str | None

默认值:"None"

当通过 from_checkpoint 传递时，指定要从中恢复的检查点。

`checkpoint` 字段值

由 Crew、Flow 和 Agent 接受。

无

default

从父级继承。

True

bool

启用并使用默认配置。

False

bool

显式退出。停止继承。

CheckpointConfig(...)

CheckpointConfig

自定义配置。

事件类型

on_events 接受 CheckpointEventType 值的任意组合。默认的 ["task_completed"] 为每个完成的任务写入一个检查点；["*"] 匹配每个事件。

["*"] 和诸如 llm_call_completed 之类的高频事件会写入大量检查点，可能会降低性能。请将其与 max_checkpoints 配合使用。

显示所有支持的事件

任务 (Task) — task_started, task_completed, task_failed, task_evaluation
Crew — crew_kickoff_started, crew_kickoff_completed, crew_kickoff_failed, crew_train_started, crew_train_completed, crew_train_failed, crew_test_started, crew_test_completed, crew_test_failed, crew_test_result
Agent — agent_execution_started, agent_execution_completed, agent_execution_error, lite_agent_execution_started, lite_agent_execution_completed, lite_agent_execution_error, agent_evaluation_started, agent_evaluation_completed, agent_evaluation_failed
Flow — flow_created, flow_started, flow_finished, flow_paused, method_execution_started, method_execution_finished, method_execution_failed, method_execution_paused, human_feedback_requested, human_feedback_received, flow_input_requested, flow_input_received
LLM — llm_call_started, llm_call_completed, llm_call_failed, llm_stream_chunk, llm_thinking_chunk
LLM Guardrail — llm_guardrail_started, llm_guardrail_completed, llm_guardrail_failed
工具 (Tool) — tool_usage_started, tool_usage_finished, tool_usage_error, tool_validate_input_error, tool_selection_error, tool_execution_error
记忆 (Memory) — memory_save_started, memory_save_completed, memory_save_failed, memory_query_started, memory_query_completed, memory_query_failed, memory_retrieval_started, memory_retrieval_completed, memory_retrieval_failed
知识 (Knowledge) — knowledge_search_query_started, knowledge_search_query_completed, knowledge_query_started, knowledge_query_completed, knowledge_query_failed, knowledge_search_query_failed
推理 (Reasoning) — agent_reasoning_started, agent_reasoning_completed, agent_reasoning_failed
MCP — mcp_connection_started, mcp_connection_completed, mcp_connection_failed, mcp_tool_execution_started, mcp_tool_execution_completed, mcp_tool_execution_failed, mcp_config_fetch_failed
观察 (Observation) — step_observation_started, step_observation_completed, step_observation_failed, plan_refinement, plan_replan_triggered, goal_achieved_early
技能 (Skill) — skill_discovery_started, skill_discovery_completed, skill_loaded, skill_activated, skill_load_failed
日志 (Logging) — agent_logs_started, agent_logs_execution
A2A — a2a_delegation_started, a2a_delegation_completed, a2a_conversation_started, a2a_conversation_completed, a2a_message_sent, a2a_response_received, a2a_polling_started, a2a_polling_status, a2a_push_notification_registered, a2a_push_notification_received, a2a_push_notification_sent, a2a_push_notification_timeout, a2a_streaming_started, a2a_streaming_chunk, a2a_agent_card_fetched, a2a_authentication_failed, a2a_artifact_received, a2a_connection_error, a2a_server_task_started, a2a_server_task_completed, a2a_server_task_canceled, a2a_server_task_failed, a2a_parallel_delegation_started, a2a_parallel_delegation_completed, a2a_transport_negotiated, a2a_content_type_negotiated, a2a_context_created, a2a_context_expired, a2a_context_idle, a2a_context_completed, a2a_context_pruned
系统信号 — SIGTERM, SIGINT, SIGHUP, SIGTSTP, SIGCONT
通配符 — "*" 匹配所有事件。

存储提供程序

JsonProvider

provider

每个检查点一个文件，位于 location 内，命名为 <timestamp>_<uuid>.json。

SqliteProvider

provider

位于 location 的单个数据库文件，带有 WAL 日志记录。

命令行界面（CLI）

命令	用途
`crewai checkpoint`	启动 TUI；自动检测存储。
`crewai checkpoint --location <path>`	针对特定位置启动 TUI。
`crewai checkpoint list <path>`	列出检查点。
`crewai checkpoint info <path>`	检查检查点文件或 SQLite 数据库中的最新条目。

文档索引

说明

教程

操作指南

参考

​说明

​什么是检查点

​检查点何时写入

​存储

​继承模型

​教程：恢复失败的 Crew

​操作指南

​参考

​CheckpointConfig

​checkpoint 字段值

​事件类型

​存储提供程序

​命令行界面（CLI）

说明

什么是检查点

检查点何时写入

存储

继承模型

教程：恢复失败的 Crew

操作指南

参考

`CheckpointConfig`

`checkpoint` 字段值

事件类型

存储提供程序

命令行界面（CLI）