> ## Documentation Index
> Fetch the complete documentation index at: https://docs.crewai.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Checkpointing
> Salve automaticamente o estado de execução para que crews, flows e agentes possam retomar após falhas.
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.
Como o checkpointing funciona: eventos, armazenamento e herança.
Um passo a passo de 5 minutos: executar, interromper, retomar.
Receitas focadas em tarefas para fluxos comuns.
`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.
```python theme={null}
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,
)
```
```python theme={null}
result = crew.kickoff()
```
Pressione `Ctrl+C` após a primeira tarefa concluir. Em `./.checkpoints/`, um arquivo `_.json` é o checkpoint.
```python theme={null}
from crewai import CheckpointConfig
result = crew.kickoff(
from_checkpoint=CheckpointConfig(
restore_from="./.checkpoints/_.json",
),
)
```
A tarefa de pesquisa é pulada, o escritor executa contra a saída de pesquisa salva e a crew finaliza.
## Guias de uso
```python theme={null}
crew = Crew(agents=[...], tasks=[...], checkpoint=True)
```
Grava em `./.checkpoints/` em cada `task_completed`.
```python theme={null}
from crewai import Crew, CheckpointConfig
crew = Crew(
agents=[...],
tasks=[...],
checkpoint=CheckpointConfig(
location="./my_checkpoints",
on_events=["task_completed", "crew_kickoff_completed"],
max_checkpoints=5,
),
)
```
```python JsonProvider theme={null}
from crewai import Crew, CheckpointConfig
from crewai.state import JsonProvider
crew = Crew(
agents=[...],
tasks=[...],
checkpoint=CheckpointConfig(
location="./my_checkpoints",
provider=JsonProvider(),
max_checkpoints=5,
),
)
```
```python SqliteProvider theme={null}
from crewai import Crew, CheckpointConfig
from crewai.state import SqliteProvider
crew = Crew(
agents=[...],
tasks=[...],
checkpoint=CheckpointConfig(
location="./.checkpoints.db",
provider=SqliteProvider(),
max_checkpoints=50,
),
)
```
O SQLite ativa o modo journal WAL para leituras concorrentes. Prefira-o para checkpointing de alta frequência.
```python theme={null}
crew = Crew(
agents=[
Agent(role="Researcher", ...),
Agent(role="Writer", ..., checkpoint=False),
],
tasks=[...],
checkpoint=True,
)
```
`fork()` restaura um checkpoint sob uma nova linhagem para que a nova execução não colida com a original.
```python theme={null}
config = CheckpointConfig(restore_from="./my_checkpoints/.json")
crew = Crew.fork(config, branch="experiment-a")
result = crew.kickoff(inputs={"strategy": "aggressive"})
```
O label `branch` é opcional; um é gerado se omitido.
```python theme={null}
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, write_task, review_task],
checkpoint=CheckpointConfig(location="./crew_cp"),
)
```
Gatilho padrão: `task_completed`.
```python theme={null}
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()
```
```python theme={null}
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"}])
```
Registre um handler em qualquer evento e chame `state.checkpoint()`.
```python Sync theme={null}
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}")
```
```python Async theme={null}
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"Checkpoint salvo: {path}")
```
Um argumento `state` é fornecido automaticamente quando o handler recebe três parâmetros. Veja [Event Listeners](/pt-BR/concepts/event-listener) para o catálogo completo de eventos.
```bash theme={null}
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.
```bash theme={null}
crewai checkpoint list ./my_checkpoints
crewai checkpoint info ./my_checkpoints/.json
crewai checkpoint info ./.checkpoints.db
```
## Referência
### `CheckpointConfig`
Destino do armazenamento. Diretório para `JsonProvider`, caminho de arquivo de banco para `SqliteProvider`.
Tipos de evento que disparam um checkpoint. `CheckpointEventType` é um `Literal` — seu type checker autocompleta e rejeita valores não suportados. Veja [tipos de evento](#tipos-de-evento) para a lista completa.
Backend de armazenamento. `JsonProvider` ou `SqliteProvider`.
Máximo de checkpoints a reter. Os mais antigos são removidos após cada gravação.
Checkpoint a restaurar quando passado via `from_checkpoint`.
### Valores do campo `checkpoint`
Aceito por `Crew`, `Flow` e `Agent`.
Herda do pai.
Ativa com padrões.
Desativação explícita. Interrompe a herança.
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`.
* **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
Um arquivo por checkpoint, nomeado `_.json` dentro de `location`.
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 ` | Inicia a TUI em uma localização específica. |
| `crewai checkpoint list ` | Lista checkpoints. |
| `crewai checkpoint info ` | Inspeciona um arquivo de checkpoint ou a entrada mais recente em um banco SQLite. |