
pipeline 中的最常见位置	在索引和查询管道中，在返回文档列表的组件（如多个Retrievers或多个Converters）之后使用。
强制运行变量	“documents”: 文档列表。此输入是`可变参数`，这意味着您可以连接任意数量的组件到它。
输出变量	“documents”：文档列表
API 参考	Joiners (连接器)
GitHub 链接	https://github.com/deepset-ai/haystack/blob/main/haystack/components/joiners/document_joiner.py

概述

DocumentJoiner 将来自多个连接的输入文档列表合并成一个列表输出。您可以通过指定join_mode 来选择您希望如何合并列表。有三种可用选项：

concatenate - 合并来自多个组件的文档，丢弃任何重复项。文档的分数来自管道中最后一个赋值分数的组件。此模式不影响文档分数。
merge - 合并来自多个组件的重复文档的分数。您还可以为分数分配权重，以影响它们如何合并，并设置 `top_k` 限制来指定您想要DocumentJoiner 返回多少文档。
reciprocal_rank_fusion - 根据文档在多个组件中获得的排名将其合并到一个列表中。然后，它根据文档在输入列表中的排名计算一个新的分数。如果同一文档出现在多个列表中（由多个组件返回），它将获得更高的分数。
distribution_based_rank_fusion – 将来自多个来源的排名合并到一个统一的排名中。它分析分数的分布情况并对其进行标准化，确保考虑每个组件的评分方法。这种标准化有助于平衡每个组件的影响力，从而实现更稳健和公平的组合排名。如果一个文档出现在多个列表中，其最终分数将根据所有列表的分数分布进行调整。

用法

单独使用

下面是一个示例，我们在其中使用DocumentJoiner 来合并两个文档列表。我们运行DocumentJoiner 并提供文档。它返回一个按组合分数排名的文档列表。默认情况下，每个检索器的分数都具有相等的权重。您还可以通过将 `weights` 参数设置为一个浮点数列表，其中包含每个输入组件一个权重，来使用自定义权重。

from haystack import Document
from haystack.components.joiners.document_joiner import DocumentJoiner

docs_1 = [Document(content="Paris is the capital of France.", score=0.5), Document(content="Berlin is the capital of Germany.", score=0.4)]
docs_2 = [Document(content="Paris is the capital of France.", score=0.6), Document(content="Rome is the capital of Italy.", score=0.5)]

joiner = DocumentJoiner(join_mode="merge")

joiner.run(documents=[docs_1, docs_2])

# {'documents': [Document(id=0f5beda04153dbfc462c8b31f8536749e43654709ecf0cfe22c6d009c9912214, content: 'Paris is the capital of France.', score: 0.55), Document(id=424beed8b549a359239ab000f33ca3b1ddb0f30a988bbef2a46597b9c27e42f2, content: 'Rome is the capital of Italy.', score: 0.25), Document(id=312b465e77e25c11512ee76ae699ce2eb201f34c8c51384003bb367e24fb6cf8, content: 'Berlin is the capital of Germany.', score: 0.2)]}

在 pipeline 中

混合检索

下面是一个混合检索管道的示例，该管道从InMemoryDocumentStore 基于关键字搜索（使用InMemoryBM25Retriever）检索文档，并进行嵌入搜索（使用InMemoryEmbeddingRetriever）。然后，它使用DocumentJoiner 及其默认的 `join_mode` 将检索到的文档合并到一个列表中。Document Store 必须包含带有嵌入的文档，否则InMemoryEmbeddingRetriever 将不会返回任何文档。

from haystack.components.joiners.document_joiner import DocumentJoiner
from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder

document_store = InMemoryDocumentStore()
p = Pipeline()
p.add_component(instance=InMemoryBM25Retriever(document_store=document_store), name="bm25_retriever")
p.add_component(
        instance=SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"),
        name="text_embedder",
    )
p.add_component(instance=InMemoryEmbeddingRetriever(document_store=document_store), name="embedding_retriever")
p.add_component(instance=DocumentJoiner(), name="joiner")
p.connect("bm25_retriever", "joiner")
p.connect("embedding_retriever", "joiner")
p.connect("text_embedder", "embedding_retriever")
query = "What is the capital of France?"
p.run(data={"bm25_retriever": {"query": query}, 
            "text_embedder": {"text": query}})

索引

这是一个索引管道的示例，该管道使用DocumentJoiner 将所有文件编译成一个文档列表，该列表可以作为整体通过其余的索引管道。

from haystack.components.writers import DocumentWriter
from haystack.components.converters import MarkdownToDocument, PyPDFToDocument, TextFileToDocument
from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner
from haystack.components.routers import FileTypeRouter
from haystack.components.joiners import DocumentJoiner
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from pathlib import Path

document_store = InMemoryDocumentStore()
file_type_router = FileTypeRouter(mime_types=["text/plain", "application/pdf", "text/markdown"])
text_file_converter = TextFileToDocument()
markdown_converter = MarkdownToDocument()
pdf_converter = PyPDFToDocument()
document_joiner = DocumentJoiner()

document_cleaner = DocumentCleaner()
document_splitter = DocumentSplitter(split_by="word", split_length=150, split_overlap=50)

document_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
document_writer = DocumentWriter(document_store)

preprocessing_pipeline = Pipeline()
preprocessing_pipeline.add_component(instance=file_type_router, name="file_type_router")
preprocessing_pipeline.add_component(instance=text_file_converter, name="text_file_converter")
preprocessing_pipeline.add_component(instance=markdown_converter, name="markdown_converter")
preprocessing_pipeline.add_component(instance=pdf_converter, name="pypdf_converter")
preprocessing_pipeline.add_component(instance=document_joiner, name="document_joiner")
preprocessing_pipeline.add_component(instance=document_cleaner, name="document_cleaner")
preprocessing_pipeline.add_component(instance=document_splitter, name="document_splitter")
preprocessing_pipeline.add_component(instance=document_embedder, name="document_embedder")
preprocessing_pipeline.add_component(instance=document_writer, name="document_writer")

preprocessing_pipeline.connect("file_type_router.text/plain", "text_file_converter.sources")
preprocessing_pipeline.connect("file_type_router.application/pdf", "pypdf_converter.sources")
preprocessing_pipeline.connect("file_type_router.text/markdown", "markdown_converter.sources")
preprocessing_pipeline.connect("text_file_converter", "document_joiner")
preprocessing_pipeline.connect("pypdf_converter", "document_joiner")
preprocessing_pipeline.connect("markdown_converter", "document_joiner")
preprocessing_pipeline.connect("document_joiner", "document_cleaner")
preprocessing_pipeline.connect("document_cleaner", "document_splitter")
preprocessing_pipeline.connect("document_splitter", "document_embedder")
preprocessing_pipeline.connect("document_embedder", "document_writer")

preprocessing_pipeline.run({"file_type_router": {"sources": list(Path(output_dir).glob("**/*"))}})

其他参考资料

📓 教程：预处理不同文件类型