LlamaIndex 集成 Astra DB 向量存储：AstraDBVectorStore 完整实战指南

LlamaIndex 集成 Astra DB 向量存储：AstraDBVectorStore 完整实战指南

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

导读

本文围绕 LlamaIndex 官方集成包llama-index-vector-stores-astra-db中的核心类AstraDBVectorStore展开，系统讲解如何在 LlamaIndex 检索流水线中使用 DataStax Astra DB 作为向量数据库后端：从安装、初始化参数、集合自动创建，到节点写入（含去重覆盖）、元数据过滤查询、默认向量检索与 MMR 多样性检索的完整链路。读完本文，你将掌握AstraDBVectorStore的全部核心 API 及其底层实现原理，能够直接将其接入VectorStoreIndex搭建生产可用的 RAG 应用。

一、AstraDBVectorStore 是什么

AstraDBVectorStore是 LlamaIndex 对 DataStax Astra DB（基于 Cassandra 的托管向量数据库）的官方适配，位于 base.py 中，继承自BasePydanticVectorStore。它的本质是：

一个具备向量相似度搜索能力的 Astra DB collection 的抽象。文档与其嵌入向量被存储在一个配了向量索引的 Astra DB collection 中；若 collection 不存在，会在向量存储初始化时自动创建。所有 Astra DB 操作均通过官方 Python 驱动 AstraPy 完成。

类声明处的模块 docstring 还指出：若使用 Astra DB，可先到 DataStax 官网注册账号获取令牌（token）与 API endpoint。

该类的stores_text与flat_metadata属性固定为True，分别表示向量存储会保留原始文本、使用扁平化（非嵌套）元数据存储格式。

二、安装与前置条件

安装命令（见 README.md）：

pip install llama-index-vector-stores-astra-db

根据 pyproject.toml 中的声明，该集成包的运行依赖为：

• astrapy~=1.5：Astra DB 官方 Python 客户端，负责与数据库通信；
• llama-index-core>=0.13.0,<0.15：LlamaIndex 核心库，提供节点、向量存储抽象基类与查询类型定义；
• 要求 Python 版本>=3.10,<4.0。

使用前你需要准备两个 Astra DB 凭证：

三、初始化：参数详解与自动建集

3.1 构造函数参数

AstraDBVectorStore.__init__的签名与参数语义（依据类 docstring 与 base.py 源码）：

关于keyspace与namespace的取舍，源码中通过keyspace_param = keyspace or namespace统一处理，keyspace优先。

3.2 最小初始化示例

README 与类 docstring 均给出了可直接运行的最小示例：

from llama_index.vector_stores.astra_db import AstraDBVectorStore vector_store = AstraDBVectorStore( token="AstraCS:xY3b...", # 你的 Astra DB Token api_endpoint="https://012...abc-us-east1.apps.astra.datastax.com", # 数据库 API 端点 collection_name="astra_v_table", # collection 名称，不存在则自动创建 embedding_dimension=1536, # 必须与所用嵌入模型的维度一致 )

初始化时底层做了三件事（base.py）：

1. 以llama_index作为caller_name、以llama_index.core.__version__作为caller_version构造DataAPIClient，并通过get_database(api_endpoint, token=token, keyspace=keyspace_param)拿到Database对象；
2. 调用database.create_collection(name=..., dimension=embedding_dimension, indexing=collection_indexing, check_exists=False)创建并连接 collection；
3. 其中collection_indexing = {"deny": NON_INDEXED_FIELDS}，NON_INDEXED_FIELDS = ["metadata._node_content", "content"]—— 即不对_node_content与content字段建索引，从而允许存入更长的文本并避免不必要的索引开销。

3.3 旧集合兼容处理

如果目标 collection 已存在且create_collection抛出DataAPIException，初始化逻辑会通过list_collections()检查该 collection 是否已存在（base.py）：

• 若已存在且没有任何 indexing 配置（视为手工创建或旧版本插件创建的"全字段索引"集合），会发出警告，提示其对单条文本长度限制更严格，建议在新 collection 上重新建索引，随后改用get_collection()连接；
• 若 indexing 配置与期望的{"deny": [...]}不完全一致，会警告"可能出现元数据过滤行为异常或文本长度限制"，同样改用get_collection()连接；
• 若配置完全一致则重新抛出异常；若 collection 不存在则直接抛出异常。

此外，该类还提供from_params类方法，参数与构造函数完全一致，可作为参数化工厂使用（base.py），以及class_name()类方法返回"AstraDBVectorStore"。

四、核心 API：写入、读取与删除

4.1add(nodes)：写入节点并幂等去重

add接收List[BaseNode]，对每个节点构造如下文档结构（base.py）：

{ "_id": node.node_id, "content": node.get_content(metadata_mode=MetadataMode.NONE), "metadata": metadata, # node_to_metadata_dict(node, remove_text=True, flat_metadata=True) "$vector": node.get_embedding(), }

其中metadata通过node_to_metadata_dict生成（remove_text=True表示文本单独存于content字段，flat_metadata=True表示扁平化元数据）。

写入策略值得注意——它天然支持幂等覆盖：

1. 先调用collection.insert_many(documents_to_insert, ordered=False)批量插入全部文档；
2. 若抛InsertManyException，则根据err.partial_result.inserted_ids找出未插入成功的_id列表（通常是因为_id已存在导致主键冲突）；
3. 对未插入成功的文档，使用ThreadPoolExecutor(max_workers=REPLACE_DOCUMENTS_MAX_THREADS)（常量为 12）并发执行collection.replace_one({"_id": ...}, document)进行覆盖写；
4. 若replace_one成功数少于待替换数，则抛出ValueError报告失败数量。

最后返回所有写入节点的_id列表。这一设计使"重复 ingest 同一批节点"成为安全的更新操作，测试用例 test_astra_db.py 专门验证了"部分插入失败后走 replace_one 覆盖"的路径。

4.2delete(ref_doc_id)与delete_nodes(...)

• delete(ref_doc_id)：按节点_id删除单个文档，内部调用collection.delete_one({"_id": ref_doc_id})；若传入额外未支持的命名参数会发出警告（base.py）。
• delete_nodes(node_ids=None, filters=None)：支持按节点 ID 列表或元数据过滤器批量删除（base.py）：
  • 单 ID 用delete_one({"_id": ...})，多 ID 用delete_many({"_id": {"$in": [...]}})；
  • 按过滤器删除时使用delete_many(filter_query)。

两个删除接口都要求node_ids与filters二选一，同时传入或都未传都会抛出ValueError。

4.3get_nodes(node_ids=None, filters=None, limit=100)

从 collection 中按 ID 或过滤器取回节点并还原为BaseNode（base.py）：

• 单 ID：{"_id": id}；多 ID：{"_id": {"$in": [...]}}；
• 元数据过滤：{"metadata.<key>": value}形式；
• 查询时使用projection={"*": True}取回全部字段，limit默认 100；
• 若返回文档的metadata中没有_node_content（非本插件写入的数据），会自动用json.dumps(match)兜底填充，再经metadata_dict_to_node还原为节点对象。

五、向量查询：DEFAULT 模式与 MMR 模式

query(query, **kwargs)是检索核心，支持两种模式（base.py）：

_available_query_modes = [ VectorStoreQueryMode.DEFAULT, VectorStoreQueryMode.MMR, ]

使用其他模式（如SPARSE）会抛出NotImplementedError。

5.1 DEFAULT 模式：原生向量相似度检索

直接调用 AstraPy 的collection.find()：

matches = list( self._collection.find( filter=query_metadata, # 元数据过滤（可选） projection={"*": True}, limit=query.similarity_top_k, # 返回 top-k sort={"$vector": query_embedding},# 按向量相似度排序 include_similarity=True, # 让数据库返回 $similarity 分数 ) ) top_k_scores = [match["$similarity"] for match in matches]

即相似度分数由 Astra DB 数据库端计算并返回（$similarity字段），配合similarity_top_k控制返回条数。

5.2 MMR 模式：兼顾相关性与多样性

MMR（最大边际相关）模式先拉取一个较大的候选集，再在 LlamaIndex 侧做多样性重排：

1. 预取数量（prefetch）控制：mmr_prefetch_k与mmr_prefetch_factor二选一，同时传入会抛ValueError。若给mmr_prefetch_k则直接使用该值；否则按similarity_top_k * mmr_prefetch_factor计算，默认因子为常量DEFAULT_MMR_PREFETCH_FACTOR = 4.0；最终prefetch_k = max(prefetch_k0, similarity_top_k)；
2. 预取：调用find(sort={"$vector": ...}, limit=prefetch_k)（此处不需要数据库返回相似度），取出候选集及其$vector嵌入；
3. MMR 重排：调用 LlamaIndex 核心工具get_top_k_mmr_embeddings(query_embedding, ...)（源自 embedding_utils.py），传入similarity_top_k、embedding_ids与mmr_threshold（取自query.mmr_threshold或 kwargs 中的mmr_threshold），得到最终 top-k 的相似度与索引；
4. 按重排索引从预取结果中取出最终匹配项，返回VectorStoreQueryResult(nodes, similarities, ids)。

5.3 结果还原

无论哪种模式，命中文档都会走与get_nodes相同的还原逻辑：metadata缺_node_content时自动兜底，随后metadata_dict_to_node(match["metadata"], text=match["content"])还原节点，最终组装成VectorStoreQueryResult。

六、元数据过滤：仅支持等值匹配

_query_filters_to_dict(query_filters)（base.py）将 LlamaIndex 的MetadataFilters转换为 Astra DB 的过滤表达式：

return {f"metadata.{f.key}": f.value for f in query_filters.filters}

重要限制：只支持ExactMatchFilter或operator == FilterOperator.EQ的MetadataFilter；其他运算符（如大于、小于、模糊匹配等）或嵌套过滤器会抛出NotImplementedError("Only filters with operator=FilterOperator.EQ are supported")。测试 test_astra_db.py 验证了{"metadata.category": "A", "metadata.score": 85}这类等值过滤的转换结果。

使用示例：

from llama_index.core.vector_stores.types import MetadataFilters, MetadataFilter, FilterOperator filters = MetadataFilters( filters=[ MetadataFilter(key="category", value="A", operator=FilterOperator.EQ), ] ) result = vector_store.query(query, filters=filters)

七、接入 LlamaIndex 索引与完整示例

AstraDBVectorStore实现了BasePydanticVectorStore的标准接口（add/delete/get_nodes/query/client），因此可以直接作为VectorStoreIndex的存储后端，与OpenAIEmbedding等嵌入模型组合成完整 RAG 流水线：

from llama_index.core import VectorStoreIndex, StorageContext from llama_index.core.node_parser import SentenceSplitter from llama_index.vector_stores.astra_db import AstraDBVectorStore # 1. 初始化向量存储（自动建集，维度需与嵌入模型一致） vector_store = AstraDBVectorStore( collection_name="astra_v_store", token="AstraCS:...", api_endpoint="https://<db-id>-<region>.apps.astra.datastax.com", embedding_dimension=1536, keyspace="default_keyspace", # 可选 ) # 2. 通过 StorageContext 绑定向量存储并构建索引 storage_context = StorageContext.from_defaults(vector_store=vector_store) index = VectorStoreIndex.from_documents( documents, # List[Document] storage_context=storage_context, transformations=[SentenceSplitter(chunk_size=512, chunk_overlap=50)], ) # 3. 检索与问答 query_engine = index.as_query_engine(similarity_top_k=5) response = query_engine.query("你的问题")

如果需要直接访问底层 collection（例如执行自定义 Astra DB 操作），可通过client属性获取：

collection = vector_store.client # astrapy.Collection 对象

八、测试与质量保障

集成包在 tests/test_astra_db.py 中提供了完整的测试覆盖，可归纳为两类：

• 单元测试（无需真实数据库，mockDataAPIClient）：验证初始化、from_params、add（含InsertManyException触发 replace 覆盖）、delete/delete_nodes、get_nodes、query、过滤器转换、ttl_seconds警告、不支持查询模式的报错、client属性与class_name等行为；
• 集成测试（需设置ASTRA_DB_APPLICATION_TOKEN与ASTRA_DB_API_ENDPOINT环境变量，未设置自动 skip）：覆盖真实建集、CRUD、向量查询，以及 150 条文档的批量覆盖写场景（test_astra_db_insertions验证旧文档被新文档完整覆盖且数量一致）。

这些测试同时是理解该类行为边界的权威参考：例如get_nodes/delete_nodes的"二选一"参数校验、过滤条件仅支持 EQ 等，均有对应的断言约束。

九、注意事项与限制小结

1. 嵌入维度必须匹配：embedding_dimension需与所用嵌入模型的输出维度一致（OpenAItext-embedding-3-small为 1536，其他模型以实际为准），创建集合后无法在线修改；
2. ttl_seconds不支持：传入仅产生警告并被忽略，勿依赖 TTL 过期能力；
3. 过滤能力受限：元数据过滤仅支持等值（EQ）匹配，复杂范围/模糊过滤需先取回再在应用侧处理；
4. 旧集合索引策略：若 collection 是全字段索引的旧版本创建，长文本写入可能受限，建议按初始化时的警告提示新建 collection；
5. 查询模式：仅支持DEFAULT与MMR两种模式，其他模式会抛NotImplementedError；
6. 写入是幂等的：重复添加相同node_id的节点会触发覆盖写，可用于增量更新场景。

综上，AstraDBVectorStore通过 AstraPy 驱动将 LlamaIndex 的标准化向量存储接口与 Astra DB 的托管 Cassandra 向量能力无缝对接，自动建集、幂等写入与 MMR 多样性检索等细节均已内置封装，适合作为云端 RAG 应用的向量后端直接使用。

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index