Haystack 与 DataStax Astra DB 集成实战：AstraDocumentStore 与 AstraEmbeddingRetriever 构建向量检索流水线

Haystack 与 DataStax Astra DB 集成实战AstraDocumentStore 与 AstraEmbeddingRetriever 构建向量检索流水线【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack本文以 Haystack 的 Astra 集成astra-haystack为核心系统讲解如何用AstraDocumentStore将海量文档写入基于 Apache Cassandra 的 serverless 向量数据库 DataStax Astra DB并通过AstraEmbeddingRetriever在 RAG、语义搜索与抽取式问答流水线中完成向量检索。读完本文你将掌握 Astra 文档库的初始化参数、去重策略、元数据过滤与文档管理 API 的完整用法并能在 HaystackPipeline中端到端搭建一套文本嵌入 → 向量检索 → 生成的生产级检索链路。一、为什么选择 Astra DB 作为 Haystack 的向量存储DataStax Astra DB 是一款构建于 Apache Cassandra 之上的 serverless 向量数据库原生支持向量搜索与自动扩缩容可部署在 AWS、GCP 或 Azure并能轻松扩展至多个云区域以获得多区域可用性、低延迟数据访问和数据主权同时避免云厂商锁定。在 Haystack 中Astra 集成以haystack_integrations包的形式提供核心包含两个组件AstraDocumentStore文档存储负责与 Astra DB 建立连接、写入/查询/删除文档并基于向量相似度执行检索AstraEmbeddingRetriever向量检索器接收查询向量从AstraDocumentStore中召回与查询最相关的文档。两者对应的 API 参考文档见 版本化 API 参考完整使用指南见 AstraDocumentStore 使用文档 与 AstraEmbeddingRetriever 使用文档。二、安装集成与获取连接凭证2.1 安装在已创建 Astra DB 账号与数据库的前提下安装astra-haystack集成pip install astra-haystack如果希望直接运行文中的嵌入示例不依赖云端嵌入 API可以一并安装 sentence-transformerspip install sentence-transformers2.2 获取凭证在 AstraDB 的 Web UI 中你需要准备两类信息数据库 ID / API Endpoint在 Astra 控制台的 Connect 标签页中选择JSON API并点击Generate Configuration即可生成 API EndpointApplication Token同样在 Connect 页面生成作为访问数据库的认证令牌。此外你还需要一个collection 名称与一个namespace。创建 collection 时必须同步指定嵌入向量维度embedding dimensions和相似度度量similarity metric。其中 namespace 用于在数据库中组织数据在 Apache Cassandra 术语中称为keyspace。2.3 通过环境变量管理密钥Haystack 强烈建议通过环境变量传递认证数据而不是把密钥硬编码进代码。运行示例前请先填充export ASTRA_DB_API_ENDPOINThttps://database-id-region.apps.astra.datastax.com export ASTRA_DB_APPLICATION_TOKENAstraCS:...这两个环境变量名与AstraDocumentStore构造函数中Secret.from_env_var(...)的默认读取来源完全对应详见下文。三、AstraDocumentStore初始化与核心参数AstraDocumentStore是 Astra 集成在 Haystack 侧的数据面连接通过Astra DB JSON API建立与管理。最简初始化方式是利用环境变量from haystack import Document from haystack_integrations.document_stores.astra import AstraDocumentStore document_store AstraDocumentStore() document_store.write_documents( [Document(contentThis is first), Document(contentThis is second)], ) print(document_store.count_documents())3.1 完整构造函数签名__init__( api_endpoint: Secret Secret.from_env_var(ASTRA_DB_API_ENDPOINT), token: Secret Secret.from_env_var(ASTRA_DB_APPLICATION_TOKEN), collection_name: str documents, embedding_dimension: int 768, duplicates_policy: DuplicatePolicy DuplicatePolicy.NONE, similarity: str cosine, namespace: str | None None, ) - None3.2 参数说明参数类型默认值说明api_endpointSecret环境变量ASTRA_DB_API_ENDPOINTAstra DB 的 JSON API 端点可在控制台 Connect 页面生成tokenSecret环境变量ASTRA_DB_APPLICATION_TOKENAstra DB 应用令牌Application Tokencollection_namestrdocuments当前 Astra DB 中 keyspace 内使用的 collection 名称embedding_dimensionint768嵌入向量的维度必须与写入向量的维度一致duplicates_policyDuplicatePolicyDuplicatePolicy.NONE处理重复文档的策略取值见下节similaritystrcosine用于比较文档向量的相似度函数namespacestr \| NoneNone数据所在命名空间Cassandra keyspace不传时使用默认 keyspace注意如果 API endpoint 或 token 未设置构造函数会抛出ValueError。从源码结构看Secret机制是 Haystack 统一的敏感信息管理方式Secret.from_env_var(...)让密钥只在真正发起请求时才从环境变量读取从而避免密钥出现在序列化后的 YAML/JSON 配置中。DuplicatePolicy与FilterPolicy等类型定义在核心库的 文档存储类型目录 下其中 policy.py 定义了去重枚举filter_policy.py 定义了过滤合并策略。四、写入文档与重复文档处理策略write_documents用于将文档索引入库供后续查询使用write_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - int入参documents为 HaystackDocument对象列表policy指定重复文档处理策略。返回实际写入的文档数量int。异常ValueError—— 传入的文档既不是Document也不是dictDuplicateDocumentError—— 已存在相同 ID 的文档且策略为FAILException—— 文档 ID 不是字符串或文档中同时包含id与_id字段。4.1 DuplicatePolicy 四种策略DuplicatePolicy枚举定义在 haystack/document_stores/types/policy.py核心库Astra 文档库直接复用该枚举策略行为DuplicatePolicy.NONE默认策略。如果相同 ID 的文档已存在则跳过、不写入DuplicatePolicy.SKIP如果相同 ID 的文档已存在则跳过、不写入DuplicatePolicy.OVERWRITE如果相同 ID 的文档已存在则覆盖写入DuplicatePolicy.FAIL如果相同 ID 的文档已存在则抛出错误注意NONE与SKIP的文档行为一致二者区别在于语义定位NONE是构造函数与write_documents的默认值SKIP更适合在明确允许跳过重复的业务场景中显式声明。该枚举同样被 Haystack 核心文档存储如 InMemoryDocumentStore与文档写入组件 DocumentWriter 复用是整个框架统一的去重语义。五、文档管理 API查询、过滤、更新与统计AstraDocumentStore实现了 Haystack 文档存储协议协议定义见 haystack/document_stores/types/protocol.py并提供如下完整的数据管理方法5.1 查询类filter_documents(filters: dict[str, Any] | None None) - list[Document]返回最多 1000 条匹配过滤条件的文档过滤条件非法或不受支持时抛出AstraDocumentStoreFilterError。get_documents_by_id(ids: list[str]) - list[Document]按 ID 列表批量获取文档。get_document_by_id(document_id: str) - Document按单个 ID 获取文档未找到时抛出MissingDocumentError。search(query_embedding: list[float], top_k: int, filters: dict[str, Any] | None None) - list[Document]基于查询向量执行相似度检索返回top_k条匹配文档。这是向量检索的底层实现AstraEmbeddingRetriever在run时最终会调用到它。5.2 计数类count_documents() - int统计文档库中的文档总数。count_documents_by_filter(filters: dict[str, Any]) - int应用过滤条件后统计匹配文档数。count_unique_metadata_by_filter(filters, metadata_fields: list[str]) - dict[str, int]对匹配文档的每个元数据字段统计唯一值个数返回{字段名: 唯一值个数}。5.3 删除类delete_documents(document_ids: list[str]) - None按 ID 删除文档若提供了 ID 但没有任何文档被删除抛出MissingDocumentError。delete_all_documents() - None清空文档库。delete_by_filter(filters: dict[str, Any]) - int删除匹配过滤条件的文档返回删除数量过滤非法时抛出AstraDocumentStoreFilterError。5.4 更新类update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int将匹配过滤条件文档的元数据更新为meta指定字段与已有元数据合并返回更新数量。5.5 元数据探查类get_metadata_fields_info() - dict[str, dict[str, str]]返回元数据字段及其类型映射形如{field_name: {type: ...}}。get_metadata_field_min_max(metadata_field: str) - dict[str, Any]返回指定字段的min与max值。get_metadata_field_unique_values(metadata_field, search_termNone, from_0, size10, filtersNone) - tuple[list[Any], int]检索某字段的唯一值支持大小写不敏感的模糊搜索search_term、分页from_/size与过滤filters返回(分页后的值列表, 总数)。关于get_metadata_field_unique_values的类型注意点不同类型但值相等的元数据会被视为不同的唯一值例如整数1、布尔True与字符串1会被分别返回但有一个例外——Astra DB 的 Data API 在存储时会无条件将整数值的浮点数如1.0规范化canonicalize为整数因此1.0float写入后读回一定是整数1int而带小数部分的浮点数如1.5不受影响可以正常往返。六、AstraEmbeddingRetriever向量检索组件AstraEmbeddingRetriever是嵌入检索组件它比较查询向量与文档向量的相似度并根据结果从AstraDocumentStore中召回最相关的文档。其 API 参考见 版本化 API 参考 的haystack_integrations.components.retrievers.astra.retriever部分。6.1 初始化__init__( document_store: AstraDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE, ) - None参数说明document_store一个AstraDocumentStore实例必填filters用于缩小搜索空间的过滤条件字典top_k最多检索的文档数量默认 10filter_policy过滤条件应用策略REPLACE或MERGE默认REPLACE6.2 run 与 run_asyncrun( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询文本的向量表示float 列表。filters运行时应用的过滤条件。运行时过滤条件的具体应用方式取决于初始化时选择的filter_policy。top_k运行时覆盖的最大检索数量可选。返回{documents: [...]}即从AstraDocumentStore检索到的文档列表。run_async的签名与返回结构与run完全一致用于异步场景从文档说明看它是将同步搜索放到线程池中执行从而避免阻塞事件循环适合在异步 Pipeline 中与run_async链路配合使用。6.3 序列化to_dict() - dict[str, Any]将组件序列化为字典便于通过 YAML/JSON 配置持久化。from_dict(data: dict[str, Any]) - AstraEmbeddingRetriever从字典反序列化重建组件。这两个方法让AstraEmbeddingRetriever可以无缝嵌入 Haystack 的声明式 Pipeline 定义序列化为 YAML/JSON中。七、filter_policy初始化过滤与运行时过滤如何合并filter_policy控制检索器初始化时设置的过滤条件init filters与run调用时传入的过滤条件runtime filters之间的关系。核心枚举与合并逻辑定义在 haystack/document_stores/types/filter_policy.pyFilterPolicy.REPLACE默认运行时传入的过滤条件替换初始化时设置的过滤条件。FilterPolicy.MERGE运行时过滤条件与初始化过滤条件合并若字段重叠运行时值覆盖初始化值。从 filter_policy.py 中apply_filter_policy的实现可以看出MERGE策略会依据初始化/运行时过滤条件各自是比较型过滤包含field、operator、value键还是逻辑型过滤包含operator与conditions键进行四种组合初始化过滤运行时过滤合并结果比较型比较型以AND合并为逻辑过滤同字段时运行时覆盖初始化比较型逻辑型当逻辑运算符一致时把比较条件并入conditions逻辑型比较型当逻辑运算符一致时把比较条件并入conditions同字段时运行时覆盖逻辑型逻辑型运算符相同则合并conditions否则忽略初始化过滤并告警例如初始化时设置{field: meta.type, operator: , value: article}运行时传入{operator: AND, conditions: [{field: meta.rating, operator: , value: 3}]}在MERGEAND策略下会合并为一个同时包含两个条件的逻辑过滤。该机制让公共过滤条件固化在组件里、个性化过滤条件每次查询动态传入成为可能是构建多租户或分类检索场景的关键开关。八、端到端示例在 Pipeline 中使用 AstraEmbeddingRetriever下面是一个完整的 RAG 语义检索流水线示例节选自 astraretriever.mdx 使用文档演示了文档嵌入写入 查询嵌入检索的完整闭环。from haystack import Document, Pipeline from haystack.components.embedders import ( SentenceTransformersTextEmbedder, SentenceTransformersDocumentEmbedder, ) from haystack_integrations.components.retrievers.astra import AstraEmbeddingRetriever from haystack_integrations.document_stores.astra import AstraDocumentStore document_store AstraDocumentStore() model sentence-transformers/all-mpnet-base-v2 documents [ Document(contentThere are over 7,000 languages spoken around the world today.), Document( contentElephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors., ), Document( contentIn certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves., ), ] document_embedder SentenceTransformersDocumentEmbedder(modelmodel) document_embedder.warm_up() documents_with_embeddings document_embedder.run(documents) document_store.write_documents( documents_with_embeddings.get(documents), policyDuplicatePolicy.SKIP, ) query_pipeline Pipeline() query_pipeline.add_component( text_embedder, SentenceTransformersTextEmbedder(modelmodel), ) query_pipeline.add_component( retriever, AstraEmbeddingRetriever(document_storedocument_store), ) query_pipeline.connect(text_embedder.embedding, retriever.query_embedding) query How many languages are there? result query_pipeline.run({text_embedder: {text: query}}) print(result[retriever][documents][0])示例输出分数与向量维度取决于所用嵌入模型Document(idcfe93bc1c274908801e6670440bf2bbba54fad792770d57421f85ffa2a4fcc94, content: There are over 7,000 languages spoken around the world today., score: 0.8929937, embedding: vector of size 768)8.1 典型流水线位置根据 astraretriever.mdx 的说明AstraEmbeddingRetriever最常见的三种流水线位置是RAG 流水线位于 Text Embedder 之后、PromptBuilder之前语义搜索流水线作为查询流水线的最后一个组件直接输出结果抽取式问答流水线位于 Text Embedder 之后、ExtractiveReader之前。使用时请确保索引流水线中已有Document Embedder、查询流水线中已有Text Embedder为检索器提供查询向量与文档向量。九、索引警告Indexing Warnings的原因与处理创建AstraDocumentStore时你可能会看到如下两类警告之一Astra DB collection...is detected as having indexing turned on for all fields (either created manually or by older versions of this plugin). This implies stricter limitations on the amount of text each string in a document can store. Consider indexing anew on a fresh collection to be able to store longer texts.或者Astra DB collection...is detected as having the following indexing policy:{...}. This does not match the requested indexing policy for this object:{...}. In particular, there may be stricter limitations on the amount of text each string in a document can store. Consider indexing anew on a fresh collection to be able to store longer texts.9.1 为什么会出现该警告collection 已经存在且被配置为对所有字段开启索引可能由你之前手动创建或由旧版本插件创建。而 Haystack 创建 collection 时会应用一套针对其用途优化的索引策略该策略允许存储更长的文本并避免索引那些你不需要过滤的字段从而降低写入开销。9.2 常见原因你在 Haystack 之外创建了 collection例如在 Astra UI 中手动创建或通过 AstraPy 的Database.create_collection()创建你使用旧版本的插件创建了 collection。9.3 影响与解决方案影响这只是一个警告。除非你尝试存储非常长的文本字段此时 Astra DB 会返回索引错误否则应用可以正常运行。解决方案推荐做法如果能够重新填充数据删除并重建 collection然后重新运行你的 Haystack 应用让插件以优化后的索引策略重新创建 collection忽略警告如果你确定不会存储很长的文本字段可以直接忽略该警告继续使用。十、异常体系Astra 文档存储的错误类型Astra 集成定义了三级错误体系见 版本化 API 参考 的haystack_integrations.document_stores.astra.errors部分与 Haystack 核心错误类型衔接异常类基类触发场景AstraDocumentStoreErrorDocumentStoreError所有 AstraDocumentStore 错误的父类AstraDocumentStoreFilterErrorFilterError向 AstraDocumentStore 传入了非法过滤条件AstraDocumentStoreConfigErrorAstraDocumentStoreError向 AstraDocumentStore 传入了非法配置在编写健壮的检索代码时建议对AstraDocumentStoreFilterError过滤条件书写错误与AstraDocumentStoreConfigError初始化配置错误分别捕获以便快速定位是查询侧问题还是存储侧问题。十一、结语Astra 集成让 Haystack 可以直接利用 DataStax Astra DB 的 serverless 向量数据库能力AstraDocumentStore承担文档写入、过滤、统计、更新与向量检索等全部数据面操作AstraEmbeddingRetriever则作为标准 Haystack 组件无缝嵌入 RAG、语义搜索与抽取式问答流水线。结合本文介绍的DuplicatePolicy去重策略、filter_policy过滤合并机制与索引警告处理方案你可以在生产环境中稳定运行基于 Astra DB 的向量检索应用。更多配套资料可参阅 AstraDocumentStore 使用文档、AstraEmbeddingRetriever 使用文档 以及 Haystack 文档存储类型定义。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考