DocumentJoiner

DocumentJoiner

DocumentJoiner는 여러 연결에서 온 입력 문서 리스트를 결합해 하나의 리스트로 출력해요. join_mode를 지정해 리스트를 어떻게 결합할지 선택할 수 있습니다. 네 가지 옵션이 있어요:

  • concatenate — 여러 컴포넌트의 문서를 중복을 버리며 결합합니다. 문서는 파이프라인에서 점수를 할당하는 마지막 컴포넌트의 점수를 받아요. 이 모드는 문서 점수에 영향을 주지 않습니다.
  • merge — 여러 컴포넌트에서 온 중복 문서의 점수를 병합합니다. 점수에 가중치를 할당해 병합 방식을 제어하고, top_k 제한을 설정해 DocumentJoiner가 반환할 문서 수를 지정할 수도 있어요.
  • reciprocal_rank_fusion — 여러 컴포넌트에서 받은 순위를 바탕으로 문서를 단일 리스트로 결합합니다. 그다음 입력 리스트의 문서 순위에 기반해 새 점수를 계산해요. 같은 Document가 둘 이상의 리스트에 나타나면(여러 컴포넌트가 반환) 더 높은 점수를 받습니다.
  • distribution_based_rank_fusion — 여러 출처의 순위를 단일의 통합 순위로 결합합니다. 점수가 어떻게 분포하는지 분석하고 정규화해서, 각 컴포넌트의 점수 방식을 고려하도록 합니다. 이 정규화는 각 컴포넌트의 영향을 균형 있게 해 더 견고하고 공정한 결합 순위를 만들어요. 문서가 여러 리스트에 나타나면, 모든 리스트의 점수 분포를 바탕으로 최종 점수가 조정됩니다.

출처: 문서

본문

Usage

On its own

두 문서 리스트를 병합하기 위해 DocumentJoiner를 사용하는 예시예요. DocumentJoiner를 실행하고 문서를 제공하면, 결합된 점수로 순위가 매겨진 문서 리스트를 반환합니다. 기본적으로 각 Retriever 점수에 동일한 가중치가 주어집니다. 입력 컴포넌트당 하나의 가중치인 float 리스트로 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)]}

In a pipeline

Hybrid Retrieval

아래는 InMemoryDocumentStore에서 키워드 검색(InMemoryBM25Retriever)과 임베딩 검색(InMemoryEmbeddingRetriever)으로 문서를 검색하는 하이브리드 검색 파이프라인 예시예요. 그다음 기본 결합 모드의 DocumentJoiner로 검색된 문서를 하나의 리스트로 연결합니다. Document Store에는 임베딩이 있는 문서가 있어야 하며, 그렇지 않으면 InMemoryEmbeddingRetriever가 문서를 반환하지 않습니다.

이 페이지의 예시는 sentence-transformers-haystack 패키지의 Sentence Transformers embedder를 사용해요. 예시를 실행하려면 설치하세요:

pip install sentence-transformers-haystack

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_integrations.components.embedders.sentence_transformers 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}})
Indexing

여기서는 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_integrations.components.embedders.sentence_transformers 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("**/*"))}},
)

Additional References

더 알아보기 (Learn more)