
pipeline 中的最常见位置	在像 `DocumentLanguageClassifier` 这样的组件对文档进行分类后，
必需的初始化变量	"rules": 一个包含元数据路由规则的字典（有关示例，请参阅我们的 API 参考）
强制运行变量	“documents”：文档或字节流列表
输出变量	“unmatched”：未匹配任何规则的文档或字节流列表 “rule_name”：匹配自定义规则的文档或字节流列表。您定义的每个规则都有一个输出。这些输出中的每一个都是文档或字节流的列表。
API 参考	Routers (路由器)
GitHub 链接	https://github.com/deepset-ai/haystack/blob/main/haystack/components/routers/metadata_router.py

概述

MetadataRouter 根据元数据将文档或字节流路由到不同的输出。您通过以下方式初始化它：rules 定义输出名称和匹配文档或字节流到其中一个连接的过滤器。这些过滤器遵循与 Document Stores 中的过滤器相同的语法。如果文档或字节流匹配多个过滤器，它将被发送到多个输出。不匹配任何规则的对象将转到名为unmatched.

在管道中，此组件在分类器（例如DocumentLanguageClassifier）之后最为有用，该分类器将分类结果添加到文档的元数据中。

此组件没有默认规则。如果您在初始化组件时未定义任何规则，它会将所有文档或字节流路由到unmatched 输出。

用法

单独使用

下面是一个使用MetadataRouter 以根据其元数据过滤文档。我们通过设置一条规则来初始化路由器，该规则将所有元数据中language 设置为en 的文档传递到名为en 的输出连接。不匹配此规则的文档将转到名为unmatched.

from haystack import Document
from haystack.components.routers import MetadataRouter

docs = [Document(content="Paris is the capital of France.", meta={"language": "en"}), Document(content="Berlin ist die Haupststadt von Deutschland.", meta={"language": "de"})]
router = MetadataRouter(rules={"en": {"field": "meta.language", "operator": "==", "value": "en"}})
router.run(documents=docs)

路由字节流

您也可以使用MetadataRouter 来路由ByteStream 对象，具体取决于它们的元数据。这在使用二进制数据或在将文件转换为文档之前需要路由文件时非常有用。

from haystack.dataclasses import ByteStream
from haystack.components.routers import MetadataRouter

streams = [
    ByteStream.from_string("Hello world", meta={"language": "en"}),
    ByteStream.from_string("Bonjour le monde", meta={"language": "fr"})
]

router = MetadataRouter(
    rules={"english": {"field": "meta.language", "operator": "==", "value": "en"}},
    output_type=list[ByteStream]
)

result = router.run(documents=streams)
# {'english': [ByteStream(...)], 'unmatched': [ByteStream(...)]}

在 pipeline 中

下面是一个索引管道的示例，该管道将文本文件转换为文档，并使用DocumentLanguageClassifier 检测文本的语言并将其添加到文档的元数据中。然后，它使用MetadataRouter 将仅限英语的文档转发到DocumentWriter。其他语言的文档不会添加到DocumentStore (文档存储).

from haystack import Pipeline
from haystack.components.file_converters import TextFileToDocument
from haystack.components.classifiers import DocumentLanguageClassifier
from haystack.components.routers import MetadataRouter
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
p = Pipeline()
p.add_component(instance=TextFileToDocument(), name="text_file_converter")
p.add_component(instance=DocumentLanguageClassifier(), name="language_classifier")
p.add_component(
    instance=MetadataRouter(rules={"en": {"field": "meta.language", "operator": "==", "value": "en"}}), name="router"
)
p.add_component(instance=DocumentWriter(document_store=document_store), name="writer")
p.connect("text_file_converter.documents", "language_classifier.documents")
p.connect("language_classifier.documents", "router.documents")
p.connect("router.en", "writer.documents")
p.run({"text_file_converter": {"sources": ["english-file-will-be-added.txt", "german-file-will-not-be-added.txt"]}})