Coverage for haystack/components/routers/metadata_router.py: 100%
38 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 09:42 +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 (`"unmatched"` is reserved), 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 :raises ValueError:
106 If `rules` contains the reserved output name `"unmatched"` or an invalid filter.
107 """
108 if "unmatched" in rules:
109 raise ValueError("The rule name 'unmatched' is reserved for documents that do not match any rule.")
111 self.rules = rules
112 self.output_type = output_type
113 self.strict_datetime_comparison = strict_datetime_comparison
114 for rule in self.rules.values():
115 if "operator" not in rule:
116 raise ValueError(
117 "Invalid filter syntax. See https://docs.haystack.deepset.ai/docs/metadata-filtering for details."
118 )
119 component.set_output_types(self, unmatched=self.output_type, **dict.fromkeys(rules, self.output_type))
121 def run(self, documents: list[Document] | list[ByteStream]) -> dict[str, list[Document] | list[ByteStream]]:
122 """
123 Routes documents or byte streams to different connections based on their metadata fields.
125 If a document or byte stream does not match any of the rules, it's routed to a connection named "unmatched".
127 :param documents: A list of `Document` or `ByteStream` objects to be routed based on their metadata.
129 :returns: A dictionary where the keys are the names of the output connections (including `"unmatched"`)
130 and the values are lists of `Document` or `ByteStream` objects that matched the corresponding rules.
131 """
133 unmatched: list[Document] | list[ByteStream] = []
134 output: dict[str, list[Document] | list[ByteStream]] = {edge: [] for edge in self.rules}
136 for doc_or_bytestream in documents:
137 current_obj_matched = False
138 for edge, rule in self.rules.items():
139 if document_matches_filter(
140 filters=rule, document=doc_or_bytestream, strict_datetime_comparison=self.strict_datetime_comparison
141 ):
142 # we need to ignore the arg-type here because the underlying
143 # filter methods use type Union[Document, ByteStream]
144 output[edge].append(doc_or_bytestream) # type: ignore[arg-type]
145 current_obj_matched = True
147 if not current_obj_matched:
148 unmatched.append(doc_or_bytestream) # type: ignore[arg-type]
150 output["unmatched"] = unmatched
151 return output
153 def to_dict(self) -> dict[str, Any]:
154 """
155 Serialize this component to a dictionary.
157 :returns:
158 The serialized component as a dictionary.
159 """
160 return default_to_dict(
161 self,
162 rules=self.rules,
163 output_type=serialize_type(self.output_type),
164 strict_datetime_comparison=self.strict_datetime_comparison,
165 )
167 @classmethod
168 def from_dict(cls, data: dict[str, Any]) -> "MetadataRouter":
169 """
170 Deserialize this component from a dictionary.
172 :param data:
173 The dictionary representation of this component.
174 :returns:
175 The deserialized component instance.
176 """
177 init_params = data.get("init_parameters", {})
178 if "output_type" in init_params:
179 # Deserialize the output_type to its original type
180 init_params["output_type"] = deserialize_type(init_params["output_type"])
181 return default_from_dict(cls, data)