Checkpointing - CrewAI

O checkpointing salva um snapshot do estado de execução durante uma execução para que uma crew, flow ou agente possa retomar após uma falha ou ser bifurcado em uma branch alternativa.

Explicação

Como o checkpointing funciona: eventos, armazenamento e herança.

Tutorial

Um passo a passo de 5 minutos: executar, interromper, retomar.

Guias de uso

Receitas focadas em tarefas para fluxos comuns.

Referência

CheckpointConfig, eventos, provedores e CLI.

Explicação

O que é um checkpoint

Um checkpoint captura tudo o que o CrewAI precisa para recriar uma execução em andamento: o estado completo da crew, flow ou agente — configuração, memória e fontes de conhecimento dos agentes, progresso das tarefas, saídas intermediárias, estado interno e atributos — junto com os inputs do kickoff, o histórico de eventos até aquele ponto e um ID de linhagem que liga o checkpoint à execução de origem. Restaurar reconstrói esse estado e continua. Tarefas concluídas são puladas, memória e conhecimento são reidratados, e o trabalho downstream roda contra as mesmas saídas que a execução original produziu. Fazer fork executa a mesma restauração sob uma nova linhagem, para que a nova branch e a execução original gravem checkpoints lado a lado sem sobrescrever uma a outra.

Quando os checkpoints são gravados

O checkpointing é orientado a eventos. O runtime se inscreve nos eventos selecionados em on_events e grava um checkpoint sempre que um é disparado. O padrão task_completed produz um checkpoint por tarefa finalizada — um equilíbrio razoável entre granularidade e uso de disco. Eventos de alta frequência como llm_call_completed estão disponíveis para recuperação mais granular, mas gravam muito mais arquivos.

Armazenamento

Dois provedores acompanham o CrewAI:

JsonProvider grava um arquivo por checkpoint. Legível e fácil de inspecionar.
SqliteProvider grava em um único banco SQLite. Melhor para checkpointing de alta frequência.

Ambos removem os checkpoints mais antigos quando max_checkpoints está definido.

Gravações de checkpoint automáticas (acionadas por evento) são best-effort: uma falha é registrada em log e a execução continua. Chamadas manuais a state.checkpoint() e state.acheckpoint() relançam a exceção.

Modelo de herança

Crew, Flow e Agent aceitam um argumento checkpoint. Filhos herdam do pai a menos que definam seu próprio valor ou passem False para desativar. Ative o checkpointing uma vez na crew e todos os agentes participam, ou exclua um agente seletivamente.

Tutorial: Retomar uma crew com falha

Este passo a passo leva cerca de 5 minutos. Você executará uma crew de duas tarefas, a interromperá no meio e a retomará a partir do checkpoint salvo.

Crie a crew com checkpointing ativado

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,
)

Execute e interrompa após a primeira tarefa

result = crew.kickoff()

Pressione Ctrl+C após a primeira tarefa concluir. Em ./.checkpoints/, um arquivo <timestamp>_<uuid>.json é o checkpoint.

Retome a partir do checkpoint

from crewai import CheckpointConfig

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

A tarefa de pesquisa é pulada, o escritor executa contra a saída de pesquisa salva e a crew finaliza.

Guias de uso

Ativar checkpointing com padrões

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

Grava em ./.checkpoints/ em cada task_completed.

Personalizar armazenamento e frequência

from crewai import Crew, CheckpointConfig

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

Escolher um provedor de armazenamento

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

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

O SQLite ativa o modo journal WAL para leituras concorrentes. Prefira-o para checkpointing de alta frequência.

Desativar um agente específico

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

Fazer fork em uma nova branch

fork() restaura um checkpoint sob uma nova linhagem para que a nova execução não colida com a original.

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

O label branch é opcional; um é gerado se omitido.

Checkpoint em Crew, Flow ou Agent

Crew
Flow
Agent

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

Gatilho padrão: 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"}])

Gravar um checkpoint manualmente

Registre um handler em qualquer evento e chame 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"Checkpoint salvo: {path}")

Um argumento state é fornecido automaticamente quando o handler recebe três parâmetros. Veja Event Listeners para o catálogo completo de eventos.

Navegar, retomar e fazer fork pela CLI

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

O painel esquerdo agrupa checkpoints por branch; forks aninham sob seu pai. Selecionar um checkpoint abre o painel de detalhes com metadados, estado da entidade e progresso das tarefas. Resume continua a execução; Fork inicia uma nova branch.

O painel de detalhes expõe duas áreas editáveis:

Inputs — os inputs originais do kickoff, preenchidos e editáveis.
Saídas das tarefas — saídas das tarefas concluídas. Editar uma saída e pressionar Fork invalida tarefas downstream para que sejam reexecutadas com o contexto modificado.

Útil para exploração de cenários: fork, ajuste, observe.

Inspecionar checkpoints sem a TUI

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

Referência

`CheckpointConfig`

location

str

padrão:"\"./.checkpoints\""

Destino do armazenamento. Diretório para JsonProvider, caminho de arquivo de banco para SqliteProvider.

on_events

list[CheckpointEventType | Literal["*"]]

padrão:"[\"task_completed\"]"

Tipos de evento que disparam um checkpoint. CheckpointEventType é um Literal — seu type checker autocompleta e rejeita valores não suportados. Veja tipos de evento para a lista completa.

provider

BaseProvider

padrão:"JsonProvider()"

Backend de armazenamento. JsonProvider ou SqliteProvider.

max_checkpoints

int | None

padrão:"None"

Máximo de checkpoints a reter. Os mais antigos são removidos após cada gravação.

restore_from

Path | str | None

padrão:"None"

Checkpoint a restaurar quando passado via from_checkpoint.

Valores do campo `checkpoint`

Aceito por Crew, Flow e Agent.

None

padrão

Herda do pai.

True

bool

Ativa com padrões.

False

bool

Desativação explícita. Interrompe a herança.

CheckpointConfig(...)

CheckpointConfig

Configuração personalizada.

Tipos de evento

on_events aceita qualquer combinação de valores CheckpointEventType. O padrão ["task_completed"] grava um checkpoint por tarefa finalizada; ["*"] corresponde a todos os eventos.

["*"] e eventos de alta frequência como llm_call_completed gravam muitos checkpoints e podem degradar o desempenho. Combine com max_checkpoints.

Mostrar Todos os eventos suportados

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
Sinais de sistema — SIGTERM, SIGINT, SIGHUP, SIGTSTP, SIGCONT
Wildcard — "*" corresponde a todos os eventos.

Provedores de armazenamento

JsonProvider

provider

Um arquivo por checkpoint, nomeado <timestamp>_<uuid>.json dentro de location.

SqliteProvider

provider

Arquivo de banco único em location com journaling WAL.

CLI

Comando	Propósito
`crewai checkpoint`	Inicia a TUI; detecta o armazenamento automaticamente.
`crewai checkpoint --location <path>`	Inicia a TUI em uma localização específica.
`crewai checkpoint list <path>`	Lista checkpoints.
`crewai checkpoint info <path>`	Inspeciona um arquivo de checkpoint ou a entrada mais recente em um banco SQLite.

Explicação

Tutorial

Guias de uso

Referência

​Explicação

​O que é um checkpoint

​Quando os checkpoints são gravados

​Armazenamento

​Modelo de herança

​Tutorial: Retomar uma crew com falha

​Guias de uso

​Referência

​CheckpointConfig

​Valores do campo checkpoint

​Tipos de evento

​Provedores de armazenamento

​CLI

Explicação

O que é um checkpoint

Quando os checkpoints são gravados

Armazenamento

Modelo de herança

Tutorial: Retomar uma crew com falha

Guias de uso

Referência

`CheckpointConfig`

Valores do campo `checkpoint`

Tipos de evento

Provedores de armazenamento

CLI