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