CockroachDB 向量存储 - Docs by LangChain

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

概述

CockroachDB 是一个分布式 SQL 数据库，提供：

原生向量支持，使用 VECTOR 数据类型（v24.2+）
分布式 C-SPANN 索引，用于近似最近邻（ANN）搜索（v25.2+）
默认可序列化隔离，确保事务正确性
水平可扩展性，支持自动分片和复制
PostgreSQL 线路兼容，便于采用

向量工作负载的关键优势

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

设置

安装

安装集成库 langchain-cockroachdb。

pip install -qU langchain-cockroachdb

CockroachDB 集群

您需要一个支持向量的 CockroachDB 集群（v24.2+）。选择一个选项：

选项 1：CockroachDB Cloud（推荐）

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

选项 2：Docker（开发环境）

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

选项 3：本地二进制文件

从 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,  # 健康检查连接
)

初始化表

创建具有适当模式的表用于向量存储：

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

可选：指定模式名称

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

# 增加波束大小以提高召回率（较慢）
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 参考

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

附加资源

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

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

​概述

​向量工作负载的关键优势

​设置

​安装

​CockroachDB 集群

​选项 1：CockroachDB Cloud（推荐）

​选项 2：Docker（开发环境）

​选项 3：本地二进制文件

​设置连接值

​初始化

​创建连接引擎

​初始化表

​创建嵌入实例

​初始化向量存储

​管理向量存储

​添加文档

​添加文本

​删除文档

​查询向量存储

​相似性搜索

​带分数的相似性搜索

​按向量搜索

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

​向量索引

​什么是 C-SPANN？

​创建向量索引

​距离策略

​调整索引参数

​查询时调整

​删除索引

​元数据过滤

​支持的运算符

​过滤示例

​同步接口

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

​清理

​API 参考

​附加资源

概述

向量工作负载的关键优势

设置

安装

CockroachDB 集群

选项 1：CockroachDB Cloud（推荐）

选项 2：Docker（开发环境）

选项 3：本地二进制文件

设置连接值

初始化

创建连接引擎

初始化表

创建嵌入实例

初始化向量存储

管理向量存储

添加文档

添加文本

删除文档

查询向量存储

相似性搜索

带分数的相似性搜索

按向量搜索

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

向量索引

什么是 C-SPANN？

创建向量索引

距离策略

调整索引参数

查询时调整

删除索引

元数据过滤

支持的运算符

过滤示例

同步接口

用于检索增强生成（RAG）的用法

清理

API 参考

附加资源