跳到主要内容

Apache Cassandra 向量存储

DeepSeek V3 中英对照 Apache Cassandra Vector Store

本节将引导您设置 CassandraVectorStore 以存储文档嵌入并执行相似性搜索。

什么是 Apache Cassandra？​

Apache Cassandra® 是一个真正的开源分布式数据库，以其线性可扩展性、经过验证的容错能力和低延迟而闻名，使其成为关键任务事务数据的理想平台。

其向量相似性搜索（Vector Similarity Search, VSS）基于 JVector 库，确保了一流的性能和相关性。

在 Apache Cassandra 中，向量搜索可以简单地通过以下方式完成：

SELECT content FROM table ORDER BY content_vector ANN OF query_embedding;
sql

更多关于此的文档可以在这里阅读。

这个 Spring AI 向量存储旨在既适用于全新的 RAG 应用程序，也能够对现有数据和表进行改造。

该存储也可以用于现有数据库中的非 RAG 场景，例如语义搜索、地理邻近搜索等。

存储将根据其配置自动创建或增强所需的 schema。如果你不希望进行 schema 修改，可以使用 disallowSchemaChanges 配置存储。

在使用 spring-boot-autoconfigure 时，根据 Spring Boot 的标准，disallowSchemaChanges 默认为 true，你必须通过在 application.properties 文件中设置 …​initialize-schema=true 来选择启用模式创建/修改。

什么是 JVector？​

JVector 是一个纯 Java 嵌入式向量搜索引擎。

它之所以在众多 HNSW 向量相似性搜索实现中脱颖而出，是因为：

  • 算法快速。JVector 使用了受 DiskANN 及相关研究启发的最先进的图算法，提供了高召回率和低延迟。

  • 实现快速。JVector 使用 Panama SIMD API 来加速索引构建和查询。

  • 内存高效。JVector 使用乘积量化压缩向量，使它们能够在搜索期间保持在内存中。

  • 磁盘感知。JVector 的磁盘布局设计为在查询时执行最少的必要 IOPS。

  • 并发。索引构建至少可以线性扩展到 32 个线程。线程数翻倍，构建时间减半。

  • 增量式。在构建索引的同时进行查询。添加向量后，无需延迟即可在搜索结果中找到它。

  • 易于嵌入。API 设计易于嵌入，适用于生产环境中使用它的人员。

先决条件​

  1. 一个 EmbeddingModel 实例，用于计算文档嵌入。这通常配置为 Spring Bean。有几种可用的选项：

    • Transformers Embedding - 在本地环境中计算嵌入。默认是通过 ONNX 和 all-MiniLM-L6-v2 Sentence Transformers 实现的。这种方式可以直接使用。

    • 如果你想使用 OpenAI 的 Embeddings - 使用 OpenAI 的嵌入端点。你需要在 OpenAI 注册 创建一个账户，并在 API Keys 生成 api-key 令牌。

    • 还有更多选择，请参阅 Embeddings API 文档。

  2. 一个 Apache Cassandra 实例，从 5.0-beta1 版本开始

    1. DIY 快速入门

    2. 对于托管服务，Astra DB 提供了一个不错的免费层级服务。

依赖项​

提示

对于依赖管理，我们建议使用 Spring AI BOM，具体内容请参考依赖管理部分。

将以下依赖项添加到你的项目中：

  • 仅针对 Cassandra 向量存储：
<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-cassandra-store</artifactId>
</dependency>
xml
  • 或者，如果您需要一个 RAG 应用程序所需的所有内容（使用默认的 ONNX Embedding 模型）：
<dependency>
    <groupId>org.springframework.ai</groupId>
    <artifactId>spring-ai-cassandra-store-spring-boot-starter</artifactId>
</dependency>
xml

配置属性​

你可以在 Spring Boot 配置中使用以下属性来自定义 Apache Cassandra 向量存储。

属性默认值
spring.ai.vectorstore.cassandra.keyspacespringframework
spring.ai.vectorstore.cassandra.tableai_vector_store
spring.ai.vectorstore.cassandra.initialize-schemafalse
spring.ai.vectorstore.cassandra.index-name
spring.ai.vectorstore.cassandra.content-column-namecontent
spring.ai.vectorstore.cassandra.embedding-column-nameembedding
spring.ai.vectorstore.cassandra.fixed-thread-pool-executor-size16

用法​

基本用法​

创建一个 CassandraVectorStore 实例作为 Spring Bean：

@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
    return CassandraVectorStore.builder(embeddingModel)
        .session(session)
        .keyspace("my_keyspace")
        .table("my_vectors")
        .build();
}
java

一旦你有了向量存储实例，你就可以添加文档并执行搜索：

// Add documents
vectorStore.add(List.of(
    new Document("1", "content1", Map.of("key1", "value1")),
    new Document("2", "content2", Map.of("key2", "value2"))
));

// Search with filters
List<Document> results = vectorStore.similaritySearch(
    SearchRequest.query("search text")
        .withTopK(5)
        .withSimilarityThreshold(0.7f)
        .withFilterExpression("metadata.key1 == 'value1'")
);
java

高级配置​

对于更复杂的用例，您可以在 Spring Bean 中配置额外的设置：

@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
    return CassandraVectorStore.builder(embeddingModel)
        .session(session)
        .keyspace("my_keyspace")
        .table("my_vectors")
        // Configure primary keys
        .partitionKeys(List.of(
            new SchemaColumn("id", DataTypes.TEXT),
            new SchemaColumn("category", DataTypes.TEXT)
        ))
        .clusteringKeys(List.of(
            new SchemaColumn("timestamp", DataTypes.TIMESTAMP)
        ))
        // Add metadata columns with optional indexing
        .addMetadataColumns(
            new SchemaColumn("category", DataTypes.TEXT, SchemaColumnTags.INDEXED),
            new SchemaColumn("score", DataTypes.DOUBLE)
        )
        // Customize column names
        .contentColumnName("text")
        .embeddingColumnName("vector")
        // Performance tuning
        .fixedThreadPoolExecutorSize(32)
        // Schema management
        .disallowSchemaChanges(false)
        // Custom batching strategy
        .batchingStrategy(new TokenCountBatchingStrategy())
        .build();
}
java

连接配置​

有两种方法可以配置与 Cassandra 的连接：

  • 使用注入的 CqlSession（推荐）：
@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
    return CassandraVectorStore.builder(embeddingModel)
        .session(session)
        .keyspace("my_keyspace")
        .table("my_vectors")
        .build();
}
java
  • 在构建器中直接使用连接详细信息：
@Bean
public VectorStore vectorStore(EmbeddingModel embeddingModel) {
    return CassandraVectorStore.builder(embeddingModel)
        .contactPoint(new InetSocketAddress("localhost", 9042))
        .localDatacenter("datacenter1")
        .keyspace("my_keyspace")
        .build();
}
java

元数据过滤​

你可以利用 CassandraVectorStore 的通用、可移植的元数据过滤器。要使元数据列可搜索，它们必须是主键或 SAI 索引。要使非主键列被索引，需要使用 SchemaColumnTags.INDEXED 配置元数据列。

例如，你可以使用文本表达式语言：

vectorStore.similaritySearch(
    SearchRequest.builder().query("The World")
        .topK(5)
        .filterExpression("country in ['UK', 'NL'] && year >= 2020").build());
java

或者以编程方式使用表达式 DSL：

Filter.Expression f = new FilterExpressionBuilder()
    .and(
        f.in("country", "UK", "NL"),
        f.gte("year", 2020)
    ).build();

vectorStore.similaritySearch(
    SearchRequest.builder().query("The World")
        .topK(5)
        .filterExpression(f).build());
java

便携式过滤表达式会自动转换为 CQL 查询。

高级示例：基于 Wikipedia 数据集的向量存储​

以下示例演示了如何在现有 schema 上使用 store。这里我们使用了来自 github.com/datastax-labs/colbert-wikipedia-data 项目的 schema，该项目附带了一个完整的维基百科数据集，并已为您进行了向量化处理。

首先，在 Cassandra 数据库中创建 schema：

wget https://s.apache.org/colbert-wikipedia-schema-cql -O colbert-wikipedia-schema.cql
cqlsh -f colbert-wikipedia-schema.cql
bash

然后使用构建器模式配置 store：

@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
    List<SchemaColumn> partitionColumns = List.of(
        new SchemaColumn("wiki", DataTypes.TEXT),
        new SchemaColumn("language", DataTypes.TEXT),
        new SchemaColumn("title", DataTypes.TEXT)
    );

    List<SchemaColumn> clusteringColumns = List.of(
        new SchemaColumn("chunk_no", DataTypes.INT),
        new SchemaColumn("bert_embedding_no", DataTypes.INT)
    );

    List<SchemaColumn> extraColumns = List.of(
        new SchemaColumn("revision", DataTypes.INT),
        new SchemaColumn("id", DataTypes.INT)
    );

    return CassandraVectorStore.builder()
        .session(session)
        .embeddingModel(embeddingModel)
        .keyspace("wikidata")
        .table("articles")
        .partitionKeys(partitionColumns)
        .clusteringKeys(clusteringColumns)
        .contentColumnName("body")
        .embeddingColumnName("all_minilm_l6_v2_embedding")
        .indexName("all_minilm_l6_v2_ann")
        .disallowSchemaChanges(true)
        .addMetadataColumns(extraColumns)
        .primaryKeyTranslator((List<Object> primaryKeys) -> {
            if (primaryKeys.isEmpty()) {
                return "test§¶0";
            }
            return String.format("%s§¶%s", primaryKeys.get(2), primaryKeys.get(3));
        })
        .documentIdTranslator((id) -> {
            String[] parts = id.split("§¶");
            String title = parts[0];
            int chunk_no = parts.length > 1 ? Integer.parseInt(parts[1]) : 0;
            return List.of("simplewiki", "en", title, chunk_no, 0);
        })
        .build();
}

@Bean
public EmbeddingModel embeddingModel() {
    // default is ONNX all-MiniLM-L6-v2 which is what we want
    return new TransformersEmbeddingModel();
}
java

加载完整的维基百科数据集​

要加载完整的维基百科数据集：

  1. 从 s.apache.org/simplewiki-sstable-tar 下载 simplewiki-sstable.tar（这将花费一些时间，文件大小为几十 GB）

  2. 加载数据：

tar -xf simplewiki-sstable.tar -C ${CASSANDRA_DATA}/data/wikidata/articles-*/
nodetool import wikidata articles ${CASSANDRA_DATA}/data/wikidata/articles-*/
备注

  • 如果此表中已有数据，请确保在执行 tar 时，tarball 中的文件不会覆盖现有的 sstables。

  • 除了使用 nodetool import，你也可以选择直接重启 Cassandra。

  • 如果索引中存在任何故障，它们将自动重建。

访问 Native Client​

Cassandra Vector Store 实现通过 getNativeClient() 方法提供了对底层原生 Cassandra 客户端（CqlSession）的访问：

CassandraVectorStore vectorStore = context.getBean(CassandraVectorStore.class);
Optional<CqlSession> nativeClient = vectorStore.getNativeClient();

if (nativeClient.isPresent()) {
    CqlSession session = nativeClient.get();
    // Use the native client for Cassandra-specific operations
}
java

原生客户端使您能够访问可能未通过 VectorStore 接口公开的 Cassandra 特定功能和操作。