
pipeline 中的最常见位置	作为预处理组件，在将文档发送到特定的转换器或预处理器之前，按类型路由文档。
必需的初始化变量	"mime_types": 用于分类的 MIME 类型或正则表达式模式列表。
强制运行变量	"documents": 要分类的文档列表。
输出变量	"unclassified": 未分类的文档列表。 "mime_types": 例如 "text/plain", "application/pdf", "image/jpeg": 分类的文档列表。
API 参考	Routers (路由器)
GitHub 链接	https://github.com/deepset-ai/haystack/blob/main/haystack/components/routers/document_type_router.py

概述

DocumentTypeRouter 根据文档的 MIME 类型路由文档，支持精确匹配和正则表达式模式。它可以从文档元数据中确定 MIME 类型，或使用标准的 Python mimetypes 模块和自定义映射从文件路径中推断。mimetypes 模块和自定义映射。

在初始化组件时，指定要路由到单独输出的 MIME 类型集合。将mime_types 参数设置为一个类型列表，例如["text/plain", "audio/x-wav", "image/jpeg"]。MIME 类型不在列表中的文档将被路由到名为 "unclassified" 的输出。

该组件需要以下至少一个参数来确定 MIME 类型：

mime_type_meta_field：包含 MIME 类型的元数据字段名称。
file_path_meta_field：包含文件路径的元数据字段名称（MIME 类型将从文件扩展名推断）。

用法

单独使用

下面是一个使用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(...)]
}

使用正则表达式模式

您可以使用正则表达式模式来匹配具有相似模式的多个 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 类型

您可以为不常见的 Mime 类型添加自定义 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)

在 pipeline 中

下面是一个 pipeline 的示例，该 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