Skip to content

Store API

LangGraph long-term memory stores for CockroachDB.

CockroachDBStore

Synchronous store.

langchain_cockroachdb.store.store.CockroachDBStore

Bases: BaseStore, BaseCockroachDBStore, PsycopgOps

LangGraph store that keeps long-term memory in CockroachDB.

Items live in one table keyed by namespace and key. Namespaces are stored as their labels joined with ".". For production, pass a psycopg_pool.ConnectionPool or use from_conn_string with pool_config: a pool replaces broken connections and runs batches concurrently, while a single connection runs one batch at a time.

A batch runs its reads first, then all its puts and deletes in one transaction. Its gets, its searches and its listings take one round trip each. On a store with a vector index, semantic searches under a namespace add one before the searches, to find their namespaces, and one after them when a search must run again exactly. A flat store adds neither. The long-term memory guide explains why.

Example
from langchain_cockroachdb import CockroachDBStore

with CockroachDBStore.from_conn_string(
    "cockroachdb://root@localhost:26257/defaultdb?sslmode=disable"
) as store:
    store.setup()
    store.put(("users", "123"), "prefs", {"theme": "dark"})
    item = store.get(("users", "123"), "prefs")

__init__(conn, *, pipe=None, index=None, ttl=None, retry_policy=None, table_name='store', schema='public')

Create a store on a connection or a pool.

Parameters:

Name Type Description Default
conn Conn

A psycopg connection, or a ConnectionPool (recommended). It must be in autocommit mode and use dict_row.

required
pipe Pipeline | None

A pipeline on conn, only with a single connection

None
index CockroachDBIndexConfig | None

Semantic search settings. When given, setup() also creates the vectors table and its C-SPANN index, and puts embed the configured fields of each item.

None
ttl TTLConfig | CockroachDBTTLConfig | None

TTL settings. When given, setup() also turns on CockroachDB row-level TTL for expired rows.

None
retry_policy RetryPolicy | None

How operations retry (default: RetryPolicy())

None
table_name str

Name of the items table (default "store")

'store'
schema str

Schema of the store's tables (default "public")

'public'

Raises:

Type Description
ValueError

If a pipeline is given with a pool, or a name or the TTL or index settings are invalid

abatch(ops) async

Run batch in a worker thread, as PostgresStore does.

batch(ops)

Run a batch of operations.

Every op is validated, and every put value serialized, before any statement runs. Gets, searches and listings run first, then all puts and deletes commit together in one transaction.

Parameters:

Name Type Description Default
ops Iterable[Op]

GetOp, SearchOp, ListNamespacesOp and PutOp operations.

required

Returns:

Type Description
list[Result]

One result per op, in the order of ops.

delete_namespace(namespace_prefix)

Delete every item in a namespace and its descendants.

("user", "a") deletes ("user", "a") and ("user", "a", "b"), not ("user", "ab"). One statement deletes the items and their vectors.

Parameters:

Name Type Description Default
namespace_prefix tuple[str, ...]

The namespace to delete. It cannot be empty.

required

Returns:

Type Description
int

The number of items deleted. If a retry follows a commit whose

int

result was lost, the count is that of the retry, usually 0.

Raises:

Type Description
InvalidNamespaceError

If the prefix is empty or a label is invalid

disable_ttl()

Turn off CockroachDB row-level TTL on the store's tables.

Expired items are still left out of reads while omit_expired is on. Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another schema change runs on the same table; that error is retried under the store's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

drop_vector_index()

Drop the store's vector index, if it exists.

Run setup() afterwards to build it again with the store's current configuration, for example to change the distance. Searches scan the vectors until the new index is built.

from_conn_string(conn_string, *, pipeline=False, pool_config=None, index=None, ttl=None, retry_policy=None, connect_timeout=10.0, prepare_threshold=5, connect_kwargs=None, table_name='store', schema='public') classmethod

Create a store on a new connection, or on a new pool with pool_config.

Parameters:

Name Type Description Default
conn_string str

CockroachDB connection string (cockroachdb://, postgresql:// and similar)

required
pipeline bool

Whether to use pipeline mode on the single connection. Ignored with pool_config.

False
pool_config PoolConfig | None

Open a ConnectionPool with these settings instead of one connection

None
index CockroachDBIndexConfig | None

Semantic search settings

None
ttl TTLConfig | CockroachDBTTLConfig | None

TTL settings

None
retry_policy RetryPolicy | None

How operations retry (default: RetryPolicy())

None
connect_timeout float | None

Seconds to wait when connecting, or None (default: 10)

10.0
prepare_threshold int | None

psycopg prepared statement threshold, None to disable (default: 5). Statements sent in a pipeline are never prepared

5
connect_kwargs Mapping[str, Any] | None

Extra psycopg connection options, e.g. application_name

None
table_name str

Name of the items table (default "store")

'store'
schema str

Schema of the store's tables (default "public")

'public'

Yields:

Type Description
CockroachDBStore

CockroachDBStore instance. The connection or pool closes on exit.

With pool_config, pipeline is ignored, as in PostgresStore: a pipeline needs a single connection.

setup()

Create the store's tables and run its migrations.

MUST be called the first time the store is used. Safe to run from several processes at once: every migration is idempotent and versions are recorded with ON CONFLICT DO NOTHING. With an index, it creates the vectors table and its C-SPANN index (unless flat) on every call, then checks an existing index against the configuration. With a TTL configuration, it then turns on CockroachDB row-level TTL on expires_at.

On CockroachDB v25.4 a TTL change fails while another setup() changes the same table. That error is retried under the store's retry policy, even if its retry_on would refuse it; max_attempts=1 turns this off.

Raises:

Type Description
ValueError

If the vectors table exists with other dims, or its index uses another distance. A different partition size, or an index while flat is set, logs a WARNING instead.

start_ttl_sweeper(sweep_interval_minutes=None)

Sweep expired items periodically in a background thread.

CockroachDB's row-level TTL job, turned on by setup() with a TTL configuration, already deletes expired rows. The sweeper deletes them sooner. Expired items are never returned either way.

Parameters:

Name Type Description Default
sweep_interval_minutes int | None

Minutes between sweeps. Defaults to the TTL configuration's sweep_interval_minutes, else 5.

None

Returns:

Type Description
Future[None]

A future that completes when the sweeper stops. Cancelling it

Future[None]

stops the sweeper.

stop_ttl_sweeper(timeout=None)

Stop the TTL sweeper thread if it is running.

Parameters:

Name Type Description Default
timeout float | None

Seconds to wait for the thread to stop, or None to wait.

None

Returns:

Type Description
bool

True if the thread stopped or was not running, False on timeout.

sweep_ttl()

Delete expired items now, without waiting for CockroachDB's TTL job.

Returns:

Type Description
int

The number of items deleted.

AsyncCockroachDBStore

Asynchronous store.

langchain_cockroachdb.store.aio.AsyncCockroachDBStore

Bases: AsyncBatchedBaseStore, BaseCockroachDBStore, AsyncPsycopgOps

Async LangGraph store that keeps long-term memory in CockroachDB.

The async twin of CockroachDBStore: same tables, same SQL, same behavior. For production, pass a psycopg_pool.AsyncConnectionPool or use from_conn_string with pool_config: a pool replaces broken connections, while a single connection cannot recover from a dropped one.

aget, aput, asearch and the other awaited calls, and the sync wrappers, go through AsyncBatchedBaseStore's queue, which runs one abatch at a time. So a pool does not run those calls in parallel, and a large pool does not speed them up. What the queue gives instead is coalescing: ops queued in the same tick run as one batch, with one round trip per read group and one transaction for the writes. Only direct abatch calls, or sync batch calls from other threads, run concurrently on a pool.

The constructor must run inside a running event loop, because AsyncBatchedBaseStore starts a task on it there. Awaited calls made in the same tick (aget, aput, asearch and the others) are coalesced into one abatch. The sync methods (get, put and the others) are meant for other threads while that loop runs. They run the async methods on it, so get, put, search, delete and list_namespaces join the same queue, while batch runs abatch directly. From the loop itself they raise asyncio.InvalidStateError. When the loop is closed or not running, or the call comes from a forked child of the process that created the store, they raise RuntimeError instead of waiting forever, as AsyncCockroachDBSaver's sync methods do.

A batch runs its reads first, then all its puts and deletes in one transaction. Its gets, its searches and its listings take one round trip each. On a store with a vector index, semantic searches under a namespace add one before the searches, to find their namespaces, and one after them when a search must run again exactly. A flat store adds neither. The long-term memory guide explains why.

Example
from langchain_cockroachdb import AsyncCockroachDBStore

async with AsyncCockroachDBStore.from_conn_string(
    "cockroachdb://root@localhost:26257/defaultdb?sslmode=disable"
) as store:
    await store.setup()
    await store.aput(("users", "123"), "prefs", {"theme": "dark"})
    item = await store.aget(("users", "123"), "prefs")

__init__(conn, *, pipe=None, index=None, ttl=None, retry_policy=None, table_name='store', schema='public')

Create a store on a connection or a pool, inside a running event loop.

Parameters:

Name Type Description Default
conn AsyncConn

A psycopg AsyncConnection, or an AsyncConnectionPool (recommended). It must be in autocommit mode and use dict_row.

required
pipe AsyncPipeline | None

A pipeline on conn, only with a single connection

None
index CockroachDBIndexConfig | None

Semantic search settings. When given, setup() also creates the vectors table and its C-SPANN index, and puts embed the configured fields of each item.

None
ttl TTLConfig | CockroachDBTTLConfig | None

TTL settings. When given, setup() also turns on CockroachDB row-level TTL for expired rows.

None
retry_policy RetryPolicy | None

How operations retry (default: RetryPolicy())

None
table_name str

Name of the items table (default "store")

'store'
schema str

Schema of the store's tables (default "public")

'public'

Raises:

Type Description
RuntimeError

If no event loop is running

ValueError

If a pipeline is given with a pool, or a name or the TTL or index settings are invalid

abatch(ops) async

Run a batch of operations.

Every op is validated, and every put value serialized, before any statement runs. Gets, searches and listings run first, then all puts and deletes commit together in one transaction.

Parameters:

Name Type Description Default
ops Iterable[Op]

GetOp, SearchOp, ListNamespacesOp and PutOp operations.

required

Returns:

Type Description
list[Result]

One result per op, in the order of ops.

adelete_namespace(namespace_prefix) async

Delete every item in a namespace and its descendants.

("user", "a") deletes ("user", "a") and ("user", "a", "b"), not ("user", "ab"). One statement deletes the items and their vectors.

Parameters:

Name Type Description Default
namespace_prefix tuple[str, ...]

The namespace to delete. It cannot be empty.

required

Returns:

Type Description
int

The number of items deleted. If a retry follows a commit whose

int

result was lost, the count is that of the retry, usually 0.

Raises:

Type Description
InvalidNamespaceError

If the prefix is empty or a label is invalid

adisable_ttl() async

Turn off CockroachDB row-level TTL on the store's tables.

Expired items are still left out of reads while omit_expired is on. Safe to run from several processes at once. On CockroachDB v25.4 a TTL change fails while another schema change runs on the same table; that error is retried under the store's retry policy, even if its retry_on would refuse it. max_attempts=1 turns this off.

adrop_vector_index() async

Drop the store's vector index, if it exists.

Run setup() afterwards to build it again with the store's current configuration, for example to change the distance. Searches scan the vectors until the new index is built.

batch(ops)

Run abatch on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

delete(namespace, key)

Run adelete on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

from_conn_string(conn_string, *, pipeline=False, pool_config=None, index=None, ttl=None, retry_policy=None, connect_timeout=10.0, prepare_threshold=5, connect_kwargs=None, table_name='store', schema='public') async classmethod

Create a store on a new connection, or on a new pool with pool_config.

Parameters:

Name Type Description Default
conn_string str

CockroachDB connection string (cockroachdb://, postgresql:// and similar)

required
pipeline bool

Whether to use pipeline mode on the single connection. Ignored with pool_config.

False
pool_config PoolConfig | None

Open an AsyncConnectionPool with these settings instead of one connection

None
index CockroachDBIndexConfig | None

Semantic search settings

None
ttl TTLConfig | CockroachDBTTLConfig | None

TTL settings

None
retry_policy RetryPolicy | None

How operations retry (default: RetryPolicy())

None
connect_timeout float | None

Seconds to wait when connecting, or None (default: 10)

10.0
prepare_threshold int | None

psycopg prepared statement threshold, None to disable (default: 5). Statements sent in a pipeline are never prepared

5
connect_kwargs Mapping[str, Any] | None

Extra psycopg connection options, e.g. application_name

None
table_name str

Name of the items table (default "store")

'store'
schema str

Schema of the store's tables (default "public")

'public'

Yields:

Type Description
AsyncIterator[AsyncCockroachDBStore]

AsyncCockroachDBStore instance. The connection or pool closes on exit.

With pool_config, pipeline is ignored, as in AsyncPostgresStore: a pipeline needs a single connection.

get(namespace, key, *, refresh_ttl=None)

Run aget on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

list_namespaces(*, prefix=None, suffix=None, max_depth=None, limit=100, offset=0)

Run alist_namespaces on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

put(namespace, key, value, index=None, *, ttl=NOT_PROVIDED)

Run aput on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

search(namespace_prefix, /, *, query=None, filter=None, limit=10, offset=0, refresh_ttl=None)

Run asearch on the store's event loop, from another thread.

Raises:

Type Description
InvalidStateError

If called from the store's loop

RuntimeError

If that loop is closed, is not running or stops before the work finishes, or in a forked child

setup() async

Create the store's tables and run its migrations.

MUST be called the first time the store is used. Safe to run from several processes at once: every migration is idempotent and versions are recorded with ON CONFLICT DO NOTHING. With an index, it creates the vectors table and its C-SPANN index (unless flat) on every call, then checks an existing index against the configuration. With a TTL configuration, it then turns on CockroachDB row-level TTL on expires_at.

On CockroachDB v25.4 a TTL change fails while another setup() changes the same table. That error is retried under the store's retry policy, even if its retry_on would refuse it; max_attempts=1 turns this off.

Raises:

Type Description
ValueError

If the vectors table exists with other dims, or its index uses another distance. A different partition size, or an index while flat is set, logs a WARNING instead.

start_ttl_sweeper(sweep_interval_minutes=None) async

Sweep expired items periodically in a background task.

CockroachDB's row-level TTL job, turned on by setup() with a TTL configuration, already deletes expired rows. The sweeper deletes them sooner. Expired items are never returned either way.

Parameters:

Name Type Description Default
sweep_interval_minutes int | None

Minutes between sweeps. Defaults to the TTL configuration's sweep_interval_minutes, else 5.

None

Returns:

Type Description
Task[None]

The sweeper task. It finishes when the sweeper stops. Without a

Task[None]

TTL configuration, a task that finishes at once.

stop_ttl_sweeper(timeout=None) async

Stop the TTL sweeper task if it is running.

Parameters:

Name Type Description Default
timeout float | None

Seconds to wait for the task to stop, or None to wait.

None

Returns:

Type Description
bool

True if the task stopped or was not running, False on timeout.

sweep_ttl() async

Delete expired items now, without waiting for CockroachDB's TTL job.

Returns:

Type Description
int

The number of items deleted.

CockroachDBIndexConfig

Semantic search settings, passed as index.

langchain_cockroachdb.store.config.CockroachDBIndexConfig

Bases: IndexConfig

CockroachDB-specific index configuration.

Extends LangGraph IndexConfig with CockroachDB vector index options.

Attributes:

Name Type Description
distance_type Literal['cosine', 'l2', 'inner_product']

Distance metric (cosine, l2, or inner_product). Default cosine.

cspann CSPANNConfig | None

C-SPANN index parameters (both or neither).

flat bool

Whether to use a flat (brute-force) index. Default False.

beam_size int

Beam width for C-SPANN search. Searches with a query set it with SET LOCAL in their own transaction. Inside a transaction the caller opened on a single connection shared with the store, it lasts until that transaction ends.

CSPANNConfig

C-SPANN partition sizes, passed as index["cspann"].

langchain_cockroachdb.store.config.CSPANNConfig

Bases: TypedDict

C-SPANN index configuration.

Attributes:

Name Type Description
min_partition_size int

Minimum vectors per partition.

max_partition_size int

Maximum vectors per partition.

CockroachDBTTLConfig

Expiry settings, passed as ttl.

langchain_cockroachdb.store.config.CockroachDBTTLConfig

Bases: TTLConfig

CockroachDB-specific TTL configuration.

Extends LangGraph TTLConfig with CockroachDB TTL options.

Attributes:

Name Type Description
omit_expired bool

Whether to exclude expired items from reads. Default True.

ttl_job_cron str

Cron expression for the TTL cleanup job.

PoolConfig

Connection pool settings, passed as pool_config to from_conn_string.

langchain_cockroachdb.store.config.PoolConfig

Bases: TypedDict

Connection pool settings for CockroachDBStore.from_conn_string.

The same meaning as PostgresStore's PoolConfig.

Attributes:

Name Type Description
min_size int

Connections the pool keeps open. Default 1.

max_size int | None

Most connections the pool opens, or None for min_size.

kwargs dict[str, Any]

Extra psycopg connection options for every pooled connection. autocommit, row_factory and prepare_threshold are set by the store.

Imports

Name langchain_cockroachdb.store langchain_cockroachdb
CockroachDBStore Yes Yes
AsyncCockroachDBStore Yes Yes
CockroachDBIndexConfig Yes Yes
CockroachDBTTLConfig Yes Yes
PoolConfig Yes No
CSPANNConfig No, import it from langchain_cockroachdb.store.config No

Key Methods

Method Async Method Description
setup() setup() Create the tables and run migrations, build the vector index with index, turn on TTL with ttl
put() aput() Store an item, or replace it
get() aget() Read an item
search() asearch() Search a namespace prefix, with a filter and an optional semantic query
delete() adelete() Delete an item
list_namespaces() alist_namespaces() List namespaces by prefix, suffix and depth
batch() abatch() Run several operations, with the writes in one transaction
delete_namespace() adelete_namespace() Delete a namespace and its descendants, return the item count
sweep_ttl() sweep_ttl() Delete expired items now, return the count
start_ttl_sweeper() start_ttl_sweeper() Sweep periodically in a background thread or task
stop_ttl_sweeper() stop_ttl_sweeper() Stop the sweeper
disable_ttl() adisable_ttl() Turn off row-level TTL on the store's tables
drop_vector_index() adrop_vector_index() Drop the vector index, to rebuild it with setup()

On AsyncCockroachDBStore, setup(), sweep_ttl(), start_ttl_sweeper() and stop_ttl_sweeper() are coroutines without the a prefix, as in AsyncPostgresStore.

Factory Methods

Method Description
from_conn_string(conn_string, ...) Create a store on a new connection, or on a new pool with pool_config (context manager)

Connection Types

Both stores accept these connection types, in autocommit mode with dict_row:

Type Description
psycopg.Connection / AsyncConnection Single connection (simple scripts)
psycopg_pool.ConnectionPool / AsyncConnectionPool Connection pool (production)

Examples

See the Long-Term Memory Guide for usage.