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 应用,也能够适配到现有数据和表之上。
该存储也可用于现有数据库中的非 RAG 用例,例如语义搜索、地理位置邻近搜索等。
该存储将根据其配置自动创建或增强 schema。如果您不希望修改 schema,请使用 disallowSchemaChanges
配置该存储。
使用 spring-boot-autoconfigure 时,根据 Spring Boot 标准,disallowSchemaChanges
默认设置为 true,您必须在 application.properties
文件中设置 …initialize-schema=true
来选择启用 schema 的创建/修改。
什么是 JVector?
JVector 是一个纯 Java 嵌入式向量搜索引擎。
它在其他 HNSW 向量相似性搜索实现中脱颖而出,原因在于其具备以下特性:
-
算法快速。JVector 使用受 DiskANN 及相关研究启发的先进图算法,提供高召回率和低延迟。
-
实现快速。JVector 使用 Panama SIMD API 加速索引构建和查询。
-
内存高效。JVector 使用乘积量化压缩向量,使其在搜索期间能够驻留在内存中。
-
磁盘感知。JVector 的磁盘布局设计旨在查询时执行最少必要的 IOPS。
-
并发。索引构建可以线性扩展到至少 32 个线程。线程加倍,构建时间减半。
-
增量。在构建索引的同时即可查询。添加向量后,可以立即在搜索结果中找到,没有延迟。
-
易于嵌入。API 由在生产环境中使用它的人员设计,易于嵌入。
先决条件
-
一个用于计算文档嵌入的
EmbeddingModel
实例。这通常被配置为一个 Spring Bean。有多种选项可用:-
Transformers Embedding - 在您的本地环境中计算嵌入。默认是通过 ONNX 和 all-MiniLM-L6-v2 Sentence Transformers。这开箱即用。
-
如果您想使用 OpenAI 的 Embedding - 使用 OpenAI 的嵌入端点。您需要在 OpenAI 注册页面创建账户,并在 API Keys 页面生成 API 密钥令牌。
-
还有更多选择,请参阅
Embeddings API
文档。
-
-
一个 Apache Cassandra 实例,版本 5.0-beta1 或更高
依赖
Spring AI 自动配置、starter 模块的 artifact 名称发生了重大变化。请参阅升级说明获取更多信息。 |
对于依赖管理,我们建议使用 Spring AI BOM,具体说明请参阅依赖管理部分。 |
将这些依赖添加到您的项目
-
仅使用 Cassandra Vector Store
<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 向量存储。
属性 | 默认值 |
---|---|
|
springframework |
|
ai_vector_store |
|
false |
|
|
|
content |
|
embedding |
|
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.query("search text")
.withTopK(5)
.withSimilarityThreshold(0.7f)
.withFilterExpression("metadata.key1 == 'value1'")
);
高级配置
对于更复杂的用例,您可以在 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();
}
连接配置
有两种方法配置与 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 上使用该存储。这里我们使用来自 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
然后使用构建器模式配置存储
@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();
}
加载完整的维基百科数据集
加载完整的维基百科数据集
-
从 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-*/
|
访问原生客户端
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 特有功能和操作。