Metadata-Version: 2.5
Name: langgraph_dynamodb_checkpoint
Version: 0.5.0
Summary: LangGraph checkpoint saver for Amazon DynamoDB with built-in message history pruning via agentstate-reducer - single table, auto create, TTL and delete handling
Project-URL: Homepage, https://github.com/skamalj/langgraph_dynamodb_checkpoint
Project-URL: Documentation, https://readthedocs.org
Project-URL: Repository, https://github.com/skamalj/langgraph_dynamodb_checkpoint.git
Project-URL: Issues, https://github.com/skamalj/langgraph_dynamodb_checkpoint/issues
Project-URL: Changelog, https://github.com/skamalj/langgraph_dynamodb_checkpoint/blob/main/CHANGELOG.md
Author-email: Kamal <skamalj@github.com>
Keywords: agent-state,aws,checkpoint,dynamodb,langchain,langgraph,memory,message-reducer
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Requires-Python: >=3.10
Requires-Dist: boto3
Requires-Dist: botocore
Requires-Dist: langchain-core
Requires-Dist: langgraph
Requires-Dist: langgraph-checkpoint>=4.1.1
Provides-Extra: dev
Requires-Dist: black; extra == 'dev'
Requires-Dist: isort; extra == 'dev'
Requires-Dist: langgraph-checkpoint-conformance; extra == 'dev'
Requires-Dist: mypy; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Provides-Extra: reducer
Requires-Dist: agentstate-reducer>=0.4.0; extra == 'reducer'
Description-Content-Type: text/markdown

# 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.

<p align="center"><img src="https://raw.githubusercontent.com/skamalj/langgraph_dynamodb_checkpoint/main/docs/agentstate-flow.svg" width="100%" alt="Messages accumulate in the checkpoint until the window's upper bound, the reducer prunes back to the lower bound, and the pruned turns flow through the on_prune hook into a long-term store"></p>

- **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`](https://pypi.org/project/langgraph-memory/) is the ready-made one.
- **One line of config.**

```python
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](#built-in-message-pruning) below, and the full story with every framework and store at [https://skamalj.github.io/agentstate-reducer/](https://skamalj.github.io/agentstate-reducer/langgraph/dynamodb/).

## Installation

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

```bash
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](#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](#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:

```bash
export LANGGRAPH_DYNAMODB_LOG_LEVEL=DEBUG
```

#### Available log levels:

* `CRITICAL`
* `ERROR`
* `WARNING`
* `INFO` (default)
* `DEBUG`
* `NOTSET`

#### Example:

```bash
LANGGRAPH_DYNAMODB_LOG_LEVEL=DEBUG 
```

---

### ⚙️ Programmatic Logging Configuration (Optional)

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

```python
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:

```python
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.

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

```python
# 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:

```python
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

```bash
pip install "langgraph_dynamodb_checkpoint[reducer]"
```

### Usage — message-count pruning

```python
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

```python
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](https://pypi.org/project/agentstate-reducer/) 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:

```python
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