检查点

检查点会在运行期间保存执行状态的快照，这样 crew、flow 或 agent 就能在失败后恢复，或者分叉到另一条分支上继续执行。

说明

检查点如何工作：事件、存储与继承。

教程

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

操作指南

面向常见工作流的任务化配方。

参考

CheckpointConfig、事件、providers 和 CLI。

说明

什么是检查点

检查点会捕获 CrewAI 在运行中途重建所需的一切：crew、flow 或 agent 的完整状态 - 配置、agent memory 和 knowledge sources、任务进度、中间输出、内部状态和属性 - 以及 kickoff 输入、截至那一刻的事件历史，还有把检查点和来源运行关联起来的 lineage ID。

恢复时会重建这些状态并继续执行。已完成的任务会被跳过，memory 和 knowledge 会重新加载，后续工作会基于原始运行产生的相同输出继续。分叉会在新的 lineage 下执行同样的恢复，因此新分支和原始运行可以并行写入检查点，而不会互相覆盖。

检查点何时写入

检查点是事件驱动的。运行时会订阅你通过 on_events 选择的事件，并在每次事件触发时写入一个检查点。默认的 task_completed 会在每个任务完成后写入一个检查点 - 这是粒度和磁盘占用之间的合理折中。像 llm_call_completed 这类高频事件适合细粒度恢复，但会写出更多文件。

存储

CrewAI 提供两个 provider：

JsonProvider 每个检查点写一个文件。可读性好，也容易检查。
SqliteProvider 写入单个 SQLite 数据库。更适合高频检查点。

当设置了 max_checkpoints 时，这两个 provider 都会清理最旧的检查点。

继承模型

Crew、Flow 和 Agent 都接受 checkpoint 参数。子对象会继承父对象的设置，除非它们设置自己的值，或者传入 False 显式退出。你可以在 crew 上只启用一次检查点，所有 agents 都会参与；也可以选择性地排除某个 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",
    ),
)

research 任务会被跳过，writer 会基于已保存的 research 输出继续执行，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,
    ),
)

选择存储 provider

Code

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

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

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

crew = Crew(
    agents=[...],
    tasks=[...],
    checkpoint=CheckpointConfig(
        location="./.checkpoints.db",
        provider=SqliteProvider(),
        max_checkpoints=50,
    ),
)

排除某个 agent

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

分叉到新分支

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

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

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

默认触发器：task_completed。

Flow

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 = 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()。

Code

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}")

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)
async def on_llm_done_async(source: Any, event: LLMCallCompletedEvent, state: RuntimeState) -> None:
    path = await state.acheckpoint("./my_checkpoints")
    print(f"Saved checkpoint: {path}")

当处理器接受三个参数时，系统会自动传入 state 参数。有关完整事件目录，请参见 Event Listeners。

通过 CLI 浏览、恢复和分叉

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

Checkpoint TUI tree view — 检查点树 - 分支和分叉会嵌套在其父级下。

左侧面板会按分支分组检查点；分叉会嵌套在父级下。选择某个检查点会打开详情面板，显示元数据、实体状态和任务进度。Resume 会继续运行；Fork 会启动一个新分支。

Checkpoint detail overview tab — 概览选项卡 - 元数据、实体状态和运行摘要。

详情面板提供两个可编辑区域：

Inputs — 原始 kickoff 输入，已经预填并可编辑。
Task outputs — 已完成任务的输出。编辑某个输出并点击 Fork 会使下游任务失效，使它们基于修改后的上下文重新运行。

Fork confirmation panel — Fork 视图 - 确认从所选检查点创建新分支。

不使用 TUI 也能查看检查点

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

参考

`CheckpointConfig`

location str default: "./.checkpoints"

存储位置。对 JsonProvider 来说是目录，对 SqliteProvider 来说是数据库文件路径。

on_events list[CheckpointEventType | Literal["*"]] default: ["task_completed"]

触发检查点的事件类型。CheckpointEventType 是一个 Literal - 类型检查器会自动补全并拒绝不支持的值。完整列表见 event types。

provider BaseProvider default: JsonProvider()

存储后端。可选 JsonProvider 或 SqliteProvider。

max_checkpoints int | None default: None

保留的检查点最大数量。每次写入后会清理最旧的条目。

restore_from Path | str | None default: None

通过 from_checkpoint 传入时要恢复的检查点。

`checkpoint` 字段值

可被 Crew、Flow 和 Agent 接受。

None default

从父级继承。

True bool

使用默认值启用。

False bool

显式退出。停止继承。

CheckpointConfig(...) CheckpointConfig

自定义配置。

事件类型

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

所有支持的事件

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
System signals — SIGTERM, SIGINT, SIGHUP, SIGTSTP, SIGCONT
Wildcard — "*" 匹配所有事件。

存储 providers

JsonProvider provider

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

SqliteProvider provider

位于 location 的单个数据库文件，并启用 WAL journaling。

CLI

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

检查点

说明

教程

操作指南

参考

说明

什么是检查点

检查点何时写入

存储

继承模型

教程：恢复一个失败的 crew

操作指南

参考

CheckpointConfig

checkpoint 字段值

事件类型

存储 providers

CLI

`CheckpointConfig`

`checkpoint` 字段值