Skip to main content
AsyncCockroachDBVectorStore 是一个 LangChain 向量存储的实现，它使用 CockroachDB 的分布式 SQL 数据库及其原生向量支持。 本笔记本介绍如何使用 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+）。选择以下选项之一：

​
选项 1：CockroachDB Cloud（推荐）

  1. 在 cockroachlabs.cloud 注册
  2. 创建一个免费集群
  3. 从集群详细信息页面获取连接字符串

​
选项 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 空间分区近似最近邻）是一个分布式向量索引，它：
  • 自动在集群节点间分片
  • 在大规模下提供亚秒级查询性能
  • 支持余弦、欧几里得（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 参考

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

​
其他资源

在 GitHub 上编辑此页面 或 提交问题。
通过 MCP 将这些文档连接到 Claude、VSCode 等 以获取实时答案。