
pipeline 中的最常见位置	1. 在 RAG 管道中的 ChatPromptBuilder 之前 2. 语义搜索管道的最后一个组件 3. 在抽取式 QA 管道中的 ExtractiveReader 之前
必需的初始化变量	"document_store": MongoDBAtlasDocumentStore 的一个实例
强制运行变量	“query”：要搜索的查询字符串。如果查询包含多个术语，Atlas Search 会单独评估每个术语是否匹配。
输出变量	“documents”：文档列表
API 参考	MongoDB Atlas
GitHub 链接	https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/mongodb_atlas

该MongoDBAtlasFullTextRetriever 是与 MongoDBAtlasDocumentStore 兼容的全文本搜索 Retriever。全文本搜索取决于MongoDBAtlasDocumentStore 中使用的 full_text_search_index。

参数

除了query，MongoDBAtlasFullTextRetriever 还接受其他可选参数，包括top_k（要检索的文档的最大数量）和filters（用于缩小搜索范围）。

在运行组件时，您可以指定更多可选参数，例如fuzzy 或同义词, match_criteria, score。有关所有参数的更多详细信息，请查看我们的 MongoDB Atlas API 参考。

用法

安装

要开始使用 MongoDB Atlas 和 Haystack，请使用以下命令安装该软件包

pip install mongodb-atlas-haystack

单独使用

Retriever 需要一个 MongoDBAtlasDocumentStore 的实例和已索引的文档才能运行。MongoDBAtlasDocumentStore 和已索引的文档才能运行。

from haystack_integrations.document_stores.mongodb_atlas import MongoDBAtlasDocumentStore
from haystack_integrations.components.retrievers.mongodb_atlas import MongoDBAtlasFullTextRetriever

store = MongoDBAtlasDocumentStore(database_name="your_existing_db",
                                  collection_name="your_existing_collection",
                                  vector_search_index="your_existing_index",
                                  full_text_search_index="your_existing_index")
retriever = MongoDBAtlasFullTextRetriever(document_store=store)

results = retriever.run(query="Your search query")
print(results["documents"])

在 Pipeline 中

这是一个利用了两个可用的 MongoDB Atlas Retriever 的混合检索管道示例

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(content="My name is Jean and I live in Paris."),
    Document(content="My name is Mark and I live in Berlin."),
    Document(content="My name is Giorgio and I live in Rome."),
    Document(content="Python is a programming language popular for data science."),
    Document(content="MongoDB Atlas offers full-text search and vector search capabilities."),
]

document_store = MongoDBAtlasDocumentStore(
    database_name="haystack_test",
    collection_name="test_collection",
    vector_search_index="test_vector_search_index",
    full_text_search_index="test_full_text_search_index",
)

# Clean out any old data so this example is repeatable
print(f"Clearing collection {document_store.collection_name} …")
document_store.collection.delete_many({})

ingest_pipe = Pipeline()

doc_embedder = SentenceTransformersDocumentEmbedder(model="intfloat/e5-base-v2")
ingest_pipe.add_component(instance=doc_embedder, name="doc_embedder")

doc_writer = DocumentWriter(
    document_store=document_store,
    policy=DuplicatePolicy.SKIP
)
ingest_pipe.add_component(instance=doc_writer, name="doc_writer")
ingest_pipe.connect("doc_embedder.documents", "doc_writer.documents")

print(f"Running ingestion on {len(documents)} in-memory docs …")
ingest_pipe.run({"doc_embedder": {"documents": documents}})

query_pipe = Pipeline()

text_embedder = SentenceTransformersTextEmbedder(model="intfloat/e5-base-v2")
query_pipe.add_component(instance=text_embedder, name="text_embedder")

embed_retriever = MongoDBAtlasEmbeddingRetriever(
    document_store=document_store,
    top_k=3
)
query_pipe.add_component(instance=embed_retriever, name="embedding_retriever")
query_pipe.connect("text_embedder", "embedding_retriever")

# (c) full-text retriever
ft_retriever = MongoDBAtlasFullTextRetriever(
    document_store=document_store,
    top_k=3
)
query_pipe.add_component(instance=ft_retriever, name="full_text_retriever")

joiner = DocumentJoiner(join_mode="reciprocal_rank_fusion", top_k=3)
query_pipe.add_component(instance=joiner, name="joiner")

query_pipe.connect("embedding_retriever", "joiner")
query_pipe.connect("full_text_retriever", "joiner")

question = "Where does Mark live?"
print(f"Running hybrid retrieval for query: '{question}'")
output = query_pipe.run(
    {
        "text_embedder": {"text": question},
        "full_text_retriever": {"query": question},
    }
)

print("\nFinal fused documents:")
for doc in output["joiner"]["documents"]:
    print(f"- {doc.content}")