CockroachDB 集成 - 以编程方式使用文档

>CockroachDB 是一个分布式 SQL 数据库，构建在事务性和强一致性键值存储之上。它可以水平扩展，能够在磁盘、机器、机架甚至数据中心发生故障时保持最小延迟中断且无需人工干预。

主要特性： - **分布式 SQL**：在保持 ACID 保证的同时横向扩展 - **原生向量支持**：内置 VECTOR 类型 (v24.2+) 和 C-SPANN 索引 (v25.2+) - **PostgreSQL 兼容**：PostgreSQL 应用程序的直接替代品 - **全局复制**：多区域部署，低延迟 - **自动分片**：数据自动分发到各个节点 - **SERIALIZABLE 隔离**：默认的最强隔离级别

安装与设置

安装 LangChain 集成：

pip install langchain-cockroachdb

获取 CockroachDB 连接字符串

您需要一个 CockroachDB 集群。选择以下一种方式：

选项 1：CockroachDB Cloud（推荐） 1. 在以下网址注册 cockroachlabs.cloud 2. 创建免费集群 3. 获取您的连接字符串： cockroachdb://user:pass@host:26257/db?sslmode=verify-full

选项 2：Docker（开发环境）

docker run -d --name cockroachdb -p 26257:26257 \
  cockroachdb/cockroach:latest start-single-node --insecure

连接字符串： cockroachdb://root@localhost:26257/defaultdb?sslmode=disable

选项 3：本地二进制文件

从以下网址下载 cockroachlabs.com/docs/releases

集成

向量存储

CockroachDB 可作为向量存储使用，支持原生 VECTOR 类型和 C-SPANN 分布式索引。

主要特性： - 原生向量支持 (v24.2+) - 针对分布式系统优化的 C-SPANN 索引 (v25.2+) - 高级元数据过滤 - 使用前缀列实现多租户 - 水平可扩展性

参见 CockroachDB 向量存储文档了解更多详情。

快速示例：

from langchain_cockroachdb import AsyncCockroachDBVectorStore, CockroachDBEngine
from langchain_openai import OpenAIEmbeddings

# Initialize
engine = CockroachDBEngine.from_connection_string(
    "cockroachdb://user:pass@host:26257/db"
)

await engine.ainit_vectorstore_table(
    table_name="documents",
    vector_dimension=1536,
)

vectorstore = AsyncCockroachDBVectorStore(
    engine=engine,
    embeddings=OpenAIEmbeddings(),
    collection_name="documents",
)

# Use it
ids = await vectorstore.aadd_texts(["Hello world"])
results = await vectorstore.asimilarity_search("Hi", k=1)

聊天消息历史记录

在 CockroachDB 中存储对话历史记录，用于持久化、分布式聊天应用。

主要特性： - 分布式存储，自动复制 - 强一致性 (SERIALIZABLE) - 基于会话的组织方式 - 高可用性

参见 CockroachDB 聊天历史记录文档查看详细用法。

快速示例：

from langchain_cockroachdb import CockroachDBChatMessageHistory


chat_history = CockroachDBChatMessageHistory(
    session_id=str(uuid.uuid4()),
    connection_string=CONNECTION_STRING,
    table_name="chat_history",
)

from langchain.messages import HumanMessage, AIMessage

await chat_history.aadd_message(HumanMessage(content="Hello!"))
await chat_history.aadd_message(AIMessage(content="Hi there!"))

messages = await chat_history.aget_messages()

LangGraph 检查点器

在 CockroachDB 中持久化 LangGraph 工作流状态，用于短期记忆、人机交互和容错。

同步（CockroachDBSaver）和异步（AsyncCockroachDBSaver）实现均可用。

Sync

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langchain_cockroachdb import CockroachDBSaver

model = init_chat_model(model="claude-haiku-4-5-20251001")

DB_URI = os.environ["COCKROACHDB_URI"]
# Example: "cockroachdb://user:password@host:26257/defaultdb?sslmode=verify-full"
with CockroachDBSaver.from_conn_string(DB_URI) as checkpointer:
    # checkpointer.setup()

    def call_model(state: MessagesState):
        response = model.invoke(state["messages"])
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "1"}}

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    stream = graph.stream_events(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,
        version="v3",
    )
    for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

Async

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langchain_cockroachdb import AsyncCockroachDBSaver

model = init_chat_model(model="claude-haiku-4-5-20251001")

DB_URI = os.environ["COCKROACHDB_URI"]
# Example: "cockroachdb://user:password@host:26257/defaultdb?sslmode=verify-full"
async with AsyncCockroachDBSaver.from_conn_string(DB_URI) as checkpointer:
    # await checkpointer.setup()

    async def call_model(state: MessagesState):
        response = await model.ainvoke(state["messages"])
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "1"}}

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,
        version="v3",
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

    stream = await graph.astream_events(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,
        version="v3",
    )
    async for snapshot in stream.values:
        snapshot["messages"][-1].pretty_print()

行级 TTL (v0.2.1+)

使用 CockroachDB 的行级 TTL:

with CockroachDBSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    # Expire checkpoints older than 30 days, clean up daily
    checkpointer.enable_ttl(ttl_interval="30 days", cron="@daily")

    # Use the checkpointer normally -- old data is cleaned up automatically
    graph = builder.compile(checkpointer=checkpointer)

    # To disable TTL later:
    # checkpointer.disable_ttl()

异步变体： await checkpointer.aenable_ttl(ttl_interval="7 days", cron="@hourly")

性能优化 (v0.2.1+)

检查点包含多项针对低延迟读取的优化：

批量获取: list() 在 2 次批量查询中获取所有 blob 并写入，而非每个检查点 2 次
原始 BYTEA：使用 psycopg3 二进制协议，而非在 SQL 中对 blob 进行十六进制编码
预处理语句: from_conn_string() 启用服务端查询计划缓存（prepare_threshold=5)

Multi-tenancy

使用可选的命名空间列按租户隔离向量数据。启用后，所有 CRUD 和搜索操作都限定在指定的命名空间内。

CockroachDB 的 C-SPANN 索引支持前缀列，因此命名空间过滤直接使用向量索引，无需单独扫描。

from langchain_cockroachdb import AsyncCockroachDBVectorStore, CockroachDBEngine
from langchain_openai import OpenAIEmbeddings

engine = CockroachDBEngine.from_connection_string(CONNECTION_STRING)

# Create the table with a namespace column
await engine.ainit_vectorstore_table(
    table_name="documents",
    vector_dimension=1536,
    namespace_column="namespace",
)

# Create a vectorstore scoped to a specific tenant
vectorstore = AsyncCockroachDBVectorStore(
    engine=engine,
    embeddings=OpenAIEmbeddings(),
    collection_name="documents",
    namespace="tenant_a",
)

# All operations are scoped to tenant_a
ids = await vectorstore.aadd_texts(["Tenant A document"])
results = await vectorstore.asimilarity_search("query", k=5)

为什么选择 CockroachDB 用于 AI 应用？

### 设计即分布式 - **水平可扩展性**：添加节点以处理更多负载 - **多区域部署**：以低延迟为全球用户提供服务 - **自动再平衡**：数据自动分配到各节点

### 生产级可靠性 - **高可用性**：在节点、机架和数据中心故障时仍能运行 - **零停机升级**：滚动更新，无需停机 - **备份与恢复**：时间点恢复

### 大规模向量搜索 - **C-SPANN 索引**：分布式近似最近邻搜索 - **原生向量类型**：一级支持嵌入向量 - **实时索引**：新增向量无需重建 - **Multi-tenancy**：前缀列实现高效租户隔离

### PostgreSQL 兼容性 - **轻松迁移**：PostgreSQL 的直接替代品 - **熟悉的 SQL**：标准 PostgreSQL 语法 - **现有工具**：兼容 PostgreSQL 驱动和工具

安装与设置

获取 CockroachDB 连接字符串

集成

向量存储

聊天消息历史记录

LangGraph 检查点器

Sync

Async

行级 TTL (v0.2.1+)

性能优化 (v0.2.1+)

Multi-tenancy

为什么选择 CockroachDB 用于 AI 应用？

资源

支持