Repository: deepset-ai/haystack
Pull request: feat: Add `MetadataRouter` (2.0)
### Related Issues

- fixes #5680 
- depends on #5823

### Proposed Changes:

 <!--- In case of a bug: Describe what caused the issue and how you solved it -->
 <!--- In case of a feature: Describe what did you add and how it works -->

This PR adds the `MetadataRouter`, a component which allows to route documents to different edges based on their metadata and specified rules. The rules should be in the form of filter expressions.

### How did you test it?

<!-- unit tests, integration tests, manual verification, instructions for manual tests -->
I added unit tests.

### Notes for the reviewer
To be consistent with names, I renamed `FileExtensionClassifier` to `FileExtensionRouter` and moved it to the `routers` package.

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
`MetadataRouter`
An additional router with a similar purpose to v1's `RouteDocuments`. This component is able to route documents on different output connections according to the content of their metadata fields.

It can use a filtering logic similar to `MemoryDocumentStore`’s filtering, and possibly even reuse it.

Draft I/O for `MetadataRouter`:
```python
@component
class MetadataRouter:
    
    def __init__(self, rules: Dict[str, Any] = None):
        """
        Rules should be in the form {'connection name': filter to match}.
        For example:
        { 
             'q1': {'created_at': {'$gt': '2023-01-01', '$lte': '2023-04-01'}},
             'q2': {'created_at': {'$gt': '2023-04-01', '$lte': '2023-07-01'}},
             'q3': {'created_at': {'$gt': '2023-07-01', '$lte': '2023-10-01'}},
             'q4': {'created_at': {'$gt': '2023-10-01', '$lte': '2024-01-01'}},
        }
        """
        self.rules = rules
        component.set_output_types(**{key: List[Document] for key in rules})

    def run(self, documents: List[Document]):
        # matches each document to the filters
        return {"q1": q1_docs, "q2": q2_docs, ...}
```

Hints from issue discussion:

Commit message: Merge branch 'main' into metadata_router_2.0

Patch to review:
diff --git a/haystack/preview/components/classifiers/file_classifier.py b/haystack/preview/components/routers/file_router.py
similarity index 86%
rename from haystack/preview/components/classifiers/file_classifier.py
rename to haystack/preview/components/routers/file_router.py
index 86ce63e667..78604f92df 100644
--- a/haystack/preview/components/classifiers/file_classifier.py
+++ b/haystack/preview/components/routers/file_router.py
@@ -10,12 +10,12 @@
 
 
 @component
-class FileExtensionClassifier:
+class FileExtensionRouter:
     """
-    A component that classifies files based on their MIME types read from their file extensions. This component
+    A component that routes files based on their MIME types read from their file extensions. This component
     does not read the file contents, but rather uses the file extension to determine the MIME type of the file.
 
-    The FileExtensionClassifier takes a list of file paths and groups them by their MIME types.
+    The FileExtensionRouter takes a list of file paths and groups them by their MIME types.
     The list of MIME types to consider is provided during the initialization of the component.
 
     This component is particularly useful when working with a large number of files, and you
@@ -24,9 +24,9 @@ class FileExtensionClassifier:
 
     def __init__(self, mime_types: List[str]):
         """
-        Initialize the FileExtensionClassifier.
+        Initialize the FileExtensionRouter.
 
-        :param mime_types: A list of file mime types to consider when classifying
+        :param mime_types: A list of file mime types to consider when routing
         files (e.g. ["text/plain", "audio/x-wav", "image/jpeg"]).
         """
         if not mime_types:
@@ -48,7 +48,7 @@ def to_dict(self) -> Dict[str, Any]:
         return default_to_dict(self, mime_types=self.mime_types)
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> "FileExtensionClassifier":
+    def from_dict(cls, data: Dict[str, Any]) -> "FileExtensionRouter":
         """
         Deserialize this component from a dictionary.
         """
@@ -56,13 +56,13 @@ def from_dict(cls, data: Dict[str, Any]) -> "FileExtensionClassifier":
 
     def run(self, paths: List[Union[str, Path]]):
         """
-        Run the FileExtensionClassifier.
+        Run the FileExtensionRouter.
 
         This method takes the input data, iterates through the provided file paths, checks the file
         mime type of each file, and groups the file paths by their mime types.
 
-        :param paths: The input data containing the file paths to classify.
-        :return: The output data containing the classified file paths.
+        :param paths: The input data containing the file paths to route.
+        :return: The output data containing the routed file paths.
         """
         mime_types = defaultdict(list)
         for path in paths:
diff --git a/haystack/preview/components/routers/metadata_router.py b/haystack/preview/components/routers/metadata_router.py
new file mode 100644
index 0000000000..490b671638
--- /dev/null
+++ b/haystack/preview/components/routers/metadata_router.py
@@ -0,0 +1,67 @@
+from typing import Any, Dict, List
+
+from haystack.preview import component, default_from_dict, default_to_dict, Document
+from haystack.preview.utils.filters import document_matches_filter
+
+
+@component
+class MetadataRouter:
+    """
+    A component that routes documents to different edges based on their metadata and specified rules.
+    """
+
+    def __init__(self, rules: Dict[str, Dict]):
+        """
+        Initialize the MetadataRouter.
+
+        :param rules: A dictionary of rules that specify which edge to route a document to based on its metadata.
+                      The keys of the dictionary are the names of the edges, and the values are dictionaries that
+                      follow the format of filtering expressions in Haystack. For example:
+                      ```python
+                      {
+                            "edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}},
+                            "edge_2": {"created_at": {"$gte": "2023-04-01", "$lt": "2023-07-01"}},
+                            "edge_3": {"created_at": {"$gte": "2023-07-01", "$lt": "2023-10-01"}},
+                            "edge_4": {"created_at": {"$gte": "2023-10-01", "$lt": "2024-01-01"}},
+                      }
+                      ```
+        """
+        self.rules = rules
+        component.set_output_types(self, unmatched=List[Document], **{edge: List[Document] for edge in rules})
+
+    def run(self, documents: List[Document]):
+        """
+        Run the MetadataRouter. This method routes the documents to different edges based on their metadata and
+        the rules specified during initialization. If a document does not match any of the rules, it is routed to
+        an edge named "unmatched".
+
+        :param documents: A list of documents to route to different edges.
+        """
+        unmatched_documents = []
+        output: Dict[str, List[Document]] = {edge: [] for edge in self.rules}
+
+        for document in documents:
+            cur_document_matched = False
+            for edge, rule in self.rules.items():
+                if document_matches_filter(rule, document):
+                    output[edge].append(document)
+                    cur_document_matched = True
+
+            if not cur_document_matched:
+                unmatched_documents.append(document)
+
+        output["unmatched"] = unmatched_documents
+        return output
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Serialize this component to a dictionary.
+        """
+        return default_to_dict(self, rules=self.rules)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "MetadataRouter":
+        """
+        Deserialize this component from a dictionary.
+        """
+        return default_from_dict(cls, data)
diff --git a/releasenotes/notes/add-metadata_router-2.0-63829ac5a0528e9d.yaml b/releasenotes/notes/add-metadata_router-2.0-63829ac5a0528e9d.yaml
new file mode 100644
index 0000000000..2c34fc8791
--- /dev/null
+++ b/releasenotes/notes/add-metadata_router-2.0-63829ac5a0528e9d.yaml
@@ -0,0 +1,4 @@
+---
+preview:
+  - |
+    Add `MetadataRouter`, a component that routes documents to different edges based on their metadata and specified rules.
diff --git a/test/preview/components/classifiers/__init__.py b/test/preview/components/routers/__init__.py
similarity index 100%
rename from test/preview/components/classifiers/__init__.py
rename to test/preview/components/routers/__init__.py
diff --git a/test/preview/components/classifiers/test_file_classifier.py b/test/preview/components/routers/test_file_router.py
similarity index 74%
rename from test/preview/components/classifiers/test_file_classifier.py
rename to test/preview/components/routers/test_file_router.py
index 833b138480..d3772d941e 100644
--- a/test/preview/components/classifiers/test_file_classifier.py
+++ b/test/preview/components/routers/test_file_router.py
@@ -2,30 +2,30 @@
 
 import pytest
 
-from haystack.preview.components.classifiers.file_classifier import FileExtensionClassifier
+from haystack.preview.components.routers.file_router import FileExtensionRouter
 
 
 @pytest.mark.skipif(
     sys.platform in ["win32", "cygwin"],
     reason="Can't run on Windows Github CI, need access to registry to get mime types",
 )
-class TestFileExtensionClassifier:
+class TestFileExtensionRouter:
     @pytest.mark.unit
     def test_to_dict(self):
-        component = FileExtensionClassifier(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
+        component = FileExtensionRouter(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
         data = component.to_dict()
         assert data == {
-            "type": "FileExtensionClassifier",
+            "type": "FileExtensionRouter",
             "init_parameters": {"mime_types": ["text/plain", "audio/x-wav", "image/jpeg"]},
         }
 
     @pytest.mark.unit
     def test_from_dict(self):
         data = {
-            "type": "FileExtensionClassifier",
+            "type": "FileExtensionRouter",
             "init_parameters": {"mime_types": ["text/plain", "audio/x-wav", "image/jpeg"]},
         }
-        component = FileExtensionClassifier.from_dict(data)
+        component = FileExtensionRouter.from_dict(data)
         assert component.mime_types == ["text/plain", "audio/x-wav", "image/jpeg"]
 
     @pytest.mark.unit
@@ -40,8 +40,8 @@ def test_run(self, preview_samples_path):
             preview_samples_path / "images" / "apple.jpg",
         ]
 
-        classifier = FileExtensionClassifier(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
-        output = classifier.run(paths=file_paths)
+        router = FileExtensionRouter(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
+        output = router.run(paths=file_paths)
         assert output
         assert len(output["text/plain"]) == 2
         assert len(output["audio/x-wav"]) == 1
@@ -53,8 +53,8 @@ def test_no_files(self):
         """
         Test that the component runs correctly when no files are provided.
         """
-        classifier = FileExtensionClassifier(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
-        output = classifier.run(paths=[])
+        router = FileExtensionRouter(mime_types=["text/plain", "audio/x-wav", "image/jpeg"])
+        output = router.run(paths=[])
         assert not output
 
     @pytest.mark.unit
@@ -67,8 +67,8 @@ def test_unlisted_extensions(self, preview_samples_path):
             preview_samples_path / "audio" / "ignored.mp3",
             preview_samples_path / "audio" / "this is the content of the document.wav",
         ]
-        classifier = FileExtensionClassifier(mime_types=["text/plain"])
-        output = classifier.run(paths=file_paths)
+        router = FileExtensionRouter(mime_types=["text/plain"])
+        output = router.run(paths=file_paths)
         assert len(output["text/plain"]) == 1
         assert "mp3" not in output
         assert len(output["unclassified"]) == 2
@@ -85,8 +85,8 @@ def test_no_extension(self, preview_samples_path):
             preview_samples_path / "txt" / "doc_2",
             preview_samples_path / "txt" / "doc_2.txt",
         ]
-        classifier = FileExtensionClassifier(mime_types=["text/plain"])
-        output = classifier.run(paths=file_paths)
+        router = FileExtensionRouter(mime_types=["text/plain"])
+        output = router.run(paths=file_paths)
         assert len(output["text/plain"]) == 2
         assert len(output["unclassified"]) == 1
 
@@ -96,4 +96,4 @@ def test_unknown_mime_type(self):
         Test that the component handles files with unknown mime types.
         """
         with pytest.raises(ValueError, match="Unknown mime type:"):
-            FileExtensionClassifier(mime_types=["type_invalid"])
+            FileExtensionRouter(mime_types=["type_invalid"])
diff --git a/test/preview/components/routers/test_metadata_router.py b/test/preview/components/routers/test_metadata_router.py
new file mode 100644
index 0000000000..3108f7cf10
--- /dev/null
+++ b/test/preview/components/routers/test_metadata_router.py
@@ -0,0 +1,41 @@
+import pytest
+
+from haystack.preview import Document
+from haystack.preview.components.routers.metadata_router import MetadataRouter
+
+
+class TestMetadataRouter:
+    @pytest.mark.unit
+    def test_to_dict(self):
+        component = MetadataRouter(rules={"edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}}})
+        data = component.to_dict()
+        assert data == {
+            "type": "MetadataRouter",
+            "init_parameters": {"rules": {"edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}}}},
+        }
+
+    @pytest.mark.unit
+    def test_from_dict(self):
+        data = {
+            "type": "MetadataRouter",
+            "init_parameters": {"rules": {"edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}}}},
+        }
+        component = MetadataRouter.from_dict(data)
+        assert component.rules == {"edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}}}
+
+    @pytest.mark.unit
+    def test_run(self):
+        rules = {
+            "edge_1": {"created_at": {"$gte": "2023-01-01", "$lt": "2023-04-01"}},
+            "edge_2": {"created_at": {"$gte": "2023-04-01", "$lt": "2023-07-01"}},
+        }
+        router = MetadataRouter(rules=rules)
+        documents = [
+            Document(metadata={"created_at": "2023-02-01"}),
+            Document(metadata={"created_at": "2023-05-01"}),
+            Document(metadata={"created_at": "2023-08-01"}),
+        ]
+        output = router.run(documents=documents)
+        assert output["edge_1"][0].metadata["created_at"] == "2023-02-01"
+        assert output["edge_2"][0].metadata["created_at"] == "2023-05-01"
+        assert output["unmatched"][0].metadata["created_at"] == "2023-08-01"


Code review comments:
diff_hunk: @@ -0,0 +1,4 @@
+---
+preview:
+  - |
+    Add `MetadataRouter`, a component that routes documents to different edges based on their metadata and specified rules.
path: releasenotes/notes/add-metadata_router-2.0-63829ac5a0528e9d.yaml
text: @user1:
Matching the docstring
```suggestion
    Add `MetadataRouter`, a component that routes documents to different edges based on the content of their fields
```

diff_hunk: @@ -0,0 +1,67 @@
+from typing import Any, Dict, List
+
+from haystack.preview import component, default_from_dict, default_to_dict, Document
+from haystack.preview.utils.filters import document_matches_filter
+
+
+@component
+class MetadataRouter:
+    """
+    A component that routes documents to different edges based on their metadata and specified rules.
path: haystack/preview/components/routers/metadata_router.py
text: @user1:
Let's always call them connections rather than edges. Also, maybe we can be a bit more generic about what the class can do

```suggestion
    A component that routes documents to different connections based on the content of their fields.
```