Frameworks & Ecosystem

LangGraph Checkpointers and State Persistence

June 2026 · By MortalApps · 25 min read · ~4,969 words

TL;DR

LangGraph Checkpointers enable persistence of agentic workflow state to a database, preventing data loss and facilitating recovery.
They solve the problem of ephemeral agent execution by saving the graph's current state, including node outputs and intermediate variables.
In production, checkpointers are critical for long-running agents, human-in-the-loop workflows, and ensuring fault tolerance against system failures.
Using a PostgreSQL checkpointer allows agents to resume execution from any prior step, supports multi-tenancy, and provides robust state management.
Checkpointers are essential for debugging complex agent behaviors and implementing time-travel debugging capabilities.

Why Concepts How Architecture Code Takeaways FAQ Related

Why This Matters

Agentic systems, especially those orchestrating complex, multi-step workflows, require robust state management to ensure reliability and maintain continuity. Without state persistence, any interruption - application crash, network error, or planned shutdown - results in complete loss of progress, forcing the agent to restart from the beginning. This problem is particularly acute for long-running processes, human-in-the-loop interactions, or tasks involving external API calls with side effects.

LangGraph Checkpointers were invented to address this by providing a mechanism to serialize and store the entire graph state at critical junctures. This allows developers to build resilient agents that can pick up exactly where they left off, significantly improving user experience and operational efficiency. For instance, a multi-stage customer support agent can save its state after each user interaction, ensuring that a browser refresh or system restart doesn't erase the conversation history or task progress. The PostgreSQL checkpointer, a common choice, offers ACID compliance and scalability for production deployments.

Ignoring state persistence in production leads to unpredictable agent behavior, increased operational costs due to re-computation, and frustrated users. Imagine a financial agent performing complex analysis over hours; without checkpointers, a transient error means re-running the entire process, wasting compute resources and delaying critical outcomes. Checkpointers are indispensable when agents interact with external systems, where idempotency is crucial, or when auditability of agent steps is required. They are preferred over simple in-memory state for any agent intended for production use or requiring more than trivial execution time, contrasting with stateless functions suitable for simple, short-lived tasks.

Core Concepts

LangGraph checkpointers rely on several core concepts to manage and persist agent state:

Agent State: The complete data snapshot of an agent's execution at a given point. In LangGraph, this is typically a dictionary representing the current values of variables within the graph, including messages, tool outputs, and custom data.

Checkpointer: An abstract interface defining methods for saving, loading, and listing agent states. Implementations like SqliteSaver and PostgresSaver provide concrete persistence mechanisms.

Graph State: The specific data structure that LangGraph uses to represent the current context of an agent's run. It evolves with each node execution and is the primary object serialized by a checkpointer.

Thread ID (thread_id): A unique identifier for a specific agent conversation or execution thread. This ID is crucial for isolating state between different users or parallel agent runs and is used by the checkpointer to retrieve the correct history.

Checkpoint: A single, immutable record of the agent's graph state at a particular moment in time. Each checkpoint is associated with a thread_id and a checkpoint_id (timestamp).

Snapshot: The most recent checkpoint for a given thread_id. When an agent resumes, it typically loads the latest snapshot to continue execution.

Thread Timestamp (checkpoint_id): A timestamp associated with each checkpoint, indicating when that specific state was saved. This allows for retrieving not just the latest state, but also historical states for debugging or specific recovery scenarios.

Configurable: A LangChain/LangGraph mechanism that allows dynamic configuration of components, including checkpointers, based on runtime parameters like thread_id for multi-tenancy.

How It Works

LangGraph checkpointers operate by intercepting state changes within the graph and persisting them to a configured storage backend. This ensures that the agent's progress is saved incrementally, allowing for recovery and inspection.

1. Initialization and Configuration

An agent graph is instantiated with a CheckpointSaver implementation, such as PostgresSaver. This saver is configured with a database connection string (e.g., for PostgreSQL or SQLite). The graph's invoke or stream methods are then called with a config dictionary that *must* include a configurable key, containing a thread_id.

2. State Capture

As the agent graph executes, each node processes inputs and produces outputs, modifying the overall graph state. Before and after each node's execution, or at specific points within the graph, the checkpointer is invoked. It captures the current GraphState object, which includes all accumulated messages, tool calls, and custom variables.

3. State Serialization and Storage

The captured GraphState is serialized, typically into a JSON or pickle format, depending on the checkpointer implementation and configuration. This serialized state, along with the thread_id, a checkpoint_id (timestamp), and a checkpoint_id (usually a UUID), is then written to the configured database. For PostgresSaver, this involves inserting a new row into a checkpoints table.

4. Checkpoint Retrieval and Resumption

When an agent needs to resume execution, the invoke or stream method is called with the same thread_id. The checkpointer queries the database for the latest checkpoint associated with that thread_id. If found, the serialized state is deserialized and used to initialize the graph's internal state. The graph then continues execution from the point where the last checkpoint was saved.

5. Multi-Tenant Isolation

For multi-tenant applications, each tenant or user is assigned a unique thread_id. The checkpointer uses this ID to ensure that state for one tenant is strictly isolated from others. When querying for checkpoints, the thread_id acts as a primary filter, preventing cross-contamination of conversational history or agent progress.

6. Failure Handling and Recovery

If an agent execution fails mid-run, the last successfully saved checkpoint remains in the database. Upon restart, the agent can load this last valid state and re-attempt the failed step or a subsequent one, minimizing lost work. If the database itself becomes unavailable, the checkpointer will typically raise an exception, preventing further state changes until connectivity is restored. Robust error handling and retry mechanisms around database operations are crucial for production systems.

Architecture

The conceptual architecture for LangGraph with persistent state involves a client application, the LangGraph runtime, a CheckpointSaver component, and a relational database backend.

Client Application: Initiates agent runs and provides the thread_id for state management. This could be a web service, a batch job, or a user interface. It sends requests to the LangGraph runtime.

LangGraph Runtime: This is the core agent orchestration engine. It defines the graph structure (nodes, edges, state transitions). When invoke or stream is called, it manages the execution flow and the evolving GraphState.

CheckpointSaver: An integral part of the LangGraph runtime, acting as an abstraction layer for state persistence. It exposes methods to get, put, and list checkpoints. It receives the current GraphState from the LangGraph runtime, serializes it, and interacts with the database.

Data Flow (LangGraph to CheckpointSaver): The LangGraph runtime passes the current GraphState (a Python dictionary or Pydantic object) to the CheckpointSaver for persistence.

Relational Database (e.g., PostgreSQL, SQLite): The persistent storage layer. It stores serialized GraphState objects along with metadata like thread_id, checkpoint_id, and checkpoint_id in a dedicated table (e.g., checkpoints).

Data Flow (CheckpointSaver to Database): The CheckpointSaver executes SQL INSERT or UPDATE statements to store serialized state. It uses SELECT statements to retrieve state for resumption.

Execution starts when the Client Application invokes an agent with a thread_id. LangGraph either loads an existing state from the Database via the CheckpointSaver or starts a new one. As the graph progresses, the CheckpointSaver continuously updates the Database. Execution ends when the graph reaches a terminal state or an explicit stop condition, with the final state persisted in the Database.

Checkpointer Interface and Implementations

LangGraph's state persistence is built around the BaseCheckpointSaver abstract class. This interface defines the contract for any checkpointer implementation, primarily requiring get, put, and list methods. The get method retrieves a specific checkpoint, put saves a new one, and list enumerates available checkpoints for a given thread.

The PostgresSaver is the production-grade option for PostgreSQL deployments implementation, leveraging SQLAlchemy to interact with various relational databases. It manages a single table, typically named checkpoints, which stores the serialized state and metadata. The SqliteSaver is a specialized PostgresSaver pre-configured for SQLite, often used for local development or lightweight, single-process applications.

PostgreSQL Checkpointer Setup and Schema

To use a PostgreSQL checkpointer, you instantiate PostgresSaver with a PostgreSQL connection string. Ensure the psycopg2 or asyncpg driver is installed. The checkpointer automatically creates the checkpoints table if it doesn't exist. The schema typically includes:

thread_id (TEXT): The unique identifier for the agent's conversation or execution thread.
checkpoint_id (TEXT): A UUID for each specific checkpoint.
parent_checkpoint_id (TEXT): References the previous checkpoint in the thread, forming a linked list for history.
checkpoint_id (TEXT): Timestamp of the checkpoint, used for ordering and retrieving the latest state.
state (JSONB): The serialized GraphState object, stored as JSON for efficient querying and partial updates.
metadata (JSONB): Optional additional metadata about the checkpoint.

The state column, using PostgreSQL's JSONB type, is highly efficient for storing and querying semi-structured data, making it suitable for the dynamic nature of agent states. Indexing thread_id and checkpoint_id is crucial for performance.

CREATE TABLE IF NOT EXISTS checkpoints (
    thread_id TEXT NOT NULL,
    checkpoint_id TEXT NOT NULL,
    parent_checkpoint_id TEXT,
    checkpoint_id TEXT NOT NULL,
    state JSONB,
    metadata JSONB,
    PRIMARY KEY (thread_id, checkpoint_id)
);
CREATE INDEX IF NOT EXISTS ix_checkpoints_thread_id_checkpoint_id ON checkpoints (thread_id, checkpoint_id DESC);

SQLite Checkpointer for Local Development

For local development, testing, and simple single-user applications, the SqliteSaver offers a lightweight, file-based persistence solution. It requires no separate database server setup. You initialize it by providing a file path for the SQLite database. The schema is identical to the PostgreSQL version, but the state column uses a generic JSON type or BLOB depending on the SQLAlchemy dialect, which might have different performance characteristics than JSONB for complex queries.

Resuming Agent Execution from Checkpoints

Resuming an agent from a checkpoint involves passing the thread_id to the graph's invoke or stream method. The checkpointer automatically fetches the *latest* checkpoint for that thread_id and reconstructs the graph's state. To resume from a *specific* historical checkpoint, you can pass `{

Code Example

This example demonstrates a basic LangGraph agent with a SQLite checkpointer. It runs a simple two-node graph, saves its state, and then shows how to retrieve the latest state.

Python

import os
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver

# Ensure OPENAI_API_KEY is set in environment variables
if not os.environ.get("OPENAI_API_KEY"):
    raise ValueError("OPENAI_API_KEY environment variable not set.")

# Define the agent state
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    current_step: str

# Define the nodes
def call_llm(state: AgentState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    response = llm.invoke(state["messages"])
    return {"messages": [response], "current_step": "LLM_CALLED"}

def analyze_response(state: AgentState):
    # Simulate some analysis
    analysis_message = HumanMessage(content=f"Analysis complete for: {state['messages'][-1].content[:50]}...")
    return {"messages": [analysis_message], "current_step": "ANALYSIS_DONE"}

# Configure the SQLite checkpointer
# The database file will be created in the current directory
saver = SqliteSaver(conn=sqlite3.connect("langgraph_state.sqlite", check_same_thread=False))

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("llm", call_llm)
workflow.add_node("analyze", analyze_response)

workflow.set_entry_point("llm")
workflow.add_edge("llm", "analyze")
workflow.add_edge("analyze", END)

app = workflow.compile(checkpointer=saver)

# Define a thread ID for this conversation
thread_id = "user_123"
config = {"configurable": {"thread_id": thread_id}}

print(f"--- Running agent for thread_id: {thread_id} ---")

# First run: Start a new conversation
print("
Initial run:")
input_message_1 = HumanMessage(content="What is the capital of France?")
for s in app.stream({"messages": [input_message_1]}, config=config):
    print(s)

# Second run: Continue the conversation, state should be loaded automatically
print("
Resuming and adding a new message:")
input_message_2 = HumanMessage(content="And what about Germany?")
for s in app.stream({"messages": [input_message_2]}, config=config):
    print(s)

# Verify the latest state from the checkpointer
print("
Retrieving latest state from checkpointer:")
latest_state = saver.get(config)
if latest_state:
    print(f"Latest state messages count: {len(latest_state['channel_values']['messages'])}")
    print(f"Last message: {latest_state['channel_values']['messages'][-1].content}")
else:
    print("No state found for this thread ID.")

Expected Output

--- Running agent for thread_id: user_123 ---

Initial run:
{'llm': {'messages': [AIMessage(content='The capital of France is Paris.', response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 15, 'total_tokens': 22}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_611b687793', 'finish_reason': 'stop', 'logprobs': None}, id='run-c97d2645-31f8-4127-b08e-8a033668f44d-0')]}, 'current_step': 'LLM_CALLED'}
{'analyze': {'messages': [HumanMessage(content='Analysis complete for: The capital of France is Paris....')]}, 'current_step': 'ANALYSIS_DONE'}
{'__end__': {'messages': [HumanMessage(content='What is the capital of France?'), AIMessage(content='The capital of France is Paris.', response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 15, 'total_tokens': 22}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_611b687793', 'finish_reason': 'stop', 'logprobs': None}, id='run-c97d2645-31f8-4127-b08e-8a033668f44d-0'), HumanMessage(content='Analysis complete for: The capital of France is Paris....')], 'current_step': 'ANALYSIS_DONE'}}

Resuming and adding a new message:
{'llm': {'messages': [AIMessage(content='The capital of Germany is Berlin.', response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 30, 'total_tokens': 37}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_611b687793', 'finish_reason': 'stop', 'logprobs': None}, id='run-7d72223a-c322-4424-b152-9442078a6316-0')]}, 'current_step': 'LLM_CALLED'}
{'analyze': {'messages': [HumanMessage(content='Analysis complete for: The capital of Germany is Berlin....')]}, 'current_step': 'ANALYSIS_DONE'}
{'__end__': {'messages': [HumanMessage(content='What is the capital of France?'), AIMessage(content='The capital of France is Paris.', response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 15, 'total_tokens': 22}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_611b687793', 'finish_reason': 'stop', 'logprobs': None}, id='run-c97d2645-31f8-4127-b08e-8a033668f44d-0'), HumanMessage(content='Analysis complete for: The capital of France is Paris....'), HumanMessage(content='And what about Germany?'), AIMessage(content='The capital of Germany is Berlin.', response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 30, 'total_tokens': 37}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': 'fp_611b687793', 'finish_reason': 'stop', 'logprobs': None}, id='run-7d72223a-c322-4424-b152-9442078a6316-0'), HumanMessage(content='Analysis complete for: The capital of Germany is Berlin....')], 'current_step': 'ANALYSIS_DONE'}}

Retrieving latest state from checkpointer:
Latest state messages count: 6
Last message: Analysis complete for: The capital of Germany is Berlin....

This example demonstrates using a PostgreSQL checkpointer and resuming execution from a specific checkpoint using `checkpoint_id`. It requires a running PostgreSQL instance and `DATABASE_URL` environment variable.

Python

import os
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.sqlalchemy import PostgresSaver
from sqlalchemy import create_engine

# Ensure OPENAI_API_KEY and DATABASE_URL are set
if not os.environ.get("OPENAI_API_KEY"):
    raise ValueError("OPENAI_API_KEY environment variable not set.")
if not os.environ.get("DATABASE_URL"):
    # Example: "postgresql+psycopg2://user:password@host:port/database_name"
    # For local testing, you might use a temporary SQLite file if PostgreSQL is not available
    print("DATABASE_URL environment variable not set. Falling back to SQLite for demonstration.")
    db_url = "sqlite:///langgraph_postgres_demo.sqlite"
else:
    db_url = os.environ["DATABASE_URL"]

# Define the agent state
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    current_task: str

# Define the nodes
def plan_task(state: AgentState):
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"Given the messages: {state['messages']}. What is the next logical task? Respond concisely."
    response = llm.invoke(prompt)
    return {"messages": [AIMessage(content=response.content)], "current_task": response.content}

def execute_task(state: AgentState):
    # Simulate task execution based on current_task
    task_output = f"Executed: {state['current_task']}. Result: Success."
    return {"messages": [HumanMessage(content=task_output)], "current_task": ""}

# Configure the SQLAlchemy (PostgreSQL/SQLite) checkpointer
engine = create_engine(db_url)
saver = PostgresSaver(sync_api_connection=engine)

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("plan", plan_task)
workflow.add_node("execute", execute_task)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", END)

app = workflow.compile(checkpointer=saver)

# Define a thread ID for this conversation
thread_id = "project_alpha_manager"
config = {"configurable": {"thread_id": thread_id}}

print(f"--- Running agent for thread_id: {thread_id} ---")

# First run: Initial planning
print("
Initial planning run:")
initial_input = HumanMessage(content="Start planning for Q3 marketing campaign.")
first_run_output = []
for s in app.stream({"messages": [initial_input]}, config=config):
    first_run_output.append(s)
    print(s)

# Extract the timestamp of the first run's last checkpoint
# This is a simplified way; in production, you'd query the checkpointer directly
first_run_checkpoint_ts = None
if first_run_output:
    # The last state in the stream corresponds to the final checkpoint for this run
    # We need to get the actual checkpoint_id from the saved checkpoint
    checkpoints = saver.list(config)
    if checkpoints:
        # Find the latest checkpoint from the first run
        # This assumes the first run creates at least one checkpoint
        first_run_checkpoint_ts = max([cp['checkpoint_id'] for cp in checkpoints if cp['thread_id'] == thread_id])
        print(f"Captured checkpoint timestamp from first run: {first_run_checkpoint_ts}")

# Second run: Simulate a crash and resume from the first run's state
if first_run_checkpoint_ts:
    print(f"
Simulating crash and resuming from timestamp: {first_run_checkpoint_ts}")
    resume_config = {"configurable": {"thread_id": thread_id, "checkpoint_id": first_run_checkpoint_ts}}
    # Provide an empty message list, as we are resuming the state, not adding new input
    for s in app.stream({"messages": []}, config=resume_config):
        print(s)
else:
    print("Could not find a checkpoint timestamp to resume from.")

# Third run: Continue normally, adding a new message
print("
Continuing with a new input after potential resume:")
next_input = HumanMessage(content="Review budget allocations for the campaign.")
for s in app.stream({"messages": [next_input]}, config=config):
    print(s)

Expected Output

--- Running agent for thread_id: project_alpha_manager ---

Initial planning run:
{'plan': {'messages': [AIMessage(content='The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.')], 'current_task': 'The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.'}}
{'execute': {'messages': [HumanMessage(content='Executed: The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}
{'__end__': {'messages': [HumanMessage(content='Start planning for Q3 marketing campaign.'), AIMessage(content='The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.'), HumanMessage(content='Executed: The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}
Captured checkpoint timestamp from first run: 2024-07-30T12:00:00.000000Z (timestamp will vary)

Simulating crash and resuming from timestamp: 2024-07-30T12:00:00.000000Z (timestamp will vary)
{'plan': {'messages': [AIMessage(content='The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.')], 'current_task': 'The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.'}}
{'execute': {'messages': [HumanMessage(content='Executed: The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}
{'__end__': {'messages': [HumanMessage(content='Start planning for Q3 marketing campaign.'), AIMessage(content='The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.'), HumanMessage(content='Executed: The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}

Continuing with a new input after potential resume:
{'plan': {'messages': [AIMessage(content='The next logical task is to analyze the current budget and allocate resources effectively for the Q3 marketing campaign.')], 'current_task': 'The next logical task is to analyze the current budget and allocate resources effectively for the Q3 marketing campaign.'}}
{'execute': {'messages': [HumanMessage(content='Executed: The next logical task is to analyze the current budget and allocate resources effectively for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}
{'__end__': {'messages': [HumanMessage(content='Start planning for Q3 marketing campaign.'), AIMessage(content='The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.'), HumanMessage(content='Executed: The next logical task is to define the target audience and key messaging for the Q3 marketing campaign.. Result: Success.'), HumanMessage(content='Review budget allocations for the campaign.'), AIMessage(content='The next logical task is to analyze the current budget and allocate resources effectively for the Q3 marketing campaign.'), HumanMessage(content='Executed: The next logical task is to analyze the current budget and allocate resources effectively for the Q3 marketing campaign.. Result: Success.')], 'current_task': ''}}

Key Takeaways

LangGraph checkpointers provide fault tolerance and state continuity for agentic workflows by persisting graph state to a database.

The `PostgresSaver` supports robust relational databases like PostgreSQL for production deployments, while `SqliteSaver` is ideal for local development.

Unique `thread_id`s are essential for isolating agent conversations and enabling multi-tenant state management.

Agents can resume execution from the latest state or a specific historical checkpoint using `thread_id` and `checkpoint_id`.

Proper database indexing, security, and data retention policies are critical for scalable and secure checkpointer implementations.

State persistence is fundamental for long-running agents, human-in-the-loop scenarios, and debugging complex agent behaviors.

Frequently Asked Questions

What is the primary purpose of a LangGraph checkpointer? +

The primary purpose is to persist the state of an agent's execution, allowing it to recover from failures, resume long-running tasks, and maintain conversational context across sessions.

What is the difference between `SqliteSaver` and `PostgresSaver`? +

`SqliteSaver` uses a local file-based SQLite database, suitable for development. `PostgresSaver` uses SQLAlchemy to connect to various relational databases like PostgreSQL or MySQL, ideal for production.

When should I avoid using a checkpointer? +

Avoid checkpointers for purely stateless, short-lived agent tasks where re-running from scratch is trivial and acceptable. The overhead of state serialization and database I/O might not be justified.

How does multi-tenancy work with LangGraph checkpointers? +

Multi-tenancy is achieved by assigning a unique `thread_id` to each tenant. The checkpointer uses this ID to store and retrieve state, ensuring strict isolation of each tenant's agent history.

What happens if the database is unavailable during a checkpoint operation? +

If the database is unavailable, the checkpointer operation will typically raise an exception. The agent will not be able to save its current state, potentially leading to data loss if not handled with retry logic.

Can I resume an agent from a specific historical point, not just the latest state? +

Yes, by providing both the `thread_id` and the `checkpoint_id` (timestamp) of the desired checkpoint in the `config` dictionary, you can instruct the checkpointer to load a specific historical state.

What kind of data is stored in the `checkpoints` table? +

The table stores the serialized `GraphState` (typically JSON), along with metadata such as `thread_id`, `checkpoint_id`, `parent_checkpoint_id`, and `checkpoint_id`.

How can I manage the size of the checkpoint history in my database? +

Implement data retention policies and scheduled cleanup jobs to periodically prune old or irrelevant checkpoints from the `checkpoints` table based on age or other criteria.

Are there performance implications for using checkpointers? +

Yes, checkpointers introduce latency due to state serialization and database I/O. Frequent, large state saves can impact throughput. Proper indexing and database optimization are crucial for performance.