Apache Cassandra 向量存储
本节将指导你配置 CassandraVectorStore，以存储文档嵌入并执行相似度搜索。
什么是 Apache Cassandra？
Apache Cassandra® 是一款真正的开源分布式数据库，以其线性可扩展性、经过验证的容错能力以及低延迟而著称，是处理关键事务数据的理想平台。
其向量相似度搜索（VSS）基于 JVector 库，可确保业界领先的性能和相关性。
在 Apache Cassandra 中，向量搜索可以简单地通过以下方式完成：
SELECT content FROM table ORDER BY content_vector ANN OF query_embedding;
有关此内容的更多文档，可以在此处阅读。
该 Spring AI 向量存储旨在同时适用于全新的 RAG 应用，并能够集成到现有的数据和表之上。
Store 也可用于现有数据库中的非 RAG 场景，例如语义搜索、地理邻近搜索等。
store 会根据其配置自动创建或增强 schema。如果你不希望修改 schema，请配置 store 使用 initializeSchema。
使用 spring-boot-autoconfigure 时，按照 Spring Boot 的标准，initializeSchema 默认为 false，您需要在 application.properties 文件中设置 …initialize-schema=true 以启用 schema 创建/修改。
什么是 JVector？
JVector 是一个纯 Java 编写的嵌入式向量搜索引擎。
与其他 HNSW 向量相似性搜索实现不同，它的特点在于：
-
算法快速。JVector 使用受 DiskANN 及相关研究启发的先进图算法，提供高召回率和低延迟。
-
实现快速。JVector 使用 Panama SIMD API 来加速索引构建和查询。
-
内存高效。JVector 使用乘积量化压缩向量，使它们在搜索期间可以保留在内存中。
-
磁盘感知。JVector 的磁盘布局旨在查询时执行最低必要的 IOPS。
-
并发。索引构建至少可线性扩展到 32 个线程。线程数量翻倍，构建时间减半。
-
增量式。在构建过程中即可查询索引。添加向量与在搜索结果中找到它之间没有延迟。
-
易于集成。API 专为易于集成而设计，由在生产环境中使用它的人员提供建议。
前置条件
-
一个
EmbeddingModel实例，用于计算文档嵌入（Embeddings）。这通常被配置为 Spring Bean。有多种可选方案：Transformers Embedding- 在您的本地环境中计算嵌入。默认通过 ONNX 和 all-MiniLM-L6-v2 Sentence Transformers 实现。即开即用。- 如果您想使用 OpenAI 的 Embeddings - 使用 OpenAI 的嵌入端点。您需要前往 OpenAI Signup 创建账户，并在 API Keys 处生成 API Key 令牌。
- 还有更多选择，请参阅
Embeddings API文档。
-
一个 Apache Cassandra 实例，版本为 5.0-beta1
- DIY Quick Start
- 对于托管服务，Astra DB 提供丰富的免费套餐。
依赖项
Spring AI 的自动配置以及 starter 模块的 artifact 名称发生了重大变更。更多信息请参阅 升级说明。
对于依赖管理，我们建议按照 依赖管理 章节中说明的那样使用 Spring AI BOM。
在你的项目中添加以下依赖项：
- 仅针对 Cassandra 向量存储：
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-cassandra-store</artifactId>
</dependency>
- 或者，对于 RAG 应用程序所需的一切（使用默认的 ONNX Embedding Model）：
<dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-starter-vector-store-cassandra</artifactId>
</dependency>
配置属性
您可以使用以下属性在 Spring Boot 配置中自定义 Apache Cassandra 向量存储。
| 属性 | 默认值 |
|---|---|
spring.ai.vectorstore.cassandra.keyspace | springframework |
spring.ai.vectorstore.cassandra.table | ai_vector_store |
spring.ai.vectorstore.cassandra.initialize-schema | false |
spring.ai.vectorstore.cassandra.index-name | |
spring.ai.vectorstore.cassandra.content-column-name | content |
spring.ai.vectorstore.cassandra.embedding-column-name | embedding |
spring.ai.vectorstore.cassandra.fixed-thread-pool-executor-size | 16 |
用法
基本用法
创建一个 CassandraVectorStore 实例并作为 Spring Bean：
@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
return CassandraVectorStore.builder(embeddingModel)
.session(session)
.keyspace("my_keyspace")
.table("my_vectors")
.build();
}
获取向量存储实例后，你可以添加文档并执行搜索：
// 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.builder().query("search text")
.topK(5)
.similarityThreshold(0.7)
.filterExpression("metadata.key1 == 'value1'")
.build()
);
高级配置
对于更复杂的使用场景，你可以在 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
.initializeSchema(true)
// Custom batching strategy
.batchingStrategy(new TokenCountBatchingStrategy())
.build();
}
连接配置
配置与 Cassandra 的连接有两种方式：
- 使用注入的 CqlSession（推荐）：
@Bean
public VectorStore vectorStore(CqlSession session, EmbeddingModel embeddingModel) {
return CassandraVectorStore.builder(embeddingModel)
.session(session)
.keyspace("my_keyspace")
.table("my_vectors")
.build();
}
- 在构建器中直接使用连接详情：
@Bean
public VectorStore vectorStore(EmbeddingModel embeddingModel) {
return CassandraVectorStore.builder(embeddingModel)
.contactPoint(new InetSocketAddress("localhost", 9042))
.localDatacenter("datacenter1")
.keyspace("my_keyspace")
.build();
}
元数据筛选
您可以利用通用且可移植的元数据过滤器与 CassandraVectorStore 配合使用。为了让元数据列可被搜索，它们必须是主键或具有 SAI 索引。若要对非主键列建立索引，请使用 SchemaColumnTags.INDEXED 配置该元数据列。
例如，您可以使用文本表达式语言：
vectorStore.similaritySearch(
SearchRequest.builder().query("The World")
.topK(5)
.filterExpression("country in ['UK', 'NL'] && year >= 2020").build());
或通过使用表达式 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());
可移植的过滤表达式会自动转换为 CQL 查询。
进阶示例：基于维基百科数据集的向量存储
以下示例展示了如何在现有 schema 上使用 store。这里我们使用了来自 github.com/datastax-labs/colbert-wikipedia-data 项目的 schema，该项目附带了完整的维基百科数据集，并且已预先完成向量化处理。
首先，在 Cassandra 数据库中创建模式：
wget https://s.apache.org/colbert-wikipedia-schema-cql -O colbert-wikipedia-schema.cql
cqlsh -f colbert-wikipedia-schema.cql
然后使用构建者模式配置存储：
@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")
.initializeSchema(false)
.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();
}
加载完整的维基百科数据集
要加载完整的维基百科数据集：
-
从 s.apache.org/simplewiki-sstable-tar 下载
simplewiki-sstable.tar（这可能需要一些时间，文件大小为数十 GB） -
加载数据：
tar -xf simplewiki-sstable.tar -C ${CASSANDRA_DATA}/data/wikidata/articles-*/
nodetool import wikidata articles ${CASSANDRA_DATA}/data/wikidata/articles-*/
-
如果该表中已有数据，执行
tar时请检查 tar 包中的文件是否会覆盖现有的 sstable。 -
nodetool import的一种替代方法是直接重启 Cassandra。 -
如果索引中存在任何失败，它们将自动重建。
访问原生客户端
Cassandra 向量存储实现通过 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
}
原生客户端允许你访问可能未通过 VectorStore 接口暴露的 Cassandra 特定功能和操作。