CockroachDB 向量存储 - LangChain LangGraph DeepAgent中文文档

AsyncCockroachDBVectorStore 是基于 CockroachDB 分布式 SQL 数据库（具备原生向量支持）的 LangChain 向量存储实现。本 notebook 介绍如何使用 AsyncCockroachDBVectorStore API。代码位于集成包中：langchain-cockroachdb。

概览

CockroachDB 是一款分布式 SQL 数据库，提供以下功能：

原生向量支持，具备 VECTOR 数据类型（v24.2+）
分布式 C-SPANN 索引，用于近似最近邻（ANN）搜索（v25.2+）
默认的 SERIALIZABLE 隔离级别，保证事务正确性
水平可扩展性，支持自动分片和复制
PostgreSQL 兼容协议，便于迁移采用

向量工作负载的核心优势

分布式向量索引：C-SPANN 索引自动分布在集群各节点
多租户支持：索引中的前缀列可高效隔离租户
强一致性：SERIALIZABLE 事务防止数据异常
高可用性：自动故障转移，零数据丢失

设置

安装

安装集成库 langchain-cockroachdb。

pip install -qU langchain-cockroachdb

CockroachDB 集群

您需要支持向量功能的 CockroachDB 集群（v24.2+）。请选择以下方式之一：

方式一：CockroachDB Cloud（推荐）

在 cockroachlabs.cloud 注册账号
创建免费集群
从集群详情页面获取连接字符串

方式二：Docker（开发环境）

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

方式三：本地二进制文件

从 cockroachlabs.com/docs/releases 下载

cockroach start-single-node --insecure --listen-addr=localhost:26257

设置连接参数

# CockroachDB Cloud
CONNECTION_STRING = "cockroachdb://user:password@host:26257/database?sslmode=verify-full"

# 本地非安全集群
CONNECTION_STRING = "cockroachdb://root@localhost:26257/defaultdb?sslmode=disable"

TABLE_NAME = "langchain_vectors"
VECTOR_DIMENSION = 1536  # 取决于您使用的嵌入模型

初始化

创建连接引擎

CockroachDBEngine 管理到集群的连接池：

from langchain_cockroachdb import CockroachDBEngine

engine = CockroachDBEngine.from_connection_string(
    url=CONNECTION_STRING,
    pool_size=10,        # 连接池大小
    max_overflow=20,     # 允许的额外连接数
    pool_pre_ping=True,  # 对连接进行健康检查
)

初始化表

创建具有向量存储所需 schema 的表：

await engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    vector_dimension=VECTOR_DIMENSION,
)

可选：指定 schema 名称

await engine.ainit_vectorstore_table(
    table_name=TABLE_NAME,
    vector_dimension=VECTOR_DIMENSION,
    schema="my_schema",  # 默认："public"
)

创建嵌入实例

使用任意 LangChain 嵌入模型。

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

初始化向量存储

from langchain_cockroachdb import AsyncCockroachDBVectorStore

vectorstore = AsyncCockroachDBVectorStore(
    engine=engine,
    embeddings=embeddings,
    collection_name=TABLE_NAME,
)

管理向量存储

添加文档

添加带元数据的文档：

import uuid
from langchain_core.documents import Document

docs = [
    Document(
        id=str(uuid.uuid4()),
        page_content="CockroachDB is a distributed SQL database",
        metadata={"source": "docs", "category": "database"},
    ),
    Document(
        id=str(uuid.uuid4()),
        page_content="Vector search enables semantic similarity",
        metadata={"source": "docs", "category": "features"},
    ),
]

ids = await vectorstore.aadd_documents(docs)

添加文本

直接添加文本，无需构造文档结构：

texts = ["First text", "Second text", "Third text"]
metadatas = [{"idx": i} for i in range(len(texts))]
ids = [str(uuid.uuid4()) for _ in texts]

ids = await vectorstore.aadd_texts(texts, metadatas=metadatas, ids=ids)

性能说明：CockroachDB 的向量索引在较小批次时性能最佳。默认的 batch_size=100 已针对向量插入进行优化。大批量插入 VECTOR 类型数据可能导致性能下降。

删除文档

按 ID 删除文档：

await vectorstore.adelete([ids[0], ids[1]])

查询向量存储

相似度搜索

使用自然语言搜索相似文档：

query = "distributed database"
docs = await vectorstore.asimilarity_search(query, k=5)

for doc in docs:
    print(f"{doc.page_content[:50]}...")

带分数的相似度搜索

获取带相关性分数的结果：

docs_with_scores = await vectorstore.asimilarity_search_with_score(query, k=5)

for doc, score in docs_with_scores:
    print(f"Score: {score:.4f} - {doc.page_content[:50]}...")

按向量搜索

使用预计算的嵌入向量进行搜索：

query_vector = await embeddings.aembed_query(query)
docs = await vectorstore.asimilarity_search_by_vector(query_vector, k=5)

最大边际相关性（MMR）搜索

检索在相关性和多样性之间取得平衡的多样化结果：

docs = await vectorstore.amax_marginal_relevance_search(
    query,
    k=5,           # 返回的结果数量
    fetch_k=20,    # 候选考虑数量
    lambda_mult=0.5,  # 0 = 最大多样性，1 = 最大相关性
)

向量索引

使用 CockroachDB 的 C-SPANN 向量索引加速相似度搜索（需要 v25.2+）。

什么是 C-SPANN？

C-SPANN（CockroachDB Space Partition Approximate Nearest Neighbor，空间分区近似最近邻）是一种分布式向量索引，具备以下特点：

自动在集群节点间分片
大规模场景下实现亚秒级查询性能
支持余弦、欧氏（L2）和内积距离
支持前缀列，适用于多租户架构

创建向量索引

from langchain_cockroachdb import CSPANNIndex, DistanceStrategy

# 创建余弦距离索引（最常用）
index = CSPANNIndex(
    distance_strategy=DistanceStrategy.COSINE,
    name="my_vector_index",
)

await vectorstore.aapply_vector_index(index)

距离策略

选择与您的使用场景匹配的距离度量：

# 余弦相似度（文本嵌入最常用）
CSPANNIndex(distance_strategy=DistanceStrategy.COSINE)

# 欧氏距离（L2）
CSPANNIndex(distance_strategy=DistanceStrategy.EUCLIDEAN)

# 内积（适用于归一化向量）
CSPANNIndex(distance_strategy=DistanceStrategy.INNER_PRODUCT)

调整索引参数

调整分区大小以优化性能：

index = CSPANNIndex(
    distance_strategy=DistanceStrategy.COSINE,
    min_partition_size=16,   # 每个分区的最小向量数
    max_partition_size=128,  # 每个分区的最大向量数
)

await vectorstore.aapply_vector_index(index)

查询时调优

在查询时调整搜索参数：

from langchain_cockroachdb import CSPANNQueryOptions

# 增大 beam size 以提高召回率（速度变慢）
query_options = CSPANNQueryOptions(beam_size=200)  # 默认：100

docs = await vectorstore.asimilarity_search(
    query,
    k=10,
    query_options=query_options,
)

删除索引

删除向量索引：

index = CSPANNIndex(name="my_vector_index")
await vectorstore.adrop_vector_index(index)

元数据过滤

使用元数据字段过滤相似度搜索。

支持的运算符

运算符	含义	示例
`$eq`	等于	`{"category": "news"}`
`$ne`	不等于	`{"category": {"$ne": "spam"}}`
`$gt`	大于	`{"year": {"$gt": 2020}}`
`$gte`	大于等于	`{"rating": {"$gte": 4.0}}`
`$lt`	小于	`{"year": {"$lt": 2023}}`
`$lte`	小于等于	`{"rating": {"$lte": 3.0}}`
`$in`	在列表中	`{"category": {"$in": ["news", "blog"]}}`
`$nin`	不在列表中	`{"source": {"$nin": ["spam", "test"]}}`
`$between`	在范围内	`{"year": {"$between": [2020, 2023]}}`
`$like`	模式匹配	`{"source": {"$like": "wiki%"}}`
`$ilike`	不区分大小写匹配	`{"category": {"$ilike": "%NEWS%"}}`
`$and`	逻辑与	`{"$and": [{...}, {...}]}`
`$or`	逻辑或	`{"$or": [{...}, {...}]}`

过滤示例

# 简单等值过滤
docs = await vectorstore.asimilarity_search(
    query,
    filter={"category": "news"},
)

# 数值比较
docs = await vectorstore.asimilarity_search(
    query,
    filter={"year": {"$gte": 2020}},
)

# 复合过滤
docs = await vectorstore.asimilarity_search(
    query,
    filter={
        "$and": [
            {"category": {"$in": ["news", "blog"]}},
            {"year": {"$gte": 2020}},
            {"rating": {"$gt": 3.5}},
        ]
    },
)

同步接口

所有异步方法都有对应的同步版本，通过同步封装器使用：

from langchain_cockroachdb import CockroachDBVectorStore

# 创建同步向量存储
vectorstore = CockroachDBVectorStore(
    engine=engine,
    embeddings=embeddings,
    collection_name=TABLE_NAME,
)

# 使用同步方法
docs = vectorstore.similarity_search(query, k=5)
ids = vectorstore.add_documents(docs)
vectorstore.apply_vector_index(index)

用于检索增强生成（RAG）

有关以 CockroachDB 作为向量存储实现 RAG 的内容，请参阅 LangChain RAG 教程。CockroachDB 向量存储可替代这些模式中的任何其他向量存储。

清理

⚠️ 此操作不可撤销

删除向量存储表：

await engine.adrop_table(TABLE_NAME)

API 参考

有关所有功能和配置的详细文档：

其他资源

在 GitHub 上编辑此页面或提交问题。

通过 MCP 将这些文档连接到 Claude、VSCode 等，获取实时解答。

Popular Providers

Integrations by component

​概览

​向量工作负载的核心优势

​设置

​安装

​CockroachDB 集群

​方式一：CockroachDB Cloud（推荐）

​方式二：Docker（开发环境）

​方式三：本地二进制文件

​设置连接参数

​初始化

​创建连接引擎

​初始化表

​创建嵌入实例

​初始化向量存储

​管理向量存储

​添加文档

​添加文本

​删除文档

​查询向量存储

​相似度搜索

​带分数的相似度搜索

​按向量搜索

​最大边际相关性（MMR）搜索

​向量索引

​什么是 C-SPANN？

​创建向量索引

​距离策略

​调整索引参数

​查询时调优

​删除索引

​元数据过滤

​支持的运算符

​过滤示例

​同步接口

​用于检索增强生成（RAG）

​清理

​API 参考

​其他资源

概览

向量工作负载的核心优势

设置

安装

CockroachDB 集群

方式一：CockroachDB Cloud（推荐）

方式二：Docker（开发环境）

方式三：本地二进制文件

设置连接参数

初始化

创建连接引擎

初始化表

创建嵌入实例

初始化向量存储

管理向量存储

添加文档

添加文本

删除文档

查询向量存储

相似度搜索

带分数的相似度搜索

按向量搜索

最大边际相关性（MMR）搜索

向量索引

什么是 C-SPANN？

创建向量索引

距离策略

调整索引参数

查询时调优

删除索引

元数据过滤

支持的运算符

过滤示例

同步接口

用于检索增强生成（RAG）

清理

API 参考

其他资源