Coverage for haystack/components/routers/metadata_router.py: 100%
36 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from typing import Any
7from haystack import Document, component, default_from_dict, default_to_dict
8from haystack.dataclasses import ByteStream
9from haystack.utils import deserialize_type, serialize_type
10from haystack.utils.filters import document_matches_filter
13@component
14class MetadataRouter:
15 """
16 Routes documents or byte streams to different connections based on their metadata fields.
18 Specify the routing rules in the `init` method.
19 If a document or byte stream does not match any of the rules, it's routed to a connection named "unmatched".
22 ### Usage examples
24 **Routing Documents by metadata:**
25 ```python
26 from haystack import Document
27 from haystack.components.routers import MetadataRouter
29 docs = [Document(content="Paris is the capital of France.", meta={"language": "en"}),
30 Document(content="Berlin ist die Haupststadt von Deutschland.", meta={"language": "de"})]
32 router = MetadataRouter(rules={"en": {"field": "meta.language", "operator": "==", "value": "en"}})
34 print(router.run(documents=docs))
35 # {'en': [Document(id=..., content: 'Paris is the capital of France.', meta: {'language': 'en'})],
36 # 'unmatched': [Document(id=..., content: 'Berlin ist die Haupststadt von Deutschland.', meta: {'language': 'de'})]}
37 ```
39 **Routing ByteStreams by metadata:**
40 ```python
41 from haystack.dataclasses import ByteStream
42 from haystack.components.routers import MetadataRouter
44 streams = [
45 ByteStream.from_string("Hello world", meta={"language": "en"}),
46 ByteStream.from_string("Bonjour le monde", meta={"language": "fr"})
47 ]
49 router = MetadataRouter(
50 rules={"english": {"field": "meta.language", "operator": "==", "value": "en"}},
51 output_type=list[ByteStream]
52 )
54 result = router.run(documents=streams)
55 # {'english': [ByteStream(...)], 'unmatched': [ByteStream(...)]}
56 ```
57 """
59 def __init__(
60 self, rules: dict[str, dict], output_type: type = list[Document], *, strict_datetime_comparison: bool = False
61 ) -> None:
62 """
63 Initializes the MetadataRouter component.
65 :param rules: A dictionary defining how to route documents or byte streams to output connections based on their
66 metadata. Keys are output connection names, and values are dictionaries of
67 [filtering expressions](https://docs.haystack.deepset.ai/docs/metadata-filtering) in Haystack.
68 For example:
69 ```python
70 {
71 "edge_1": {
72 "operator": "AND",
73 "conditions": [
74 {"field": "meta.created_at", "operator": ">=", "value": "2023-01-01"},
75 {"field": "meta.created_at", "operator": "<", "value": "2023-04-01"},
76 ],
77 },
78 "edge_2": {
79 "operator": "AND",
80 "conditions": [
81 {"field": "meta.created_at", "operator": ">=", "value": "2023-04-01"},
82 {"field": "meta.created_at", "operator": "<", "value": "2023-07-01"},
83 ],
84 },
85 "edge_3": {
86 "operator": "AND",
87 "conditions": [
88 {"field": "meta.created_at", "operator": ">=", "value": "2023-07-01"},
89 {"field": "meta.created_at", "operator": "<", "value": "2023-10-01"},
90 ],
91 },
92 "edge_4": {
93 "operator": "AND",
94 "conditions": [
95 {"field": "meta.created_at", "operator": ">=", "value": "2023-10-01"},
96 {"field": "meta.created_at", "operator": "<", "value": "2024-01-01"},
97 ],
98 },
99 }
100 ```
101 :param output_type: The type of the output produced. Lists of Documents or ByteStreams can be specified.
102 :param strict_datetime_comparison:
103 If `True`, timezone-naive and timezone-aware datetimes never match each other.
104 If `False` (the default), the timezone from the aware datetime is copied to the naive one before comparing.
105 """
106 self.rules = rules
107 self.output_type = output_type
108 self.strict_datetime_comparison = strict_datetime_comparison
109 for rule in self.rules.values():
110 if "operator" not in rule:
111 raise ValueError(
112 "Invalid filter syntax. See https://docs.haystack.deepset.ai/docs/metadata-filtering for details."
113 )
114 component.set_output_types(self, unmatched=self.output_type, **dict.fromkeys(rules, self.output_type))
116 def run(self, documents: list[Document] | list[ByteStream]) -> dict[str, list[Document] | list[ByteStream]]:
117 """
118 Routes documents or byte streams to different connections based on their metadata fields.
120 If a document or byte stream does not match any of the rules, it's routed to a connection named "unmatched".
122 :param documents: A list of `Document` or `ByteStream` objects to be routed based on their metadata.
124 :returns: A dictionary where the keys are the names of the output connections (including `"unmatched"`)
125 and the values are lists of `Document` or `ByteStream` objects that matched the corresponding rules.
126 """
128 unmatched: list[Document] | list[ByteStream] = []
129 output: dict[str, list[Document] | list[ByteStream]] = {edge: [] for edge in self.rules}
131 for doc_or_bytestream in documents:
132 current_obj_matched = False
133 for edge, rule in self.rules.items():
134 if document_matches_filter(
135 filters=rule, document=doc_or_bytestream, strict_datetime_comparison=self.strict_datetime_comparison
136 ):
137 # we need to ignore the arg-type here because the underlying
138 # filter methods use type Union[Document, ByteStream]
139 output[edge].append(doc_or_bytestream) # type: ignore[arg-type]
140 current_obj_matched = True
142 if not current_obj_matched:
143 unmatched.append(doc_or_bytestream) # type: ignore[arg-type]
145 output["unmatched"] = unmatched
146 return output
148 def to_dict(self) -> dict[str, Any]:
149 """
150 Serialize this component to a dictionary.
152 :returns:
153 The serialized component as a dictionary.
154 """
155 return default_to_dict(
156 self,
157 rules=self.rules,
158 output_type=serialize_type(self.output_type),
159 strict_datetime_comparison=self.strict_datetime_comparison,
160 )
162 @classmethod
163 def from_dict(cls, data: dict[str, Any]) -> "MetadataRouter":
164 """
165 Deserialize this component from a dictionary.
167 :param data:
168 The dictionary representation of this component.
169 :returns:
170 The deserialized component instance.
171 """
172 init_params = data.get("init_parameters", {})
173 if "output_type" in init_params:
174 # Deserialize the output_type to its original type
175 init_params["output_type"] = deserialize_type(init_params["output_type"])
176 return default_from_dict(cls, data)