LangGraph DynamoDB Checkpointer

A DynamoDB-based checkpoint saver implementation for LangGraph, providing persistent state management for your LangGraph applications.

Features

• ✅ Persistent State Management - Store and retrieve graph execution checkpoints in DynamoDB
• ✅ Thread-based Organization - Organize checkpoints by thread ID for conversation continuity
• ✅ Automatic Table Creation - Sets up DynamoDB table automatically if it doesn't exist
• ✅ Parent-Child Relationships - Support for checkpoint hierarchies and branching
• ✅ Type Safety - Full TypeScript support with type definitions
• ✅ Flexible Configuration - Support for AWS credentials, regions, and custom table names

Installation

npm install langgraph-dynamodb-checkpointer

or with yarn:

yarn add langgraph-dynamodb-checkpointer

Prerequisites

• AWS Account with DynamoDB access
• AWS credentials configured (via environment variables, IAM role, or AWS config)
• Node.js 16+ and TypeScript 4.5+

Quick Start

Basic Usage

import { DynamoDBSaver } from 'langgraph-dynamodb-checkpointer';
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';

// Create DynamoDB client
const client = new DynamoDBClient({
  region: 'us-east-1',
  credentials: {
    accessKeyId: 'your-access-key',
    secretAccessKey: 'your-secret-key'
  }
});

// Initialize the checkpointer
const checkpointer = new DynamoDBSaver(client, 'my-checkpoints');

// Use with LangGraph
import { StateGraph } from '@langchain/langgraph';

const workflow = new StateGraph({
  channels: {
    messages: {
      value: (x, y) => [...x, ...y],
      default: () => []
    }
  }
});

// Add your nodes and edges...

// Compile with checkpointer
const app = workflow.compile({
  checkpointer: checkpointer
});

Using Environment Variables

Create a .env file:

AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=your-access-key
AWS_SECRET_ACCESS_KEY=your-secret-key
DYNAMODB_CHECKPOINTER_NAME=my-checkpoints

Then use the convenience factory method:

import { DynamoDBSaver } from 'langgraph-dynamodb-checkpointer';
import { getDynamoDBConfig } from 'langgraph-dynamodb-checkpointer/config';

// Load configuration from environment
const config = getDynamoDBConfig();

// Create checkpointer from config
const checkpointer = await DynamoDBSaver.fromConfig(config);

// Use with your LangGraph application
const app = workflow.compile({
  checkpointer: checkpointer
});

API Reference

DynamoDBSaver

The main class for checkpoint management.

Constructor

constructor(
  client: DynamoDBClient,
  tableName?: string,
  serde?: SerializerProtocol<unknown>
)

• client: AWS DynamoDB client instance
• tableName: Name of the DynamoDB table (default: 'checkpoints')
• serde: Optional custom serializer/deserializer

Methods

fromConfig(config: DynamoDBConfig): Promise<DynamoDBSaver>

Factory method to create a checkpointer from configuration.

const checkpointer = await DynamoDBSaver.fromConfig({
  region: 'us-east-1',
  credentials: {
    accessKeyId: 'key',
    secretAccessKey: 'secret'
  },
  tableName: 'my-checkpoints'
});

getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined>

Retrieve a specific checkpoint or the latest checkpoint for a thread.

// Get latest checkpoint for a thread
const checkpoint = await checkpointer.getTuple({
  configurable: {
    thread_id: 'user-123'
  }
});

// Get specific checkpoint
const checkpoint = await checkpointer.getTuple({
  configurable: {
    thread_id: 'user-123',
    checkpoint_id: 'checkpoint-456'
  }
});

put(config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata): Promise<RunnableConfig>

Save a new checkpoint.

const newConfig = await checkpointer.put(
  { configurable: { thread_id: 'user-123' } },
  checkpoint,
  { createdAt: new Date().toISOString() }
);

list(config: RunnableConfig, limit?: number, before?: RunnableConfig): AsyncGenerator<CheckpointTuple>

List checkpoints for a thread.

// List all checkpoints for a thread
for await (const checkpoint of checkpointer.list(
  { configurable: { thread_id: 'user-123' } }
)) {
  console.log(checkpoint);
}

// List with limit
for await (const checkpoint of checkpointer.list(
  { configurable: { thread_id: 'user-123' } },
  10
)) {
  console.log(checkpoint);
}

delete(thread_id: string, checkpoint_id?: string): Promise<void>

Delete checkpoints.

// Delete specific checkpoint
await checkpointer.delete('user-123', 'checkpoint-456');

// Delete all checkpoints for a thread
await checkpointer.delete('user-123');

DynamoDB Table Structure

The checkpointer creates a DynamoDB table with the following schema:

| Attribute | Type | Description | |-----------|------|-------------| | thread_id | String (HASH) | Unique identifier for the conversation thread | | checkpoint_id | String (RANGE) | Unique identifier for the checkpoint | | parent_id | String | Optional parent checkpoint ID | | checkpoint | String | Serialized checkpoint data | | metadata | String | Serialized metadata

Configuration Options

DynamoDBConfig Interface

interface DynamoDBConfig {
  region?: string;              // AWS region (default: 'us-east-1')
  endpoint?: string;            // Custom endpoint (for local DynamoDB)
  credentials?: {               // AWS credentials (optional if using IAM roles)
    accessKeyId: string;
    secretAccessKey: string;
  };
  tableName?: string;           // DynamoDB table name (default: 'checkpoints')
}

Advanced Usage

Custom Serialization

You can provide a custom serializer for checkpoint data:

import { SerializerProtocol } from '@langchain/langgraph/dist/serde/base';

class CustomSerializer implements SerializerProtocol<any> {
  stringify(obj: any): string {
    // Custom serialization logic
    return JSON.stringify(obj);
  }
  
  parse(s: string): any {
    // Custom deserialization logic
    return JSON.parse(s);
  }
}

const checkpointer = new DynamoDBSaver(
  client,
  'checkpoints',
  new CustomSerializer()
);

Using with LangGraph Agents

import { createReactAgent } from '@langchain/langgraph/prebuilt';
import { ChatOpenAI } from '@langchain/openai';

const agent = createReactAgent({
  llm: new ChatOpenAI(),
  tools: [...yourTools],
  checkpointer: checkpointer
});

// Run with thread persistence
const result = await agent.invoke(
  { messages: [{ role: 'user', content: 'Hello!' }] },
  { configurable: { thread_id: 'user-123' } }
);

Error Handling

The checkpointer includes comprehensive error handling:

try {
  const checkpoint = await checkpointer.getTuple({
    configurable: { thread_id: 'user-123' }
  });
} catch (error) {
  if (error.name === 'ResourceNotFoundException') {
    console.log('Table does not exist');
  } else {
    console.error('Error retrieving checkpoint:', error);
  }
}

AWS IAM Permissions

Your AWS credentials need the following DynamoDB permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:CreateTable",
        "dynamodb:DescribeTable",
        "dynamodb:PutItem",
        "dynamodb:GetItem",
        "dynamodb:Query",
        "dynamodb:DeleteItem",
        "dynamodb:UpdateTimeToLive"
      ],
      "Resource": [
        "arn:aws:dynamodb:*:*:table/checkpoints",
        "arn:aws:dynamodb:*:*:table/checkpoints/*"
      ]
    }
  ]
}

Troubleshooting

Table Creation Issues

If the table isn't created automatically:

• Ensure your AWS credentials have CreateTable permissions
• Check the AWS region is correctly configured
• Verify network connectivity to AWS

Checkpoint Retrieval Issues

If checkpoints aren't being retrieved:

• Verify the thread_id is consistent across requests
• Ensure proper read permissions on the DynamoDB table

Performance Considerations

• Use DynamoDB on-demand billing for variable workloads
• Consider provisioned capacity for consistent high-volume usage
• Checkpoints are automatically indexed by thread_id for efficient queries

Development

Building from Source

# Clone the repository
git clone https://github.com/SerezhaGRig/langgraph-dynamodb-checkpointer.git

# Install dependencies
npm install

# Build the package
npm run build

# Run tests (when available)
npm test

Project Structure

langgraph-dynamodb-checkpointer/
├── src/
│   ├── index.ts        # Main exports
│   ├── dynamo.ts       # DynamoDBSaver implementation
│   └── config.ts       # Configuration utilities
├── dist/               # Compiled JavaScript output
├── package.json
├── tsconfig.json
└── README.md

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/AmazingFeature)
3. Commit your changes (git commit -m 'Add some AmazingFeature')
4. Push to the branch (git push origin feature/AmazingFeature)
5. Open a Pull Request

License

This project is licensed under the ISC License - see the LICENSE file for details.

Support

For issues, questions, or suggestions, please open an issue on the GitHub repository.

Acknowledgments

• Built for LangGraph by LangChain
• Uses AWS SDK v3 for DynamoDB operations
• Inspired by the need for persistent state management in conversational AI applications

Changelog

Version 1.0.0 (Initial Release)

• Basic checkpoint save/load functionality
• Automatic DynamoDB table creation
• Thread-based checkpoint organization
• Full TypeScript support