LangGraph DynamoDB Checkpoint Saver

A LangGraph checkpointer for Amazon DynamoDB with two things a plain checkpointer does not have: bounded message history and a long-term memory hook, both inside the save path, with no change to your graph.

• Bounded history. Every thread's message list is pruned before each checkpoint is written, by message count or token budget, whole messages only, tool-call pairs kept intact. The sawtooth above is the checkpoint size over time. Without a reducer it is a straight line up.
• Long-term memory. The turns that leave the window are handed to the reducer's on_prune hook, once each, together with the memory_namespace your app put in the run config. Wire that hook to any store or extraction engine; langgraph-memory is the ready-made one.
• One line of config.

from agentstate_reducer import MessageReducer, ReducerConfig, Background
from langgraph_dynamodb_checkpoint import DynamoDBSaver

reducer = MessageReducer(config=ReducerConfig(max_messages=20))            # add on_prune=[Background(engine.on_prune)] for memory
saver = DynamoDBSaver("checkpoints", reducer=reducer)
graph = builder.compile(checkpointer=saver)

graph.invoke(input, config={"configurable": {"thread_id": thread_id, "memory_namespace": ("memories", user_id)}})

Details: Built-in Message Pruning below, and the full story with every framework and store at https://skamalj.github.io/agentstate-reducer/.

Installation

pip install "langgraph-dynamodb-checkpoint[reducer]"

pip install langgraph-dynamodb-checkpoint     # checkpointer only; history is unbounded

Without reducer=, the saver logs one INFO line per process saying so, with the link above. Set AGENTSTATE_QUIET=1 to silence it.

Requires Python 3.10+. Single-table design, TTL expiry, delete by thread, sync and async, subgraphs. Passes all eight capabilities of langgraph-checkpoint-conformance (base plus copy_thread, delete_for_runs, prune), which is what LangSmith Deployment probes at startup; see LangSmith Deployment.

Usage

DynamoDBSaver Constructor

• table_name (str): Name of the DynamoDB table to use for storing checkpoints
• max_read_request_units (int, optional): Maximum read request units for the DynamoDB table. Defaults to 100
• max_write_request_units (int, optional): Maximum write request units for the DynamoDB table. Defaults to 100
• ttl_seconds (int, optional): TTL value set for all checkpoint items.
• reducer (MessageReducer, optional): Prunes the message history before each checkpoint is stored. See Built-in Message Pruning.
• messages_key (str, optional): State channel that holds the message list. Defaults to "messages".

Import

from langgraph_dynamodb_checkpoint import DynamoDBSaver

🔍 Enable Logging

langgraph_dynamodb uses Python's standard logging module and emits logs under the logger name langgraph_dynamodb.

You can control logging verbosity using the LANGGRAPH_DYNAMODB_LOG_LEVEL environment variable:

export LANGGRAPH_DYNAMODB_LOG_LEVEL=DEBUG

Available log levels:

• CRITICAL
• ERROR
• WARNING
• INFO (default)
• DEBUG
• NOTSET

Example:

LANGGRAPH_DYNAMODB_LOG_LEVEL=DEBUG 

⚙️ Programmatic Logging Configuration (Optional)

You can also configure logging directly in your code using configure_logging:

from langgraph_dynamodb_checkpoint import configure_logging
import logging

configure_logging(
    level=logging.DEBUG,
    log_format="%(levelname)s: %(message)s"
)

Redirect logs to a file:

with open("log.txt", "a") as logfile:
    configure_logging(level=logging.INFO, stream=logfile)

This gives you full control over the log destination, level, and format.

Initialize the saver with a table name

saver = DynamoDBSaver(
    table_name="your-dynamodb-table-name",
    max_read_request_units=10,  # Optional, default is 100
    max_write_request_units=10  # Optional, default is 100
    ttl_seconds=86400
)

Table has ttl enabled with attribute name set to ttl

Alternative Initialization Using Context Manager

from langgraph_dynamodb_checkpoint import DynamoDBSaver

with DynamoDBSaver.from_conn_info(table_name="your-dynamodb-table-name") as saver:
    # Use the saver here
    pass

Supports Delete

Checkpointer supports delete basis given thread_id

config = {"configurable": {"thread_id": "900"}}
memory.delete(config)

Table Structure

The saver automatically creates a DynamoDB table if it doesn't exist, with the following structure:

• Partition Key (PK): String type, used for thread_id
• Sort Key (SK): String type, used for checkpoint_id

AWS Configuration

Ensure you have proper AWS credentials configured either through:

• Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
• AWS credentials file (~/.aws/credentials)
• IAM role when running on AWS services

The AWS credentials should have permissions to:

• Create DynamoDB tables (if table doesn't exist)
• Read and write to DynamoDB tables

LangSmith Deployment

LangSmith Deployment (Agent Server) accepts a custom checkpointer through langgraph.json and checks its capabilities at startup. This saver implements the full set, so thread forking (copy_thread), the rollback multitask strategy (delete_for_runs) and history pruning (prune) are all available.

{
  "dependencies": ["."],
  "graphs": {"agent": "./src/agent/graph.py:graph"},
  "checkpointer": {"path": "./src/agent/checkpointer.py:generate_checkpointer"}
}

# src/agent/checkpointer.py
from contextlib import asynccontextmanager
from agentstate_reducer import MessageReducer, ReducerConfig
from langgraph_dynamodb_checkpoint import DynamoDBSaver

@asynccontextmanager
async def generate_checkpointer():
    reducer = MessageReducer(config=ReducerConfig(max_messages=20))
    yield DynamoDBSaver("checkpoints", reducer=reducer)

Housekeeping methods, also usable outside the platform:

saver.prune([thread_id], strategy="keep_latest")   # keep only the newest checkpoint per namespace
saver.delete_for_runs([run_id])                    # remove everything a run wrote
saver.copy_thread(thread_id, new_thread_id)        # fork a conversation

prune is not DeltaChannel-aware; do not use keep_latest on threads whose graph uses DeltaChannel. delete_for_runs finds checkpoints by a run_id attribute written since this version; older checkpoints are not matched.

Built-in Message Pruning

Long-running agents accumulate message history with every turn. Left unchecked this inflates checkpoint size, increases DynamoDB storage/throughput cost, and eventually blows past LLM context limits.

This checkpointer solves that at the persistence layer: pass a MessageReducer and it automatically prunes the message list inside put() before the checkpoint is serialised and written to DynamoDB. Your graph code, state definition, and node logic stay untouched.

This is an alternative to — or complement of — the LangGraph Annotated[list, reducer_fn] pattern. Use the checkpoint-layer approach when:

• You don't own the graph or state definition (e.g. using a pre-built LangGraph agent)
• You want pruning to happen unconditionally at every save
• You want to keep all in-memory state intact and only prune what gets persisted

Install with reducer support

pip install "langgraph_dynamodb_checkpoint[reducer]"

Usage — message-count pruning

from agentstate_reducer import MessageReducer
from langgraph_dynamodb_checkpoint import DynamoDBSaver

reducer = MessageReducer(min_messages=10, max_messages=20)

saver = DynamoDBSaver(
    table_name="your-table",
    reducer=reducer,          # prune before each checkpoint save
    messages_key="messages",  # state channel holding the message list (default)
)

Usage — token-budget pruning

from agentstate_reducer import MessageReducer, ReducerConfig
from langgraph_dynamodb_checkpoint import DynamoDBSaver

# Prune when the conversation exceeds 4000 tokens, down to ~2000 — whole messages only, never truncated
reducer = MessageReducer(config=ReducerConfig(max_tokens=4000, target_tokens=2000))
saver = DynamoDBSaver(table_name="your-table", reducer=reducer)

When pruning triggers, the oldest human/ai messages are removed until the target is met. The following are never pruned:

• Index 0 (typically the system prompt) — controlled by preserve_first=True
• system and function messages
• tool messages — unless their parent ai message is pruned (cascade behaviour, configurable)

See agentstate-reducer on PyPI for full configuration: message-count vs token-budget modes, preserve_first, cascade_tool_messages, summarize_fn, and role alias support (user/assistant/agent).

Usage — long-term memory via on_prune (agentstate-reducer >= 0.4.0)

Messages pruned from the checkpoint are exactly the ones leaving the model's view. The saver forwards a memory namespace to the reducer, and any on_prune hook receives (pruned_messages, namespace) — so pruned turns can flow straight into a LangGraph BaseStore (or LangMem, or any memory engine) with no extra node and no package coupling:

from uuid import uuid4
from agentstate_reducer import MessageReducer, ReducerConfig, Background
from langgraph_dynamodb_checkpoint import DynamoDBSaver

store = ...  # any langgraph BaseStore

def remember(pruned, namespace):
    for m in pruned:
        store.put(tuple(namespace), key=str(uuid4()), value={"role": m.type, "content": m.content})

reducer = MessageReducer(config=ReducerConfig(max_messages=20, on_prune=[remember]))
# on_prune=[Background(remember)] runs a slow hook (e.g. LLM extraction) off the request path
saver = DynamoDBSaver(table_name="your-table", reducer=reducer)
graph = builder.compile(checkpointer=saver, store=store)

graph.invoke(input, config={"configurable": {
    "thread_id": uuid4().hex,                    # short-term scope (this checkpoint)
    "memory_namespace": ("memories", user_id),   # long-term scope (the store)
}})

The saver reads memory_namespace (or whatever ReducerConfig.namespace_key names) from config["configurable"] on every put() and passes it through untouched. If the app never sets it, the namespace falls back to ("memories", thread_id). Each pruned message reaches the hooks once, even though LangGraph writes several checkpoints per turn.

Notes

• The saver automatically creates the DynamoDB table if it doesn't exist
• Uses on-demand billing mode for DynamoDB
• Implements all methods required by the LangGraph BaseCheckpointSaver interface