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.8+
• Couchbase Server (7.0+ recommended)
• LangGraph 0.3.22+
• LangChain OpenAI 0.3.11+

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_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_name="gpt-4o-mini", temperature=0)

Synchronous Usage

import os
from langgraph_checkpointer_couchbase import CouchbaseSaver
from langgraph.graph import create_react_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_react_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 langgraph.graph import create_react_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_react_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()

Configuration Options

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!