c56bef871b
Sync docs with Docusaurus / sync (push) Waiting to run
Tests / Check if changed (push) Waiting to run
Tests / format (push) Blocked by required conditions
Tests / check-imports (push) Blocked by required conditions
Tests / Unit / macos-latest (push) Blocked by required conditions
Tests / Unit / ubuntu-latest (push) Blocked by required conditions
Tests / Unit / windows-latest (push) Blocked by required conditions
Tests / mypy (push) Blocked by required conditions
Tests / Integration / ubuntu-latest (push) Blocked by required conditions
Tests / Integration / macos-latest (push) Blocked by required conditions
Tests / Integration / windows-latest (push) Blocked by required conditions
Tests / notify-slack-on-failure (push) Blocked by required conditions
Tests / Mark tests as completed (push) Blocked by required conditions
Docker image release / Build base image (push) Waiting to run
CodeQL / Analyze (python) (push) Has been cancelled
Update Platform Components Table / update (push) Has been cancelled
164 lines
6.7 KiB
Python
164 lines
6.7 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from typing import Any
|
|
|
|
from haystack import Document, component, default_from_dict, default_to_dict
|
|
from haystack.dataclasses import ByteStream
|
|
from haystack.utils import deserialize_type, serialize_type
|
|
from haystack.utils.filters import document_matches_filter
|
|
|
|
|
|
@component
|
|
class MetadataRouter:
|
|
"""
|
|
Routes documents or byte streams to different connections based on their metadata fields.
|
|
|
|
Specify the routing rules in the `init` method.
|
|
If a document or byte stream does not match any of the rules, it's routed to a connection named "unmatched".
|
|
|
|
|
|
### Usage examples
|
|
|
|
**Routing Documents by metadata:**
|
|
```python
|
|
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"}})
|
|
|
|
print(router.run(documents=docs))
|
|
# {'en': [Document(id=..., content: 'Paris is the capital of France.', meta: {'language': 'en'})],
|
|
# 'unmatched': [Document(id=..., content: 'Berlin ist die Haupststadt von Deutschland.', meta: {'language': 'de'})]}
|
|
```
|
|
|
|
**Routing ByteStreams by metadata:**
|
|
```python
|
|
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(...)]}
|
|
```
|
|
"""
|
|
|
|
def __init__(self, rules: dict[str, dict], output_type: type = list[Document]) -> None:
|
|
"""
|
|
Initializes the MetadataRouter component.
|
|
|
|
:param rules: A dictionary defining how to route documents or byte streams to output connections based on their
|
|
metadata. Keys are output connection names, and values are dictionaries of
|
|
[filtering expressions](https://docs.haystack.deepset.ai/docs/metadata-filtering) in Haystack.
|
|
For example:
|
|
```python
|
|
{
|
|
"edge_1": {
|
|
"operator": "AND",
|
|
"conditions": [
|
|
{"field": "meta.created_at", "operator": ">=", "value": "2023-01-01"},
|
|
{"field": "meta.created_at", "operator": "<", "value": "2023-04-01"},
|
|
],
|
|
},
|
|
"edge_2": {
|
|
"operator": "AND",
|
|
"conditions": [
|
|
{"field": "meta.created_at", "operator": ">=", "value": "2023-04-01"},
|
|
{"field": "meta.created_at", "operator": "<", "value": "2023-07-01"},
|
|
],
|
|
},
|
|
"edge_3": {
|
|
"operator": "AND",
|
|
"conditions": [
|
|
{"field": "meta.created_at", "operator": ">=", "value": "2023-07-01"},
|
|
{"field": "meta.created_at", "operator": "<", "value": "2023-10-01"},
|
|
],
|
|
},
|
|
"edge_4": {
|
|
"operator": "AND",
|
|
"conditions": [
|
|
{"field": "meta.created_at", "operator": ">=", "value": "2023-10-01"},
|
|
{"field": "meta.created_at", "operator": "<", "value": "2024-01-01"},
|
|
],
|
|
},
|
|
}
|
|
```
|
|
:param output_type: The type of the output produced. Lists of Documents or ByteStreams can be specified.
|
|
"""
|
|
self.rules = rules
|
|
self.output_type = output_type
|
|
for rule in self.rules.values():
|
|
if "operator" not in rule:
|
|
raise ValueError(
|
|
"Invalid filter syntax. See https://docs.haystack.deepset.ai/docs/metadata-filtering for details."
|
|
)
|
|
component.set_output_types(self, unmatched=self.output_type, **dict.fromkeys(rules, self.output_type))
|
|
|
|
def run(self, documents: list[Document] | list[ByteStream]) -> dict[str, list[Document] | list[ByteStream]]:
|
|
"""
|
|
Routes documents or byte streams to different connections based on their metadata fields.
|
|
|
|
If a document or byte stream does not match any of the rules, it's routed to a connection named "unmatched".
|
|
|
|
:param documents: A list of `Document` or `ByteStream` objects to be routed based on their metadata.
|
|
|
|
:returns: A dictionary where the keys are the names of the output connections (including `"unmatched"`)
|
|
and the values are lists of `Document` or `ByteStream` objects that matched the corresponding rules.
|
|
"""
|
|
|
|
unmatched: list[Document] | list[ByteStream] = []
|
|
output: dict[str, list[Document] | list[ByteStream]] = {edge: [] for edge in self.rules}
|
|
|
|
for doc_or_bytestream in documents:
|
|
current_obj_matched = False
|
|
for edge, rule in self.rules.items():
|
|
if document_matches_filter(filters=rule, document=doc_or_bytestream):
|
|
# we need to ignore the arg-type here because the underlying
|
|
# filter methods use type Union[Document, ByteStream]
|
|
output[edge].append(doc_or_bytestream) # type: ignore[arg-type]
|
|
current_obj_matched = True
|
|
|
|
if not current_obj_matched:
|
|
unmatched.append(doc_or_bytestream) # type: ignore[arg-type]
|
|
|
|
output["unmatched"] = unmatched
|
|
return output
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Serialize this component to a dictionary.
|
|
|
|
:returns:
|
|
The serialized component as a dictionary.
|
|
"""
|
|
return default_to_dict(self, rules=self.rules, output_type=serialize_type(self.output_type))
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "MetadataRouter":
|
|
"""
|
|
Deserialize this component from a dictionary.
|
|
|
|
:param data:
|
|
The dictionary representation of this component.
|
|
:returns:
|
|
The deserialized component instance.
|
|
"""
|
|
init_params = data.get("init_parameters", {})
|
|
if "output_type" in init_params:
|
|
# Deserialize the output_type to its original type
|
|
init_params["output_type"] = deserialize_type(init_params["output_type"])
|
|
return default_from_dict(cls, data)
|