We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Weights and score normalization for DocumentJoiner with reciprocal rank fusion - 2.x
Complete details in #5551.
Implemented for 1.x by @robpasternak in #5704.

We should port this improvement to 2.x.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `haystack/components/joiners/document_joiner.py`
Content:
```
1 import itertools
2 import logging
3 from collections import defaultdict
4 from math import inf
5 from typing import List, Optional
6 from haystack.core.component.types import Variadic
7 
8 from haystack import component, Document
9 
10 
11 logger = logging.getLogger(__name__)
12 
13 
14 @component
15 class DocumentJoiner:
16     """
17     A component that joins input lists of Documents from multiple connections and outputs them as one list.
18 
19     The component allows multiple join modes:
20     * concatenate: Combine Documents from multiple components. Discards duplicate Documents.
21                     Documents get their scores from the last component in the pipeline that assigns scores.
22                     This join mode doesn't influence Document scores.
23     * merge: Merge scores of duplicate Documents coming from multiple components.
24             Optionally, you can assign a weight to the scores and set the top_k limit for this join mode.
25             You can also use this join mode to rerank retrieved Documents.
26     * reciprocal_rank_fusion: Combine Documents into a single list based on their ranking received from multiple components.
27 
28     Example usage in a hybrid retrieval pipeline:
29     ```python
30     document_store = InMemoryDocumentStore()
31     p = Pipeline()
32     p.add_component(instance=InMemoryBM25Retriever(document_store=document_store), name="bm25_retriever")
33     p.add_component(
34             instance=SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"),
35             name="text_embedder",
36         )
37     p.add_component(instance=InMemoryEmbeddingRetriever(document_store=document_store), name="embedding_retriever")
38     p.add_component(instance=DocumentJoiner(), name="joiner")
39     p.connect("bm25_retriever", "joiner")
40     p.connect("embedding_retriever", "joiner")
41     p.connect("text_embedder", "embedding_retriever")
42     query = "What is the capital of France?"
43     p.run(data={"bm25_retriever": {"query": query},
44                 "text_embedder": {"text": query}})
45     ```
46     """
47 
48     def __init__(
49         self,
50         join_mode: str = "concatenate",
51         weights: Optional[List[float]] = None,
52         top_k: Optional[int] = None,
53         sort_by_score: bool = True,
54     ):
55         """
56         Initialize the DocumentJoiner.
57 
58         :param join_mode: Specifies the join mode to use. Available modes: `concatenate` to combine Documents from multiple Retrievers, `merge` to aggregate the scores of
59                           individual Documents, `reciprocal_rank_fusion` to apply rank-based scoring.
60         :param weights: A component-wise list (the length of the list must be equal to the number of input components) of weights for
61                         adjusting Document scores when using the `merge` join_mode. By default, equal weight is given
62                         to each Retriever score. This param is not compatible with the `concatenate` join_mode.
63         :param top_k: The maximum number of Documents to be returned as output. By default, returns all Documents.
64         :param sort_by_score: Whether the output list of Documents should be sorted by Document scores in descending order.
65                               By default, the output is sorted.
66                               Documents without score are handled as if their score was -infinity.
67         """
68         if join_mode not in ["concatenate", "merge", "reciprocal_rank_fusion"]:
69             raise ValueError(f"DocumentJoiner component does not support '{join_mode}' join_mode.")
70         self.join_mode = join_mode
71         self.weights = [float(i) / sum(weights) for i in weights] if weights else None
72         self.top_k = top_k
73         self.sort_by_score = sort_by_score
74 
75     @component.output_types(documents=List[Document])
76     def run(self, documents: Variadic[List[Document]]):
77         """
78         Run the DocumentJoiner. This method joins the input lists of Documents into one output list based on the join_mode specified during initialization.
79 
80         :param documents: An arbitrary number of lists of Documents to join.
81         """
82         output_documents = []
83         if self.join_mode == "concatenate":
84             output_documents = self._concatenate(documents)
85         elif self.join_mode == "merge":
86             output_documents = self._merge(documents)
87         elif self.join_mode == "reciprocal_rank_fusion":
88             output_documents = self._reciprocal_rank_fusion(documents)
89 
90         if self.sort_by_score:
91             output_documents = sorted(
92                 output_documents, key=lambda doc: doc.score if doc.score is not None else -inf, reverse=True
93             )
94             if any(doc.score is None for doc in output_documents):
95                 logger.info(
96                     "Some of the Documents DocumentJoiner got have score=None. It was configured to sort Documents by "
97                     "score, so those with score=None were sorted as if they had a score of -infinity."
98                 )
99 
100         if self.top_k:
101             output_documents = output_documents[: self.top_k]
102         return {"documents": output_documents}
103 
104     def _concatenate(self, document_lists):
105         """
106         Concatenate multiple lists of Documents and return only the Document with the highest score for duplicate Documents.
107         """
108         output = []
109         docs_per_id = defaultdict(list)
110         for doc in itertools.chain.from_iterable(document_lists):
111             docs_per_id[doc.id].append(doc)
112         for docs in docs_per_id.values():
113             doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score else -inf)
114             output.append(doc_with_best_score)
115         return output
116 
117     def _merge(self, document_lists):
118         """
119         Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.
120         """
121         scores_map = defaultdict(int)
122         documents_map = {}
123         weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)
124 
125         for documents, weight in zip(document_lists, weights):
126             for doc in documents:
127                 scores_map[doc.id] += (doc.score if doc.score else 0) * weight
128                 documents_map[doc.id] = doc
129 
130         for doc in documents_map.values():
131             doc.score = scores_map[doc.id]
132 
133         return documents_map.values()
134 
135     def _reciprocal_rank_fusion(self, document_lists):
136         """
137         Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.
138         The constant k is set to 61 (60 was suggested by the original paper,
139         plus 1 as python lists are 0-based and the paper used 1-based ranking).
140         """
141         k = 61
142 
143         scores_map = defaultdict(int)
144         documents_map = {}
145         for documents in document_lists:
146             for rank, doc in enumerate(documents):
147                 scores_map[doc.id] += 1 / (k + rank)
148                 documents_map[doc.id] = doc
149 
150         for doc in documents_map.values():
151             doc.score = scores_map[doc.id]
152 
153         return documents_map.values()
154
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/haystack/components/joiners/document_joiner.py b/haystack/components/joiners/document_joiner.py
--- a/haystack/components/joiners/document_joiner.py
+++ b/haystack/components/joiners/document_joiner.py
@@ -142,11 +142,19 @@
 
         scores_map = defaultdict(int)
         documents_map = {}
-        for documents in document_lists:
+        weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)
+
+        # Calculate weighted reciprocal rank fusion score
+        for documents, weight in zip(document_lists, weights):
             for rank, doc in enumerate(documents):
-                scores_map[doc.id] += 1 / (k + rank)
+                scores_map[doc.id] += (weight * len(document_lists)) / (k + rank)
                 documents_map[doc.id] = doc
 
+        # Normalize scores. Note: len(results) / k is the maximum possible score,
+        # achieved by being ranked first in all doc lists with non-zero weight.
+        for id in scores_map:
+            scores_map[id] /= len(document_lists) / k
+
         for doc in documents_map.values():
             doc.score = scores_map[doc.id]

{"golden_diff": "diff --git a/haystack/components/joiners/document_joiner.py b/haystack/components/joiners/document_joiner.py\n--- a/haystack/components/joiners/document_joiner.py\n+++ b/haystack/components/joiners/document_joiner.py\n@@ -142,11 +142,19 @@\n \n         scores_map = defaultdict(int)\n         documents_map = {}\n-        for documents in document_lists:\n+        weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)\n+\n+        # Calculate weighted reciprocal rank fusion score\n+        for documents, weight in zip(document_lists, weights):\n             for rank, doc in enumerate(documents):\n-                scores_map[doc.id] += 1 / (k + rank)\n+                scores_map[doc.id] += (weight * len(document_lists)) / (k + rank)\n                 documents_map[doc.id] = doc\n \n+        # Normalize scores. Note: len(results) / k is the maximum possible score,\n+        # achieved by being ranked first in all doc lists with non-zero weight.\n+        for id in scores_map:\n+            scores_map[id] /= len(document_lists) / k\n+\n         for doc in documents_map.values():\n             doc.score = scores_map[doc.id]\n", "issue": "Weights and score normalization for DocumentJoiner with reciprocal rank fusion - 2.x\nComplete details in #5551.\r\nImplemented for 1.x by @robpasternak in #5704.\r\n\r\nWe should port this improvement to 2.x.\n", "before_files": [{"content": "import itertools\nimport logging\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\nfrom haystack.core.component.types import Variadic\n\nfrom haystack import component, Document\n\n\nlogger = logging.getLogger(__name__)\n\n\n@component\nclass DocumentJoiner:\n    \"\"\"\n    A component that joins input lists of Documents from multiple connections and outputs them as one list.\n\n    The component allows multiple join modes:\n    * concatenate: Combine Documents from multiple components. Discards duplicate Documents.\n                    Documents get their scores from the last component in the pipeline that assigns scores.\n                    This join mode doesn't influence Document scores.\n    * merge: Merge scores of duplicate Documents coming from multiple components.\n            Optionally, you can assign a weight to the scores and set the top_k limit for this join mode.\n            You can also use this join mode to rerank retrieved Documents.\n    * reciprocal_rank_fusion: Combine Documents into a single list based on their ranking received from multiple components.\n\n    Example usage in a hybrid retrieval pipeline:\n    ```python\n    document_store = InMemoryDocumentStore()\n    p = Pipeline()\n    p.add_component(instance=InMemoryBM25Retriever(document_store=document_store), name=\"bm25_retriever\")\n    p.add_component(\n            instance=SentenceTransformersTextEmbedder(model=\"sentence-transformers/all-MiniLM-L6-v2\"),\n            name=\"text_embedder\",\n        )\n    p.add_component(instance=InMemoryEmbeddingRetriever(document_store=document_store), name=\"embedding_retriever\")\n    p.add_component(instance=DocumentJoiner(), name=\"joiner\")\n    p.connect(\"bm25_retriever\", \"joiner\")\n    p.connect(\"embedding_retriever\", \"joiner\")\n    p.connect(\"text_embedder\", \"embedding_retriever\")\n    query = \"What is the capital of France?\"\n    p.run(data={\"bm25_retriever\": {\"query\": query},\n                \"text_embedder\": {\"text\": query}})\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        join_mode: str = \"concatenate\",\n        weights: Optional[List[float]] = None,\n        top_k: Optional[int] = None,\n        sort_by_score: bool = True,\n    ):\n        \"\"\"\n        Initialize the DocumentJoiner.\n\n        :param join_mode: Specifies the join mode to use. Available modes: `concatenate` to combine Documents from multiple Retrievers, `merge` to aggregate the scores of\n                          individual Documents, `reciprocal_rank_fusion` to apply rank-based scoring.\n        :param weights: A component-wise list (the length of the list must be equal to the number of input components) of weights for\n                        adjusting Document scores when using the `merge` join_mode. By default, equal weight is given\n                        to each Retriever score. This param is not compatible with the `concatenate` join_mode.\n        :param top_k: The maximum number of Documents to be returned as output. By default, returns all Documents.\n        :param sort_by_score: Whether the output list of Documents should be sorted by Document scores in descending order.\n                              By default, the output is sorted.\n                              Documents without score are handled as if their score was -infinity.\n        \"\"\"\n        if join_mode not in [\"concatenate\", \"merge\", \"reciprocal_rank_fusion\"]:\n            raise ValueError(f\"DocumentJoiner component does not support '{join_mode}' join_mode.\")\n        self.join_mode = join_mode\n        self.weights = [float(i) / sum(weights) for i in weights] if weights else None\n        self.top_k = top_k\n        self.sort_by_score = sort_by_score\n\n    @component.output_types(documents=List[Document])\n    def run(self, documents: Variadic[List[Document]]):\n        \"\"\"\n        Run the DocumentJoiner. This method joins the input lists of Documents into one output list based on the join_mode specified during initialization.\n\n        :param documents: An arbitrary number of lists of Documents to join.\n        \"\"\"\n        output_documents = []\n        if self.join_mode == \"concatenate\":\n            output_documents = self._concatenate(documents)\n        elif self.join_mode == \"merge\":\n            output_documents = self._merge(documents)\n        elif self.join_mode == \"reciprocal_rank_fusion\":\n            output_documents = self._reciprocal_rank_fusion(documents)\n\n        if self.sort_by_score:\n            output_documents = sorted(\n                output_documents, key=lambda doc: doc.score if doc.score is not None else -inf, reverse=True\n            )\n            if any(doc.score is None for doc in output_documents):\n                logger.info(\n                    \"Some of the Documents DocumentJoiner got have score=None. It was configured to sort Documents by \"\n                    \"score, so those with score=None were sorted as if they had a score of -infinity.\"\n                )\n\n        if self.top_k:\n            output_documents = output_documents[: self.top_k]\n        return {\"documents\": output_documents}\n\n    def _concatenate(self, document_lists):\n        \"\"\"\n        Concatenate multiple lists of Documents and return only the Document with the highest score for duplicate Documents.\n        \"\"\"\n        output = []\n        docs_per_id = defaultdict(list)\n        for doc in itertools.chain.from_iterable(document_lists):\n            docs_per_id[doc.id].append(doc)\n        for docs in docs_per_id.values():\n            doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score else -inf)\n            output.append(doc_with_best_score)\n        return output\n\n    def _merge(self, document_lists):\n        \"\"\"\n        Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.\n        \"\"\"\n        scores_map = defaultdict(int)\n        documents_map = {}\n        weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)\n\n        for documents, weight in zip(document_lists, weights):\n            for doc in documents:\n                scores_map[doc.id] += (doc.score if doc.score else 0) * weight\n                documents_map[doc.id] = doc\n\n        for doc in documents_map.values():\n            doc.score = scores_map[doc.id]\n\n        return documents_map.values()\n\n    def _reciprocal_rank_fusion(self, document_lists):\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.\n        The constant k is set to 61 (60 was suggested by the original paper,\n        plus 1 as python lists are 0-based and the paper used 1-based ranking).\n        \"\"\"\n        k = 61\n\n        scores_map = defaultdict(int)\n        documents_map = {}\n        for documents in document_lists:\n            for rank, doc in enumerate(documents):\n                scores_map[doc.id] += 1 / (k + rank)\n                documents_map[doc.id] = doc\n\n        for doc in documents_map.values():\n            doc.score = scores_map[doc.id]\n\n        return documents_map.values()\n", "path": "haystack/components/joiners/document_joiner.py"}], "after_files": [{"content": "import itertools\nimport logging\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\nfrom haystack.core.component.types import Variadic\n\nfrom haystack import component, Document\n\n\nlogger = logging.getLogger(__name__)\n\n\n@component\nclass DocumentJoiner:\n    \"\"\"\n    A component that joins input lists of Documents from multiple connections and outputs them as one list.\n\n    The component allows multiple join modes:\n    * concatenate: Combine Documents from multiple components. Discards duplicate Documents.\n                    Documents get their scores from the last component in the pipeline that assigns scores.\n                    This join mode doesn't influence Document scores.\n    * merge: Merge scores of duplicate Documents coming from multiple components.\n            Optionally, you can assign a weight to the scores and set the top_k limit for this join mode.\n            You can also use this join mode to rerank retrieved Documents.\n    * reciprocal_rank_fusion: Combine Documents into a single list based on their ranking received from multiple components.\n\n    Example usage in a hybrid retrieval pipeline:\n    ```python\n    document_store = InMemoryDocumentStore()\n    p = Pipeline()\n    p.add_component(instance=InMemoryBM25Retriever(document_store=document_store), name=\"bm25_retriever\")\n    p.add_component(\n            instance=SentenceTransformersTextEmbedder(model=\"sentence-transformers/all-MiniLM-L6-v2\"),\n            name=\"text_embedder\",\n        )\n    p.add_component(instance=InMemoryEmbeddingRetriever(document_store=document_store), name=\"embedding_retriever\")\n    p.add_component(instance=DocumentJoiner(), name=\"joiner\")\n    p.connect(\"bm25_retriever\", \"joiner\")\n    p.connect(\"embedding_retriever\", \"joiner\")\n    p.connect(\"text_embedder\", \"embedding_retriever\")\n    query = \"What is the capital of France?\"\n    p.run(data={\"bm25_retriever\": {\"query\": query},\n                \"text_embedder\": {\"text\": query}})\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        join_mode: str = \"concatenate\",\n        weights: Optional[List[float]] = None,\n        top_k: Optional[int] = None,\n        sort_by_score: bool = True,\n    ):\n        \"\"\"\n        Initialize the DocumentJoiner.\n\n        :param join_mode: Specifies the join mode to use. Available modes: `concatenate` to combine Documents from multiple Retrievers, `merge` to aggregate the scores of\n                          individual Documents, `reciprocal_rank_fusion` to apply rank-based scoring.\n        :param weights: A component-wise list (the length of the list must be equal to the number of input components) of weights for\n                        adjusting Document scores when using the `merge` join_mode. By default, equal weight is given\n                        to each Retriever score. This param is not compatible with the `concatenate` join_mode.\n        :param top_k: The maximum number of Documents to be returned as output. By default, returns all Documents.\n        :param sort_by_score: Whether the output list of Documents should be sorted by Document scores in descending order.\n                              By default, the output is sorted.\n                              Documents without score are handled as if their score was -infinity.\n        \"\"\"\n        if join_mode not in [\"concatenate\", \"merge\", \"reciprocal_rank_fusion\"]:\n            raise ValueError(f\"DocumentJoiner component does not support '{join_mode}' join_mode.\")\n        self.join_mode = join_mode\n        self.weights = [float(i) / sum(weights) for i in weights] if weights else None\n        self.top_k = top_k\n        self.sort_by_score = sort_by_score\n\n    @component.output_types(documents=List[Document])\n    def run(self, documents: Variadic[List[Document]]):\n        \"\"\"\n        Run the DocumentJoiner. This method joins the input lists of Documents into one output list based on the join_mode specified during initialization.\n\n        :param documents: An arbitrary number of lists of Documents to join.\n        \"\"\"\n        output_documents = []\n        if self.join_mode == \"concatenate\":\n            output_documents = self._concatenate(documents)\n        elif self.join_mode == \"merge\":\n            output_documents = self._merge(documents)\n        elif self.join_mode == \"reciprocal_rank_fusion\":\n            output_documents = self._reciprocal_rank_fusion(documents)\n\n        if self.sort_by_score:\n            output_documents = sorted(\n                output_documents, key=lambda doc: doc.score if doc.score is not None else -inf, reverse=True\n            )\n            if any(doc.score is None for doc in output_documents):\n                logger.info(\n                    \"Some of the Documents DocumentJoiner got have score=None. It was configured to sort Documents by \"\n                    \"score, so those with score=None were sorted as if they had a score of -infinity.\"\n                )\n\n        if self.top_k:\n            output_documents = output_documents[: self.top_k]\n        return {\"documents\": output_documents}\n\n    def _concatenate(self, document_lists):\n        \"\"\"\n        Concatenate multiple lists of Documents and return only the Document with the highest score for duplicate Documents.\n        \"\"\"\n        output = []\n        docs_per_id = defaultdict(list)\n        for doc in itertools.chain.from_iterable(document_lists):\n            docs_per_id[doc.id].append(doc)\n        for docs in docs_per_id.values():\n            doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score else -inf)\n            output.append(doc_with_best_score)\n        return output\n\n    def _merge(self, document_lists):\n        \"\"\"\n        Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.\n        \"\"\"\n        scores_map = defaultdict(int)\n        documents_map = {}\n        weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)\n\n        for documents, weight in zip(document_lists, weights):\n            for doc in documents:\n                scores_map[doc.id] += (doc.score if doc.score else 0) * weight\n                documents_map[doc.id] = doc\n\n        for doc in documents_map.values():\n            doc.score = scores_map[doc.id]\n\n        return documents_map.values()\n\n    def _reciprocal_rank_fusion(self, document_lists):\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.\n        The constant k is set to 61 (60 was suggested by the original paper,\n        plus 1 as python lists are 0-based and the paper used 1-based ranking).\n        \"\"\"\n        k = 61\n\n        scores_map = defaultdict(int)\n        documents_map = {}\n        weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)\n\n        # Calculate weighted reciprocal rank fusion score\n        for documents, weight in zip(document_lists, weights):\n            for rank, doc in enumerate(documents):\n                scores_map[doc.id] += (weight * len(document_lists)) / (k + rank)\n                documents_map[doc.id] = doc\n\n        # Normalize scores. Note: len(results) / k is the maximum possible score,\n        # achieved by being ranked first in all doc lists with non-zero weight.\n        for id in scores_map:\n            scores_map[id] /= len(document_lists) / k\n\n        for doc in documents_map.values():\n            doc.score = scores_map[doc.id]\n\n        return documents_map.values()\n", "path": "haystack/components/joiners/document_joiner.py"}]}