We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add Distribution-based rank fusion in JoinDocuments
**Is your feature request related to a problem? Please describe.**
Add [Distribution-based rank fusion](https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18) in JoinDocuments

**Describe the solution you'd like**
```
def _distribution_based_rank_fusion(self, document_lists):
        """
        Merge multiple lists of Documents and assign scores based on Distribution-Based Score Fusion.
        (https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18)

        If a Document is in more than one retriever, the sone with the highest score is used.
        """
        for documents in document_lists:
            scores_list = []

            for doc in documents:
                scores_list.append(doc.score)

            mean_score = sum(scores_list) / len(scores_list)
            std_dev = (
                sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)
            ) ** 0.5
            min_score = mean_score - 3 * std_dev
            max_score = mean_score + 3 * std_dev

            for doc in documents:
                doc.score = (doc.score - min_score) / (max_score - min_score)

        output = self._concatenate(document_lists=document_lists)

        return output
```

--- 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 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2 #
3 # SPDX-License-Identifier: Apache-2.0
4 
5 import itertools
6 from collections import defaultdict
7 from math import inf
8 from typing import List, Optional
9 
10 from haystack import Document, component, logging
11 from haystack.core.component.types import Variadic
12 
13 logger = logging.getLogger(__name__)
14 
15 
16 @component
17 class DocumentJoiner:
18     """
19     A component that joins multiple list of Documents into a single list.
20 
21     It supports different joins modes:
22     - concatenate: Keeps the highest scored Document in case of duplicates.
23     - merge: Merge a calculate a weighted sum of the scores of duplicate Documents.
24     - reciprocal_rank_fusion: Merge and assign scores based on reciprocal rank fusion.
25     - distribution_based_rank_fusion: Merge and assign scores based on scores distribution in each retriever
26 
27     Usage example:
28     ```python
29     document_store = InMemoryDocumentStore()
30     p = Pipeline()
31     p.add_component(instance=InMemoryBM25Retriever(document_store=document_store), name="bm25_retriever")
32     p.add_component(
33             instance=SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"),
34             name="text_embedder",
35         )
36     p.add_component(instance=InMemoryEmbeddingRetriever(document_store=document_store), name="embedding_retriever")
37     p.add_component(instance=DocumentJoiner(), name="joiner")
38     p.connect("bm25_retriever", "joiner")
39     p.connect("embedding_retriever", "joiner")
40     p.connect("text_embedder", "embedding_retriever")
41     query = "What is the capital of France?"
42     p.run(data={"query": query})
43     ```
44     """
45 
46     def __init__(
47         self,
48         join_mode: str = "concatenate",
49         weights: Optional[List[float]] = None,
50         top_k: Optional[int] = None,
51         sort_by_score: bool = True,
52     ):
53         """
54         Create an DocumentJoiner component.
55 
56         :param join_mode:
57             Specifies the join mode to use. Available modes:
58             - `concatenate`
59             - `merge`
60             - `reciprocal_rank_fusion`
61             - `distribution_based_rank_fusion`
62         :param weights:
63             Weight for each list of Documents received, must have the same length as the number of inputs.
64             If `join_mode` is `concatenate` or `distribution_based_rank_fusion` this parameter is ignored.
65         :param top_k:
66             The maximum number of Documents to return.
67         :param sort_by_score:
68             If True sorts the Documents by score in descending order.
69             If a Document has no score, it is handled as if its score is -infinity.
70         """
71         if join_mode not in ["concatenate", "merge", "reciprocal_rank_fusion", "distribution_based_rank_fusion"]:
72             raise ValueError(f"DocumentJoiner component does not support '{join_mode}' join_mode.")
73         self.join_mode = join_mode
74         self.weights = [float(i) / sum(weights) for i in weights] if weights else None
75         self.top_k = top_k
76         self.sort_by_score = sort_by_score
77 
78     @component.output_types(documents=List[Document])
79     def run(self, documents: Variadic[List[Document]], top_k: Optional[int] = None):
80         """
81         Joins multiple lists of Documents into a single list depending on the `join_mode` parameter.
82 
83         :param documents:
84             List of list of Documents to be merged.
85         :param top_k:
86             The maximum number of Documents to return. Overrides the instance's `top_k` if provided.
87 
88         :returns:
89             A dictionary with the following keys:
90             - `documents`: Merged list of Documents
91         """
92         output_documents = []
93 
94         documents = list(documents)
95         if self.join_mode == "concatenate":
96             output_documents = self._concatenate(documents)
97         elif self.join_mode == "merge":
98             output_documents = self._merge(documents)
99         elif self.join_mode == "reciprocal_rank_fusion":
100             output_documents = self._reciprocal_rank_fusion(documents)
101         elif self.join_mode == "distribution_based_rank_fusion":
102             output_documents = self._distribution_based_rank_fusion(documents)
103 
104         if self.sort_by_score:
105             output_documents = sorted(
106                 output_documents, key=lambda doc: doc.score if doc.score is not None else -inf, reverse=True
107             )
108             if any(doc.score is None for doc in output_documents):
109                 logger.info(
110                     "Some of the Documents DocumentJoiner got have score=None. It was configured to sort Documents by "
111                     "score, so those with score=None were sorted as if they had a score of -infinity."
112                 )
113 
114         if top_k:
115             output_documents = output_documents[:top_k]
116         elif self.top_k:
117             output_documents = output_documents[: self.top_k]
118 
119         return {"documents": output_documents}
120 
121     def _concatenate(self, document_lists: List[List[Document]]) -> List[Document]:
122         """
123         Concatenate multiple lists of Documents and return only the Document with the highest score for duplicates.
124         """
125         output = []
126         docs_per_id = defaultdict(list)
127         for doc in itertools.chain.from_iterable(document_lists):
128             docs_per_id[doc.id].append(doc)
129         for docs in docs_per_id.values():
130             doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score else -inf)
131             output.append(doc_with_best_score)
132         return output
133 
134     def _merge(self, document_lists: List[List[Document]]) -> List[Document]:
135         """
136         Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.
137         """
138         scores_map: dict = defaultdict(int)
139         documents_map = {}
140         weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)
141 
142         for documents, weight in zip(document_lists, weights):
143             for doc in documents:
144                 scores_map[doc.id] += (doc.score if doc.score else 0) * weight
145                 documents_map[doc.id] = doc
146 
147         for doc in documents_map.values():
148             doc.score = scores_map[doc.id]
149 
150         return list(documents_map.values())
151 
152     def _reciprocal_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:
153         """
154         Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.
155 
156         The constant k is set to 61 (60 was suggested by the original paper,
157         plus 1 as python lists are 0-based and the paper used 1-based ranking).
158         """
159         k = 61
160 
161         scores_map: dict = defaultdict(int)
162         documents_map = {}
163         weights = self.weights if self.weights else [1 / len(document_lists)] * len(document_lists)
164 
165         # Calculate weighted reciprocal rank fusion score
166         for documents, weight in zip(document_lists, weights):
167             for rank, doc in enumerate(documents):
168                 scores_map[doc.id] += (weight * len(document_lists)) / (k + rank)
169                 documents_map[doc.id] = doc
170 
171         # Normalize scores. Note: len(results) / k is the maximum possible score,
172         # achieved by being ranked first in all doc lists with non-zero weight.
173         for _id in scores_map:
174             scores_map[_id] /= len(document_lists) / k
175 
176         for doc in documents_map.values():
177             doc.score = scores_map[doc.id]
178 
179         return list(documents_map.values())
180 
181     def _distribution_based_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:
182         """
183         Merge multiple lists of Documents and assign scores based on Distribution-Based Score Fusion.
184 
185         (https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18)
186         If a Document is in more than one retriever, the one with the highest score is used.
187         """
188         for documents in document_lists:
189             scores_list = []
190 
191             for doc in documents:
192                 scores_list.append(doc.score if doc.score is not None else 0)
193 
194             mean_score = sum(scores_list) / len(scores_list)
195             std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5
196             min_score = mean_score - 3 * std_dev
197             max_score = mean_score + 3 * std_dev
198 
199             for doc in documents:
200                 doc.score = (doc.score - min_score) / (max_score - min_score)
201 
202         output = self._concatenate(document_lists=document_lists)
203 
204         return output
205
```
--- 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
@@ -195,9 +195,11 @@
             std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5
             min_score = mean_score - 3 * std_dev
             max_score = mean_score + 3 * std_dev
+            delta_score = max_score - min_score
 
             for doc in documents:
-                doc.score = (doc.score - min_score) / (max_score - min_score)
+                doc.score = (doc.score - min_score) / delta_score if delta_score != 0.0 else 0.0
+                # if all docs have the same score delta_score is 0, the docs are uninformative for the query
 
         output = self._concatenate(document_lists=document_lists)

{"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@@ -195,9 +195,11 @@\n             std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5\n             min_score = mean_score - 3 * std_dev\n             max_score = mean_score + 3 * std_dev\n+            delta_score = max_score - min_score\n \n             for doc in documents:\n-                doc.score = (doc.score - min_score) / (max_score - min_score)\n+                doc.score = (doc.score - min_score) / delta_score if delta_score != 0.0 else 0.0\n+                # if all docs have the same score delta_score is 0, the docs are uninformative for the query\n \n         output = self._concatenate(document_lists=document_lists)\n", "issue": "Add Distribution-based rank fusion in JoinDocuments\n**Is your feature request related to a problem? Please describe.**\r\nAdd [Distribution-based rank fusion](https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18) in JoinDocuments\r\n\r\n**Describe the solution you'd like**\r\n```\r\ndef _distribution_based_rank_fusion(self, document_lists):\r\n        \"\"\"\r\n        Merge multiple lists of Documents and assign scores based on Distribution-Based Score Fusion.\r\n        (https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18)\r\n\r\n        If a Document is in more than one retriever, the sone with the highest score is used.\r\n        \"\"\"\r\n        for documents in document_lists:\r\n            scores_list = []\r\n\r\n            for doc in documents:\r\n                scores_list.append(doc.score)\r\n\r\n            mean_score = sum(scores_list) / len(scores_list)\r\n            std_dev = (\r\n                sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)\r\n            ) ** 0.5\r\n            min_score = mean_score - 3 * std_dev\r\n            max_score = mean_score + 3 * std_dev\r\n\r\n            for doc in documents:\r\n                doc.score = (doc.score - min_score) / (max_score - min_score)\r\n\r\n        output = self._concatenate(document_lists=document_lists)\r\n\r\n        return output\r\n```\n", "before_files": [{"content": "# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>\n#\n# SPDX-License-Identifier: Apache-2.0\n\nimport itertools\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\n\nfrom haystack import Document, component, logging\nfrom haystack.core.component.types import Variadic\n\nlogger = logging.getLogger(__name__)\n\n\n@component\nclass DocumentJoiner:\n    \"\"\"\n    A component that joins multiple list of Documents into a single list.\n\n    It supports different joins modes:\n    - concatenate: Keeps the highest scored Document in case of duplicates.\n    - merge: Merge a calculate a weighted sum of the scores of duplicate Documents.\n    - reciprocal_rank_fusion: Merge and assign scores based on reciprocal rank fusion.\n    - distribution_based_rank_fusion: Merge and assign scores based on scores distribution in each retriever\n\n    Usage example:\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={\"query\": 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        Create an DocumentJoiner component.\n\n        :param join_mode:\n            Specifies the join mode to use. Available modes:\n            - `concatenate`\n            - `merge`\n            - `reciprocal_rank_fusion`\n            - `distribution_based_rank_fusion`\n        :param weights:\n            Weight for each list of Documents received, must have the same length as the number of inputs.\n            If `join_mode` is `concatenate` or `distribution_based_rank_fusion` this parameter is ignored.\n        :param top_k:\n            The maximum number of Documents to return.\n        :param sort_by_score:\n            If True sorts the Documents by score in descending order.\n            If a Document has no score, it is handled as if its score is -infinity.\n        \"\"\"\n        if join_mode not in [\"concatenate\", \"merge\", \"reciprocal_rank_fusion\", \"distribution_based_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]], top_k: Optional[int] = None):\n        \"\"\"\n        Joins multiple lists of Documents into a single list depending on the `join_mode` parameter.\n\n        :param documents:\n            List of list of Documents to be merged.\n        :param top_k:\n            The maximum number of Documents to return. Overrides the instance's `top_k` if provided.\n\n        :returns:\n            A dictionary with the following keys:\n            - `documents`: Merged list of Documents\n        \"\"\"\n        output_documents = []\n\n        documents = list(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        elif self.join_mode == \"distribution_based_rank_fusion\":\n            output_documents = self._distribution_based_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 top_k:\n            output_documents = output_documents[:top_k]\n        elif self.top_k:\n            output_documents = output_documents[: self.top_k]\n\n        return {\"documents\": output_documents}\n\n    def _concatenate(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Concatenate multiple lists of Documents and return only the Document with the highest score for duplicates.\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: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.\n        \"\"\"\n        scores_map: dict = 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 list(documents_map.values())\n\n    def _reciprocal_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.\n\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: dict = 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 list(documents_map.values())\n\n    def _distribution_based_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on Distribution-Based Score Fusion.\n\n        (https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18)\n        If a Document is in more than one retriever, the one with the highest score is used.\n        \"\"\"\n        for documents in document_lists:\n            scores_list = []\n\n            for doc in documents:\n                scores_list.append(doc.score if doc.score is not None else 0)\n\n            mean_score = sum(scores_list) / len(scores_list)\n            std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5\n            min_score = mean_score - 3 * std_dev\n            max_score = mean_score + 3 * std_dev\n\n            for doc in documents:\n                doc.score = (doc.score - min_score) / (max_score - min_score)\n\n        output = self._concatenate(document_lists=document_lists)\n\n        return output\n", "path": "haystack/components/joiners/document_joiner.py"}], "after_files": [{"content": "# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>\n#\n# SPDX-License-Identifier: Apache-2.0\n\nimport itertools\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\n\nfrom haystack import Document, component, logging\nfrom haystack.core.component.types import Variadic\n\nlogger = logging.getLogger(__name__)\n\n\n@component\nclass DocumentJoiner:\n    \"\"\"\n    A component that joins multiple list of Documents into a single list.\n\n    It supports different joins modes:\n    - concatenate: Keeps the highest scored Document in case of duplicates.\n    - merge: Merge a calculate a weighted sum of the scores of duplicate Documents.\n    - reciprocal_rank_fusion: Merge and assign scores based on reciprocal rank fusion.\n    - distribution_based_rank_fusion: Merge and assign scores based on scores distribution in each retriever\n\n    Usage example:\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={\"query\": 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        Create an DocumentJoiner component.\n\n        :param join_mode:\n            Specifies the join mode to use. Available modes:\n            - `concatenate`\n            - `merge`\n            - `reciprocal_rank_fusion`\n            - `distribution_based_rank_fusion`\n        :param weights:\n            Weight for each list of Documents received, must have the same length as the number of inputs.\n            If `join_mode` is `concatenate` or `distribution_based_rank_fusion` this parameter is ignored.\n        :param top_k:\n            The maximum number of Documents to return.\n        :param sort_by_score:\n            If True sorts the Documents by score in descending order.\n            If a Document has no score, it is handled as if its score is -infinity.\n        \"\"\"\n        if join_mode not in [\"concatenate\", \"merge\", \"reciprocal_rank_fusion\", \"distribution_based_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]], top_k: Optional[int] = None):\n        \"\"\"\n        Joins multiple lists of Documents into a single list depending on the `join_mode` parameter.\n\n        :param documents:\n            List of list of Documents to be merged.\n        :param top_k:\n            The maximum number of Documents to return. Overrides the instance's `top_k` if provided.\n\n        :returns:\n            A dictionary with the following keys:\n            - `documents`: Merged list of Documents\n        \"\"\"\n        output_documents = []\n\n        documents = list(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        elif self.join_mode == \"distribution_based_rank_fusion\":\n            output_documents = self._distribution_based_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 top_k:\n            output_documents = output_documents[:top_k]\n        elif self.top_k:\n            output_documents = output_documents[: self.top_k]\n\n        return {\"documents\": output_documents}\n\n    def _concatenate(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Concatenate multiple lists of Documents and return only the Document with the highest score for duplicates.\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: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and calculate a weighted sum of the scores of duplicate Documents.\n        \"\"\"\n        scores_map: dict = 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 list(documents_map.values())\n\n    def _reciprocal_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on reciprocal rank fusion.\n\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: dict = 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 list(documents_map.values())\n\n    def _distribution_based_rank_fusion(self, document_lists: List[List[Document]]) -> List[Document]:\n        \"\"\"\n        Merge multiple lists of Documents and assign scores based on Distribution-Based Score Fusion.\n\n        (https://medium.com/plain-simple-software/distribution-based-score-fusion-dbsf-a-new-approach-to-vector-search-ranking-f87c37488b18)\n        If a Document is in more than one retriever, the one with the highest score is used.\n        \"\"\"\n        for documents in document_lists:\n            scores_list = []\n\n            for doc in documents:\n                scores_list.append(doc.score if doc.score is not None else 0)\n\n            mean_score = sum(scores_list) / len(scores_list)\n            std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5\n            min_score = mean_score - 3 * std_dev\n            max_score = mean_score + 3 * std_dev\n            delta_score = max_score - min_score\n\n            for doc in documents:\n                doc.score = (doc.score - min_score) / delta_score if delta_score != 0.0 else 0.0\n                # if all docs have the same score delta_score is 0, the docs are uninformative for the query\n\n        output = self._concatenate(document_lists=document_lists)\n\n        return output\n", "path": "haystack/components/joiners/document_joiner.py"}]}