AsyncCockroachDBVectorStore 是一个使用 CockroachDB 分布式 SQL 数据库（支持原生向量功能）实现的 LangChain 向量存储。

本笔记本介绍如何使用 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

设置您的连接值

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

# For local insecure cluster
CONNECTION_STRING = "cockroachdb://root@localhost:26257/defaultdb?sslmode=disable"

TABLE_NAME = "langchain_vectors"
VECTOR_DIMENSION = 1536  # Depends on your embedding model

初始化

创建连接引擎

该 CockroachDBEngine 管理到集群的连接池：

from langchain_cockroachdb import CockroachDBEngine

engine = CockroachDBEngine.from_connection_string(
    url=CONNECTION_STRING,
    pool_size=10,        # Connection pool size
    max_overflow=20,     # Additional connections allowed
    pool_pre_ping=True,  # Health check connections
)

初始化表

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

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

创建嵌入实例

使用任意 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,
)

管理向量存储

添加文档

添加带有元数据的文档：

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)

删除文档

按 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,           # Number of results to return
    fetch_k=20,    # Number of candidates to consider
    lambda_mult=0.5,  # 0 = max diversity, 1 = max relevance
)

向量索引

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

什么是 C-SPANN？

C-SPANN（CockroachDB 空间分区近似最近邻）是一个分布式向量索引，具有以下特点：

• - 自动分片到集群节点上
• - 在大规模场景下提供亚秒级查询性能
• - 支持余弦、欧几里得（L2）和内积距离
• - 支持多租户架构的前缀列

创建向量索引

from langchain_cockroachdb import CSPANNIndex, DistanceStrategy

# Create a cosine distance index (most common)
index = CSPANNIndex(
    distance_strategy=DistanceStrategy.COSINE,
    name="my_vector_index",
)

await vectorstore.aapply_vector_index(index)

距离策略

选择适合您用例的距离度量：

# Cosine similarity (most common for text embeddings)
CSPANNIndex(distance_strategy=DistanceStrategy.COSINE)

# Euclidean distance (L2)
CSPANNIndex(distance_strategy=DistanceStrategy.EUCLIDEAN)

# Inner product (for normalized vectors)
CSPANNIndex(distance_strategy=DistanceStrategy.INNER_PRODUCT)

调整索引参数

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

index = CSPANNIndex(
    distance_strategy=DistanceStrategy.COSINE,
    min_partition_size=16,   # Minimum vectors per partition
    max_partition_size=128,  # Maximum vectors per partition
)

await vectorstore.aapply_vector_index(index)

查询时调优

在查询时调整搜索参数：

from langchain_cockroachdb import CSPANNQueryOptions

# Increase beam size for better recall (slower)
query_options = CSPANNQueryOptions(beam_size=200)  # Default: 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)

元数据过滤

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

支持的运算符

过滤示例

# Simple equality
docs = await vectorstore.asimilarity_search(
    query,
    filter={"category": "news"},
)

# Numeric comparison
docs = await vectorstore.asimilarity_search(
    query,
    filter={"year": {"$gte": 2020}},
)

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

同步接口

所有异步方法都有使用同步包装器的同步等价方法：

from langchain_cockroachdb import CockroachDBVectorStore

# Create sync vectorstore
vectorstore = CockroachDBVectorStore(
    engine=engine,
    embeddings=embeddings,
    collection_name=TABLE_NAME,
)

# Use sync methods
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 仓库
• - PyPI 包

其他资源

• - CockroachDB 向量索引文档
• - CockroachDB Cloud