DocumentTypeRouter

파이프라인에서 문서를 MIME 타입에 따라 서로 다른 출력으로 라우팅해 추가 처리를 하게 해 주는 Router 컴포넌트예요.

출처: 문서

본문

파이프라인에서의 일반적인 위치: 특정 Converter나 Preprocessor로 보내기 전에 문서를 타입별로 라우팅하는 전처리 컴포넌트로 사용해요.
필수 초기화 변수: mime_types(분류에 사용할 MIME 타입 또는 정규 표현식 패턴의 리스트)
필수 실행 변수: documents(분류할 문서 리스트)
출력 변수: unclassified(분류되지 않은 문서 리스트), mime_types(예: "text/plain", "application/pdf", "image/jpeg" — 분류된 문서 리스트)

개요 (Overview)

DocumentTypeRouter는 문서를 MIME 타입에 따라 라우팅해요. 정확한 일치뿐 아니라 정규 표현식 패턴도 지원해요. MIME 타입을 문서 메타데이터에서 알아내거나, 표준 Python mimetypes 모듈과 커스텀 매핑을 사용해 파일 경로에서 추론할 수 있어요.

컴포넌트를 초기화할 때 별도의 출력으로 라우팅할 MIME 타입 집합을 지정해요. mime_types 파라미터에 타입 리스트를 설정하면 되는데, 예: ["text/plain", "audio/x-wav", "image/jpeg"]. 나열되지 않은 MIME 타입의 문서는 "unclassified"라는 출력으로 라우팅돼요.

MIME 타입을 결정하려면 다음 파라미터 중 하나 이상이 필요해요.

mime_type_meta_field: MIME 타입이 들어 있는 메타데이터 필드 이름
file_path_meta_field: 파일 경로가 들어 있는 메타데이터 필드 이름(MIME 타입은 파일 확장자에서 추론돼요)

단독 사용 (On its own)

DocumentTypeRouter로 문서를 MIME 타입별로 분류하는 예시예요.

from haystack.components.routers import DocumentTypeRouter
from haystack.dataclasses import Document

docs = [
    Document(content="Example text", meta={"file_path": "example.txt"}),
    Document(content="Another document", meta={"mime_type": "application/pdf"}),
    Document(content="Unknown type"),
]

router = DocumentTypeRouter(
    mime_type_meta_field="mime_type",
    file_path_meta_field="file_path",
    mime_types=["text/plain", "application/pdf"],
)

result = router.run(documents=docs)
print(result)

예상 출력:

{
    "text/plain": [Document(...)],
    "application/pdf": [Document(...)],
    "unclassified": [Document(...)],
}

정규 표현식 패턴 사용 (Using regex patterns)

정규 표현식 패턴을 사용하면 비슷한 패턴을 가진 여러 MIME 타입을 한 번에 매칭할 수 있어요.

from haystack.components.routers import DocumentTypeRouter
from haystack.dataclasses import Document

docs = [
    Document(content="Plain text", meta={"mime_type": "text/plain"}),
    Document(content="HTML text", meta={"mime_type": "text/html"}),
    Document(content="Markdown text", meta={"mime_type": "text/markdown"}),
    Document(content="JPEG image", meta={"mime_type": "image/jpeg"}),
    Document(content="PNG image", meta={"mime_type": "image/png"}),
    Document(content="PDF document", meta={"mime_type": "application/pdf"}),
]

router = DocumentTypeRouter(
    mime_type_meta_field="mime_type",
    mime_types=[r"text/.*", r"image/.*"],
)

result = router.run(documents=docs)

# Result will have:
# - "text/.*": 3 documents (text/plain, text/html, text/markdown)
# - "image/.*": 2 documents (image/jpeg, image/png)
# - "unclassified": 1 document (application/pdf)

커스텀 MIME 타입 사용 (Using custom MIME types)

흔하지 않은 파일 타입을 위해 커스텀 MIME 타입 매핑을 추가할 수 있어요.

from haystack.components.routers import DocumentTypeRouter
from haystack.dataclasses import Document

docs = [
    Document(content="Word document", meta={"file_path": "document.docx"}),
    Document(content="Markdown file", meta={"file_path": "readme.md"}),
    Document(content="Outlook message", meta={"file_path": "email.msg"}),
]

router = DocumentTypeRouter(
    file_path_meta_field="file_path",
    mime_types=[
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        "text/markdown",
        "application/vnd.ms-outlook",
    ],
    additional_mimetypes={
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
    },
)

result = router.run(documents=docs)

파이프라인에서 사용 (In a pipeline)

DocumentTypeRouter로 문서를 타입별로 분류한 뒤 서로 다르게 처리하는 파이프라인 예시예요. 텍스트 문서는 DocumentSplitter로 처리한 뒤 저장되고, PDF 문서는 바로 저장돼요.

from haystack import Pipeline
from haystack.components.routers import DocumentTypeRouter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.writers import DocumentWriter
from haystack.dataclasses import Document

# Create document store
document_store = InMemoryDocumentStore()

# Create pipeline
p = Pipeline()
p.add_component(
    instance=DocumentTypeRouter(
        mime_types=["text/plain", "application/pdf"],
        mime_type_meta_field="mime_type",
    ),
    name="document_type_router",
)
p.add_component(instance=DocumentSplitter(), name="text_splitter")
p.add_component(
    instance=DocumentWriter(document_store=document_store),
    name="text_writer",
)
p.add_component(
    instance=DocumentWriter(document_store=document_store),
    name="pdf_writer",
)

# Connect components
p.connect("document_type_router.text/plain", "text_splitter.documents")
p.connect("text_splitter.documents", "text_writer.documents")
p.connect("document_type_router.application/pdf", "pdf_writer.documents")

# Create test documents
docs = [
    Document(
        content="This is a text document that will be split and stored.",
        meta={"mime_type": "text/plain"},
    ),
    Document(
        content="This is a PDF document that will be stored directly.",
        meta={"mime_type": "application/pdf"},
    ),
    Document(
        content="This is an image document that will be unclassified.",
        meta={"mime_type": "image/jpeg"},
    ),
]

# Run pipeline
result = p.run({"document_type_router": {"documents": docs}})

# The pipeline will route documents based on their MIME types:
# - Text documents (text/plain) → DocumentSplitter → DocumentWriter
# - PDF documents (application/pdf) → DocumentWriter (direct)
# - Other documents → unclassified output

DocumentTypeRouter