MetadataRouter
MetadataRouter
문서나 바이트 스트림을 메타데이터 필드 내용에 따라 서로 다른 출력 연결로 라우팅해주는 컴포넌트예요. DocumentLanguageClassifier처럼 문서를 분류하는 컴포넌트 뒤에 두면 돼요.
출처: MetadataRouter
본문
개요
MetadataRouter는 문서나 바이트 스트림을 메타데이터에 따라 서로 다른 출력으로 라우팅해요. rules로 출력 이름과, 문서·바이트 스트림을 연결 중 하나에 매칭할 필터를 정의해서 초기화해요. 필터는 Document Store의 필터와 같은 문법을 따라요. 문서나 바이트 스트림이 여러 필터와 일치하면 여러 출력으로 보내져요. 어떤 규칙에도 일치하지 않는 객체는 unmatched라는 이름의 출력 연결로 갑니다.
파이프라인에서 이 컴포넌트는 분류 결과를 문서의 메타데이터에 추가하는 Classifier(예: 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)
ByteStream 라우팅:
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(...)]}
파이프라인 안에서:
아래는 텍스트 파일을 문서로 변환하고, DocumentLanguageClassifier로 텍스트 언어를 감지해 문서 메타데이터에 추가하는 인덱싱 파이프라인 예시예요. 그런 뒤 MetadataRouter로 영어 문서만 DocumentWriter로 보내요. 다른 언어의 문서는 DocumentStore에 추가되지 않아요.
이 페이지의 예시들은 langdetect-haystack 패키지의 언어 분류 컴포넌트를 사용해요. 설치하려면:
pip install langdetect-haystack
from haystack import Pipeline
from haystack.components.converters import TextFileToDocument
from haystack_integrations.components.classifiers.langdetect 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",
],
},
},
)