Haystack 集成 MongoDB Atlas向量检索、全文检索与文档存储实战指南【免费下载链接】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/haystackMongoDB Atlas 是 MongoDB 官方提供的多云托管数据库服务原生支持向量搜索Atlas Vector Search与全文搜索Atlas Search。在 Haystack 中mongodb-atlas-haystack集成包提供了MongoDBAtlasDocumentStore文档存储、MongoDBAtlasEmbeddingRetriever向量检索器与MongoDBAtlasFullTextRetriever全文检索器三个核心组件可用于构建基于 Atlas 的 RAG、语义搜索与混合检索系统。读完本文你将掌握 Atlas 连接与索引配置、文档读写删改、两类检索器的全部参数与调用方式以及如何在 Haystack Pipeline 中组合出可运行的检索链路。本文内容以仓库中 Haystack 2.22 版本文档为准对应参考文档见 mongodb_atlas.md配套的组件使用指南位于 mongodbatlasdocumentstore.mdx、mongodbatlasembeddingretriever.mdx 与 mongodbatlasfulltextretriever.mdx。集成概览三个组件各司其职mongodb-atlas-haystack集成围绕三个核心类展开全部位于haystack_integrations命名空间下组件模块职责MongoDBAtlasDocumentStorehaystack_integrations.document_stores.mongodb_atlas.document_store管理 MongoDB Atlas 中既有 collection 的文档读写、过滤、删除、更新与统计MongoDBAtlasEmbeddingRetrieverhaystack_integrations.components.retrievers.mongodb_atlas.embedding_retriever基于 embedding 相似度从文档存储中检索文档MongoDBAtlasFullTextRetrieverhaystack_integrations.components.retrievers.mongodb_atlas.full_text_retriever基于 Atlas Search 全文索引进行文本检索其中 Document Store 负责数据面两个 Retriever 负责查询面。向量检索的相似度度量取决于创建vector_search_index时选择的指标cosine、dot product 或 euclidean全文检索则依赖full_text_search_index的配置——两个索引都需要在 MongoDB Atlas Web UI 中手动创建这也是开始使用前最重要的前置步骤。安装与连接准备安装集成包pip install mongodb-atlas-haystack获取连接字符串MongoDBAtlasDocumentStore需要一条标准 MongoDB 连接字符串才能连接 Atlas格式如下mongodbsrv://{mongo_atlas_username}:{mongo_atlas_password}{mongo_atlas_host}/?{mongo_atlas_params_string}获取方式登录 MongoDB Atlas Dashboard点击集群的CONNECT按钮选择 Python 作为驱动后复制连接字符串。连接字符串既可以作为mongo_connection_string参数直接传入构造函数也可以通过环境变量MONGO_CONNECTION_STRING提供构造函数默认从该环境变量读取。export MONGO_CONNECTION_STRINGmongodbsrv://username:passwordcluster_name.mongodb.net/?retryWritestruewmajority创建索引必做在初始化 Document Store 之前需要为将要使用的 collection 创建两类索引vector_search_index向量搜索索引用于 embedding 相似度检索创建时需选择度量方式cosine、dot product 或 euclidean并确保 embedding 字段默认embedding被纳入索引配置。full_text_search_index全文搜索索引用于 Atlas Search 文本检索。两个索引均可通过 Atlas Web UI 创建。需要注意的是如果后续要在检索时使用 filters 过滤字段这些字段也必须包含在对应索引的配置中——这一配置同样要在 Atlas Web UI 中手动完成。MongoDBAtlasDocumentStore文档存储详解MongoDBAtlasDocumentStore的核心职责是对一个已存在的 collection 进行读写。它不会替用户创建数据库或集合——创建数据库和集合通常应在 Atlas Web UI 或通过 MongoDB Python 驱动完成超出该组件的职责范围。初始化参数from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore store MongoDBAtlasDocumentStore( database_nameyour_existing_db, collection_nameyour_existing_collection, vector_search_indexyour_existing_index, full_text_search_indexyour_existing_index, ) print(store.count_documents())完整构造签名及参数说明如下__init__( *, mongo_connection_string: Secret Secret.from_env_var(MONGO_CONNECTION_STRING), database_name: str, collection_name: str, vector_search_index: str, full_text_search_index: str, embedding_field: str embedding, content_field: str content ) - None参数类型默认值说明mongo_connection_stringSecret环境变量MONGO_CONNECTION_STRINGAtlas 连接字符串格式见上文未显式传入时自动读取环境变量database_namestr必填要使用的数据库名称collection_namestr必填要使用的集合名称若用于向量检索该集合需在embedding字段上配置向量搜索索引vector_search_indexstr必填用于向量检索的索引名称需在 Atlas Web UI 创建后填入full_text_search_indexstr必填用于全文检索的索引名称需在 Atlas Web UI 创建后填入embedding_fieldstrembedding存放文档 embedding 的字段名content_fieldstrcontent加载到 HaystackDocument对象中作为 content 的字段名在对接既有 collection 做检索时尤其有用。官方不建议在使用 Haystack 创建的集合时修改此参数初始化时会校验collection_name是否包含非法字符若包含则抛出ValueError。连接与集合访问Document Store 暴露了两个只读属性便于底层操作connection当前活跃的 MongoDB 客户端连接类型为AsyncMongoClient | MongoClientcollection当前活跃的 MongoDB 集合类型为AsyncCollection | Collection。例如在混合检索示例中可通过document_store.collection.delete_many({})清空集合旧数据保证示例可重复执行该用法出现在 mongodbatlasfulltextretriever.mdx 的 Pipeline 示例中。文档写入write_documents( documents: list[Document], policy: DuplicatePolicy DuplicatePolicy.NONE ) - intdocuments要写入的 HaystackDocument列表policy重复文档处理策略类型为DuplicatePolicy。该枚举定义在 Haystack 核心的 policy.py 中包含NONE、SKIP跳过已存在文档、OVERWRITE覆盖已存在文档、FAIL遇到重复即报错四种取值默认NONE返回实际写入的文档数量若文档不是Document类型则抛出ValueError若与DuplicatePolicy.FAIL冲突则抛出DuplicateDocumentError。过滤、统计与元数据查询Document Store 提供了一套与 Haystack 元数据过滤语法对齐的查询方法全部方法均有_async异步版本方法功能count_documents()/count_documents_async()返回集合中的文档总数count_documents_by_filter(filters)/_async应用过滤器后统计匹配文档数count_unique_metadata_by_filter(filters, metadata_fields)/_async对匹配文档的每个元数据字段统计唯一值个数返回{字段名: 唯一值个数}get_metadata_fields_info()/_async返回元数据字段及其类型由于 MongoDB 无固定 schema该方法采样最近 50 篇文档来推断字段与类型get_metadata_field_min_max(metadata_field)/_async返回指定元数据字段的最小值与最大值结果为包含min、max键的字典get_metadata_field_unique_values(metadata_field, search_term, from_, size, filters)/_async分页获取某字段的唯一值search_term按大小写不敏感的子串匹配from_为分页起始下标size为返回条数filters用于限定候选文档返回(唯一值列表, 匹配总数)filter_documents(filters)/_async返回满足过滤条件的Document列表过滤语法遵循 Haystack 元数据过滤规范get_metadata_field_unique_values有一个值得注意的类型语义不同类型的值即使 Python 中相等也会被区分保留例如整数1、布尔值True、字符串1会被视为三个独立值唯一的例外是 MongoDB 聚合$group会跨 BSON 数值子类型比较因此整数值的浮点数如1.0会与数值相等的整数1合并二者只保留一个而带小数部分的浮点数如1.5不受影响仍与整数保持独立。删除与更新delete_documents(document_ids: list[str]) - None # 按 id 删除 delete_by_filter(filters: dict[str, Any]) - int # 按过滤条件删除返回删除数 update_by_filter(filters: dict[str, Any], meta: dict[str, Any]) - int # 更新匹配文档的元数据返回更新数 delete_all_documents(*, recreate_collection: bool False) - None # 清空全部文档delete_all_documents的recreate_collection参数值得关注为True时直接 drop 并重建集合保留原配置与索引对非常大的集合更快为False时仅删除文档而保留集合结构。序列化与资源释放to_dict()将组件序列化为字典便于 Pipeline 的 YAML/JSON 持久化from_dict(data)从字典反序列化还原组件实例close()/close_async()释放 Document Store 底层的同步/异步资源在不再使用时调用以清理连接。MongoDBAtlasEmbeddingRetriever向量相似度检索MongoDBAtlasEmbeddingRetriever将查询 embedding 与集合中存储的文档 embedding 进行相似度比较返回最相关的文档。相似度结果取决于MongoDBAtlasDocumentStore使用的vector_search_index及其创建时选择的度量指标cosine、dot product 或 euclidean。初始化与运行__init__( *, document_store: MongoDBAtlasDocumentStore, filters: dict[str, Any] | None None, top_k: int 10, filter_policy: str | FilterPolicy FilterPolicy.REPLACE ) - None参数说明document_storeMongoDBAtlasDocumentStore实例若传入非该类型实例则抛出ValueErrorfilters应用于检索结果的过滤条件。注意过滤使用的字段必须包含在vector_search_index的配置中该配置需在 Atlas Web UI 手动完成top_k最大返回文档数默认10filter_policy过滤策略决定运行时 filters 如何与初始化 filters 组合默认FilterPolicy.REPLACErun方法签名run( query_embedding: list[float], filters: dict[str, Any] | None None, top_k: int | None None, ) - dict[str, list[Document]]query_embedding查询的 embedding 向量必填其维度必须与 Document Store 中存储的 embedding 维度一致否则无法正确匹配filters运行时过滤条件具体应用方式取决于初始化时选择的filter_policytop_k运行时覆盖初始化时的top_k值返回字典documents键对应与查询 embedding 最相似的Document列表。独立使用示例import numpy as np from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore from haystack_integrations.components.retrievers.mongodb_atlas import MongoDBAtlasEmbeddingRetriever store MongoDBAtlasDocumentStore(database_namehaystack_integration_test, collection_nametest_embeddings_collection, vector_search_indexcosine_index, full_text_search_indexfull_text_index) retriever MongoDBAtlasEmbeddingRetriever(document_storestore) results retriever.run(query_embeddingnp.random.random(768).tolist()) print(results[documents])以上示例从 Document Store 中检索与随机查询 embedding 最相似的 10 篇文档若随机向量的维度此处为 768与集合中存储的 embedding 维度不一致检索将无法得到有意义的结果。在 RAG Pipeline 中使用MongoDBAtlasEmbeddingRetriever最常见的 Pipeline 位置是Text Embedder 之后、PromptBuilder之前RAG 场景或作为语义搜索管线的末端组件。完整示例见 mongodbatlasembeddingretriever.mdx核心链路如下from haystack import Pipeline, Document from haystack.document_stores.types import DuplicatePolicy from haystack.components.writers import DocumentWriter from haystack.components.generators import OpenAIGenerator from haystack.components.builders.prompt_builder import PromptBuilder from haystack.components.embedders import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore from haystack_integrations.components.retrievers.mongodb_atlas import MongoDBAtlasEmbeddingRetriever documents [ Document(contentMy name is Jean and I live in Paris.), Document(contentMy name is Mark and I live in Berlin.), Document(contentMy name is Giorgio and I live in Rome.), ] document_store MongoDBAtlasDocumentStore() # 索引管线文档嵌入 写入 doc_writer DocumentWriter(document_storedocument_store, policyDuplicatePolicy.SKIP) doc_embedder SentenceTransformersDocumentEmbedder(modelintfloat/e5-base-v2) query_embedder SentenceTransformersTextEmbedder(modelintfloat/e5-base-v2) ingestion_pipe Pipeline() ingestion_pipe.add_component(instancedoc_embedder, namedoc_embedder) ingestion_pipe.add_component(instancedoc_writer, namedoc_writer) ingestion_pipe.connect(doc_embedder.documents, doc_writer.documents) ingestion_pipe.run({doc_embedder: {documents: documents}}) # RAG 管线查询嵌入 - 检索 - 组装提示 - 生成 prompt_template Given these documents, answer the question.\nDocuments: {% for doc in documents %} {{ doc.content }} {% endfor %} \nQuestion: {{question}} \nAnswer: rag_pipeline Pipeline() rag_pipeline.add_component(instancequery_embedder, namequery_embedder) rag_pipeline.add_component( instanceMongoDBAtlasEmbeddingRetriever(document_storedocument_store), nameretriever, ) rag_pipeline.add_component(instancePromptBuilder(templateprompt_template), nameprompt_builder) rag_pipeline.add_component(instanceOpenAIGenerator(), namellm) rag_pipeline.connect(query_embedder, retriever.query_embedding) rag_pipeline.connect(retriever, prompt_builder.documents) rag_pipeline.connect(prompt_builder, llm) question Where does Mark live? result rag_pipeline.run({ query_embedder: {text: question}, prompt_builder: {question: question}, }) print(result[answer_builder][answers])注意上述示例中DocumentWriter与DocumentJoiner等组件来自 Haystack 核心包见 writers 与 joiners 目录向量检索器本身来自集成包。MongoDBAtlasFullTextRetriever全文检索MongoDBAtlasFullTextRetriever基于MongoDBAtlasDocumentStore中配置的full_text_search_index执行 Atlas Search 全文检索适合关键词匹配、拼写容忍等场景且不需要 embedding。初始化与运行初始化签名与 Embedding Retriever 完全一致document_store、filters、top_k、filter_policy且同样要求document_store必须是MongoDBAtlasDocumentStore实例过滤所用字段需包含在full_text_search_index配置中。run方法签名明显更丰富run( query: str | list[str], fuzzy: dict[str, int] | None None, match_criteria: Literal[any, all] | None None, score: dict[str, dict] | None None, synonyms: str | None None, filters: dict[str, Any] | None None, top_k: int 10, ) - dict[str, list[Document]]参数类型说明querystr \| list[str]查询字符串或查询字符串列表若包含多个词项Atlas Search 会分别对每个词项求匹配fuzzydict[str, int] \| None启用模糊匹配用于查找与搜索词相似的字符串。可配置项包括maxEdits、prefixLength、maxExpansions。注意fuzzy不能与synonyms同时使用match_criteriaLiteral[any, all] \| None定义查询词项的匹配方式支持any与allscoredict[str, dict] \| None指定匹配结果的评分方式支持boost、constant、function等选项synonymsstr \| None索引中同义词映射定义的名称不能为空字符串不能与fuzzy同时使用filtersdict[str, Any] \| None运行时过滤条件应用方式取决于初始化的filter_policytop_kint最大返回文档数默认10会覆盖初始化时的值返回字典的documents键为与查询最匹配的Document列表。run_async提供同样的异步能力参数与返回值完全一致。独立使用示例from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore from haystack_integrations.components.retrievers.mongodb_atlas import MongoDBAtlasFullTextRetriever store MongoDBAtlasDocumentStore(database_nameyour_existing_db, collection_nameyour_existing_collection, vector_search_indexyour_existing_index, full_text_search_indexyour_existing_index) retriever MongoDBAtlasFullTextRetriever(document_storestore) results retriever.run(queryLorem ipsum) print(results[documents])混合检索向量 全文 结果融合全文检索与向量检索可以互补向量检索捕捉语义相似全文检索保证关键词精确命中。官方示例见 mongodbatlasfulltextretriever.mdx用DocumentJoiner以reciprocal_rank_fusion倒数排名融合模式合并两条检索结果from haystack import Pipeline, Document from haystack.document_stores.types import DuplicatePolicy from haystack.components.writers import DocumentWriter from haystack.components.embedders import ( SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, ) from haystack.components.joiners import DocumentJoiner from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore from haystack_integrations.components.retrievers.mongodb_atlas import ( MongoDBAtlasEmbeddingRetriever, MongoDBAtlasFullTextRetriever, ) documents [ Document(contentMy name is Jean and I live in Paris.), Document(contentMy name is Mark and I live in Berlin.), Document(contentMy name is Giorgio and I live in Rome.), Document(contentPython is a programming language popular for data science.), Document(contentMongoDB Atlas offers full-text search and vector search capabilities.), ] document_store MongoDBAtlasDocumentStore( database_namehaystack_test, collection_nametest_collection, vector_search_indextest_vector_search_index, full_text_search_indextest_full_text_search_index, ) # 清理旧数据保证示例可重复执行 document_store.collection.delete_many({}) # 索引管线 ingest_pipe Pipeline() doc_embedder SentenceTransformersDocumentEmbedder(modelintfloat/e5-base-v2) ingest_pipe.add_component(instancedoc_embedder, namedoc_embedder) doc_writer DocumentWriter(document_storedocument_store, policyDuplicatePolicy.SKIP) ingest_pipe.add_component(instancedoc_writer, namedoc_writer) ingest_pipe.connect(doc_embedder.documents, doc_writer.documents) ingest_pipe.run({doc_embedder: {documents: documents}}) # 查询管线语义检索 全文检索 倒数排名融合 query_pipe Pipeline() text_embedder SentenceTransformersTextEmbedder(modelintfloat/e5-base-v2) query_pipe.add_component(instancetext_embedder, nametext_embedder) embed_retriever MongoDBAtlasEmbeddingRetriever(document_storedocument_store, top_k3) query_pipe.add_component(instanceembed_retriever, nameembedding_retriever) query_pipe.connect(text_embedder, embedding_retriever) ft_retriever MongoDBAtlasFullTextRetriever(document_storedocument_store, top_k3) query_pipe.add_component(instanceft_retriever, namefull_text_retriever) joiner DocumentJoiner(join_modereciprocal_rank_fusion, top_k3) query_pipe.add_component(instancejoiner, namejoiner) query_pipe.connect(embedding_retriever, joiner) query_pipe.connect(full_text_retriever, joiner) question Where does Mark live? output query_pipe.run({ text_embedder: {text: question}, full_text_retriever: {query: question}, }) for doc in output[joiner][documents]: print(f- {doc.content})过滤机制与 FilterPolicy两个 Retriever 都在初始化与运行时接受filters参数两者的组合方式由filter_policy决定。该枚举定义在 Haystack 核心的 filter_policy.py 中支持两个取值FilterPolicy.REPLACE默认运行时传入的 filters 直接替换初始化时的 filtersFilterPolicy.MERGE运行时 filters 与初始化 filters 合并重叠字段以运行时值为准。当使用MERGE时Haystack 会依据过滤条件的结构比较型过滤器{field, operator, value}与逻辑型过滤器{operator, conditions}进行智能组合具体合并规则实现在 apply_filter_policy 函数 中两个逻辑过滤器运算符一致时合并conditions不一致时以运行时为准比较过滤器字段冲突时运行时值覆盖初始化值。过滤字段的完整语法遵循 Haystack 元数据过滤规范。需要特别强调的是在 Atlas 场景下过滤所用字段必须预先纳入vector_search_index或full_text_search_index的索引配置否则过滤不会生效。这一步只能通过 Atlas Web UI 手动完成属于集成使用的常见坑点。常见问题与使用注意事项Embedding 维度必须一致MongoDBAtlasEmbeddingRetriever的query_embedding维度必须与集合中存储的 embedding 维度一致否则向量检索无法正确执行。实际项目中应让 Text Embedder 与 Document Embedder 使用同一模型如intfloat/e5-base-v2保证两端向量空间一致。索引需提前在 Atlas Web UI 创建vector_search_index与full_text_search_index均不由 Haystack 自动创建若使用 filters相关字段还需加入索引配置。fuzzy与synonyms互斥全文检索中二者不可同时启用同时传入会出错。Document Store 不负责建库建集MongoDBAtlasDocumentStore只读写既有 collection数据库与集合的创建应通过 Atlas Web UI 或 MongoDB 驱动完成。content_field谨慎修改对接既有 collection 做检索时才建议调整该字段对 Haystack 自己写入的集合保持默认content即可。资源释放使用完毕后调用close()或异步场景下的close_async()释放底层连接资源。序列化支持三个组件均实现了to_dict()/from_dict()可以无缝集成进 Haystack Pipeline 的 YAML 序列化体系中。总结mongodb-atlas-haystack集成让 Haystack 应用可以直接把 MongoDB Atlas 作为生产级文档存储与双模检索引擎MongoDBAtlasDocumentStore承担文档全生命周期管理MongoDBAtlasEmbeddingRetriever提供向量相似度检索MongoDBAtlasFullTextRetriever提供可配置模糊匹配、同义词与评分策略的全文检索。三者配合FilterPolicy过滤机制与DocumentJoiner结果融合即可在一条 Pipeline 中构建覆盖语义与关键词的混合检索系统。相关 API 细节可继续查阅本仓库的 MongoDB Atlas 参考文档。【免费下载链接】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），仅供参考