LangGraph Checkpoint Couchbase

A Couchbase implementation of the LangGraph CheckpointSaver interface that enables persisting agent state and conversation history in a Couchbase database.

Overview

This package provides a seamless way to persist LangGraph agent states in Couchbase, enabling:

• State persistence across application restarts
• Retrieval of historical conversation steps
• Continued conversations from previous checkpoints
• Both synchronous and asynchronous interfaces

Installation

pip install langgraph-checkpointer-couchbase

Requirements

• Python 3.10+
• Couchbase Server (7.0+ recommended)
• Couchbase Python SDK 4.6.3+
• LangGraph 1.0.5+
• LangChain 1.1.3+ and LangChain OpenAI 1.1.3+

Prerequisites

• A running Couchbase cluster
• A bucket created for storing checkpoints
• Appropriate credentials with read/write access

Quick Start

First, set up your agent tools and model:

from typing import Literal
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: Literal["nyc", "sf"]):
    """Use this to get weather information."""
    if city == "nyc":
        return "It might be cloudy in nyc"
    elif city == "sf":
        return "It's always sunny in sf"
    else:
        raise AssertionError("Unknown city")


tools = [get_weather]
model = ChatOpenAI(model="gpt-5-mini", temperature=0)

Synchronous Usage

import os
from langgraph_checkpointer_couchbase import CouchbaseSaver
from langchain.agents import create_agent

with CouchbaseSaver.from_conn_info(
        cb_conn_str=os.getenv("CB_CLUSTER") or "couchbase://localhost",
        cb_username=os.getenv("CB_USERNAME") or "Administrator",
        cb_password=os.getenv("CB_PASSWORD") or "password",
        bucket_name=os.getenv("CB_BUCKET") or "test",
        scope_name=os.getenv("CB_SCOPE") or "langgraph",
    ) as checkpointer:
    # Create the agent with checkpointing
    graph = create_agent(model, tools=tools, checkpointer=checkpointer)
    
    # Configure with a unique thread ID
    config = {"configurable": {"thread_id": "1"}}
    
    # Run the agent
    res = graph.invoke({"messages": [("human", "what's the weather in sf")]}, config)
    
    # Retrieve checkpoints
    latest_checkpoint = checkpointer.get(config)
    latest_checkpoint_tuple = checkpointer.get_tuple(config)
    checkpoint_tuples = list(checkpointer.list(config))

    print(latest_checkpoint)
    print(latest_checkpoint_tuple)
    print(checkpoint_tuples)

Asynchronous Usage

import os
from acouchbase.cluster import Cluster as ACluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions
from langgraph_checkpointer_couchbase import AsyncCouchbaseSaver
from langchain.agents import create_agent

auth = PasswordAuthenticator(
    os.getenv("CB_USERNAME") or "Administrator",
    os.getenv("CB_PASSWORD") or "password",
)
options = ClusterOptions(auth)
cluster = await ACluster.connect(os.getenv("CB_CLUSTER") or "couchbase://localhost", options)

bucket_name = os.getenv("CB_BUCKET") or "test"
scope_name = os.getenv("CB_SCOPE") or "langgraph"

async with AsyncCouchbaseSaver.from_cluster(
        cluster=cluster,
        bucket_name=bucket_name,
        scope_name=scope_name,
    ) as checkpointer:
    # Create the agent with checkpointing
    graph = create_agent(model, tools=tools, checkpointer=checkpointer)
    
    # Configure with a unique thread ID
    config = {"configurable": {"thread_id": "2"}}
    
    # Run the agent asynchronously
    res = await graph.ainvoke(
        {"messages": [("human", "what's the weather in nyc")]}, config
    )

    # Retrieve checkpoints asynchronously
    latest_checkpoint = await checkpointer.aget(config)
    latest_checkpoint_tuple = await checkpointer.aget_tuple(config)
    checkpoint_tuples = [c async for c in checkpointer.alist(config)]

    print(latest_checkpoint)
    print(latest_checkpoint_tuple)
    print(checkpoint_tuples)

# Close the cluster when done
await cluster.close()

Recommended Indexes

The saver creates the checkpoints and checkpoint_writes collections on first use, but not indexes. Without them, queries fall back to sequential scans, which are slow on large collections and may not return a checkpoint written immediately before. Create these indexes once per scope (replace test and langgraph with your bucket and scope):

CREATE INDEX idx_checkpoints_thread IF NOT EXISTS
  ON `test`.`langgraph`.`checkpoints`(thread_id, checkpoint_ns, checkpoint_id);
CREATE INDEX idx_checkpoint_writes_thread IF NOT EXISTS
  ON `test`.`langgraph`.`checkpoint_writes`(thread_id, checkpoint_ns, checkpoint_id);

Configuration Options

Running the Tests

The checkpointer tests run against a live Couchbase cluster and do not need an LLM. Create the bucket, scope, and the recommended indexes first, then:

pip install -e . pytest pytest-asyncio
export CB_CLUSTER=couchbase://localhost CB_USERNAME=Administrator CB_PASSWORD=password CB_BUCKET=test CB_SCOPE=langgraph
pytest tests/test_checkpointer.py

The end-to-end agent example in tests/agent_e2e_test.py additionally requires OPENAI_API_KEY:

python tests/agent_e2e_test.py

Usage Data

This product automatically collects usage and performance data (such as product name and version) and browser information (such as IP address) (collectively, "Usage Data"). Couchbase uses Usage Data, along with other data you may provide to Couchbase (such as your user name or email address), to develop and improve our products as well as inform our sales and marketing programs. We do not access or collect any data you store in Couchbase products. We use Usage Data to understand aggregate usage patterns and make our products more useful to you. For more information on how Couchbase collects, protects, and processes information, please refer to the Couchbase Privacy Policy viewable at https://www.couchbase.com/privacy-policy.

License

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

Contributing

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

📢 Support Policy

We truly appreciate your interest in this project!
This project is community-maintained, which means it's not officially supported by our support team.

If you need help, have found a bug, or want to contribute improvements, the best place to do that is right here — by opening a GitHub issue.
Our support portal is unable to assist with requests related to this project, so we kindly ask that all inquiries stay within GitHub.

Your collaboration helps us all move forward together — thank you!