We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`JoinDocuments` now fails if any document is without a score
**Describe the bug**
Since the latest release (more specifically this commit: https://github.com/deepset-ai/haystack/commit/32e87d37c153cc8e8c5de5c72e5f939b2efd1022) the `JoinDocument` node will error out if any of the documents joined has not score, even if the `JoinDocuments` is used with `sort_by_score=False`

This break the documented functionality of the `sort_by_score` param, which allows (well, allowed before the above commit) the use of JoinDocuments with documents without scores:
https://github.com/deepset-ai/haystack/blob/main/haystack/nodes/other/join_docs.py#L40

Also this likely means that there is no unit test to verify that `sort_by_score=False` actually works, otherwise that should have caught this bug when it got introduced.

**Error message**
The error you will be seeing if you have some documents without scores even if you have `sort_by_score=False` set :
```
  File ".../lib/python3.10/site-packages/haystack/nodes/base.py", line 317, in _adispatch_run_general
    output, stream = await run_method(**run_inputs, **run_params)
                           │            │             └ {}
                           │            └ {'inputs': [{'documents': [<Document: {'content': "Zoltan Fedor is the Director, AI Platform ...
                           └ <bound method JoinNode.run of <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>>
  File ".../lib/python3.10/site-packages/haystack/nodes/other/join.py", line 24, in run
    results = self.run_accumulated(inputs, top_k_join=top_k_join)
              │    │               │                  └ None
              │    │               └ [{'documents': [<Document: {'content': "Zoltan Fedor is the Director, AI Platform ...
              │    └ <function JoinDocuments.run_accumulated at 0x7f7af5d160e0>
              └ <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>
  File ".../lib/python3.10/site-packages/haystack/nodes/other/join_docs.py", line 66, in run_accumulated
    scores_map = self._concatenate_results(results, document_map)
                 │    │                    │        └ {'8a51d965-082b-055f-f486-a372fc7254e6': <Document: {'content': "Zoltan Fedor is the Director, AI Platform ...
                 │    │                    └ [[<Document: {'content': "Zoltan Fedor is the Director, AI Platform ...
                 │    └ <function JoinDocuments._concatenate_results at 0x7f7af5d164d0>
                 └ <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>
  File ".../lib/python3.10/site-packages/haystack/nodes/other/join_docs.py", line 133, in _concatenate_results
    item_best_score = max(tmp, key=lambda x: x.score)
                          └ [<Document: {'content': "Zoltan Fedor is the Director, AI Platform ...

TypeError: '>' not supported between instances of 'NoneType' and 'float'
```

**Expected behavior**
The document functionality of `sort_by_score=False` of `JoinDocuments` should not get broken by new functionality being introduced.

**Additional context**
As mentioned above this bug could have not been introduced it there would have been a unit test for the `sort_by_score=False` functionality - so I would recommend adding one to avoid this happening again in the future.

**To Reproduce**
Create a document set with some documents with no score and use `JoinDocuments` on them.

**FAQ Check**
- [ X ] Have you had a look at [our new FAQ page](https://docs.haystack.deepset.ai/docs/faq)?

**System:**
 - OS: all
 - GPU/CPU: all
 - Haystack version (commit or version number): 1.22.0
 - DocumentStore: Any
 - Reader: Any
 - Retriever: Any that produces some documents with no score


--- 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/nodes/other/join_docs.py`
Content:
```
1 import logging
2 from collections import defaultdict
3 from math import inf
4 from typing import List, Optional
5 
6 from haystack.nodes.other.join import JoinNode
7 from haystack.schema import Document
8 
9 logger = logging.getLogger(__name__)
10 
11 
12 class JoinDocuments(JoinNode):
13     """
14     A node to join documents outputted by multiple retriever nodes.
15 
16     The node allows multiple join modes:
17     * concatenate: combine the documents from multiple nodes. Any duplicate documents are discarded.
18                    The score is only determined by the last node that outputs the document.
19     * merge: merge scores of documents from multiple nodes. Optionally, each input score can be given a different
20              `weight` & a `top_k` limit can be set. This mode can also be used for "reranking" retrieved documents.
21     * reciprocal_rank_fusion: combines the documents based on their rank in multiple nodes.
22     """
23 
24     outgoing_edges = 1
25 
26     def __init__(
27         self,
28         join_mode: str = "concatenate",
29         weights: Optional[List[float]] = None,
30         top_k_join: Optional[int] = None,
31         sort_by_score: bool = True,
32     ):
33         """
34         :param join_mode: `concatenate` to combine documents from multiple retrievers `merge` to aggregate scores of
35                           individual documents, `reciprocal_rank_fusion` to apply rank based scoring.
36         :param weights: A node-wise list(length of list must be equal to the number of input nodes) of weights for
37                         adjusting document scores when using the `merge` join_mode. By default, equal weight is given
38                         to each retriever score. This param is not compatible with the `concatenate` join_mode.
39         :param top_k_join: Limit documents to top_k based on the resulting scores of the join.
40         :param sort_by_score: Whether to sort the incoming documents by their score. Set this to True if all your
41                               Documents are coming with `score` values. Set to False if any of the Documents come
42                               from sources where the `score` is set to `None`, like `TfidfRetriever` on Elasticsearch.
43         """
44         assert join_mode in [
45             "concatenate",
46             "merge",
47             "reciprocal_rank_fusion",
48         ], f"JoinDocuments node does not support '{join_mode}' join_mode."
49 
50         assert not (
51             weights is not None and join_mode == "concatenate"
52         ), "Weights are not compatible with 'concatenate' join_mode."
53 
54         super().__init__()
55 
56         self.join_mode = join_mode
57         self.weights = [float(i) / sum(weights) for i in weights] if weights else None
58         self.top_k_join = top_k_join
59         self.sort_by_score = sort_by_score
60 
61     def run_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore
62         results = [inp["documents"] for inp in inputs]
63         document_map = {doc.id: doc for result in results for doc in result}
64 
65         if self.join_mode == "concatenate":
66             scores_map = self._concatenate_results(results, document_map)
67         elif self.join_mode == "merge":
68             scores_map = self._calculate_comb_sum(results)
69         elif self.join_mode == "reciprocal_rank_fusion":
70             scores_map = self._calculate_rrf(results)
71         else:
72             raise ValueError(f"Invalid join_mode: {self.join_mode}")
73 
74         # only sort the docs if that was requested
75         if self.sort_by_score:
76             sorted_docs = sorted(scores_map.items(), key=lambda d: d[1] if d[1] is not None else -inf, reverse=True)
77             if any(s is None for s in scores_map.values()):
78                 logger.info(
79                     "The `JoinDocuments` node has received some documents with `score=None` - and was requested "
80                     "to sort the documents by score, so the `score=None` documents got sorted as if their "
81                     "score would be `-infinity`."
82                 )
83         else:
84             sorted_docs = list(scores_map.items())
85 
86         if not top_k_join:
87             top_k_join = self.top_k_join
88         if not top_k_join:
89             top_k_join = len(sorted_docs)
90 
91         docs = []
92         for id, score in sorted_docs[:top_k_join]:
93             doc = document_map[id]
94             doc.score = score
95             docs.append(doc)
96 
97         output = {"documents": docs, "labels": inputs[0].get("labels", None)}
98 
99         return output, "output_1"
100 
101     def run_batch_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore
102         # Join single document lists
103         if isinstance(inputs[0]["documents"][0], Document):
104             return self.run(inputs=inputs, top_k_join=top_k_join)
105         # Join lists of document lists
106         else:
107             output_docs = []
108             incoming_edges = [inp["documents"] for inp in inputs]
109             for idx in range(len(incoming_edges[0])):
110                 cur_docs_to_join = []
111                 for edge in incoming_edges:
112                     cur_docs_to_join.append({"documents": edge[idx]})
113                 cur, _ = self.run(inputs=cur_docs_to_join, top_k_join=top_k_join)
114                 output_docs.append(cur["documents"])
115 
116             output = {"documents": output_docs, "labels": inputs[0].get("labels", None)}
117 
118             return output, "output_1"
119 
120     def _concatenate_results(self, results, document_map):
121         """
122         Concatenates multiple document result lists.
123         Return the documents with the higher score.
124         """
125         list_id = list(document_map.keys())
126         scores_map = {}
127         for idx in list_id:
128             tmp = []
129             for result in results:
130                 for doc in result:
131                     if doc.id == idx:
132                         tmp.append(doc)
133             item_best_score = max(tmp, key=lambda x: x.score)
134             scores_map.update({idx: item_best_score.score})
135         return scores_map
136 
137     def _calculate_comb_sum(self, results):
138         """
139         Calculates a combination sum by multiplying each score by its weight.
140         """
141         scores_map = defaultdict(int)
142         weights = self.weights if self.weights else [1 / len(results)] * len(results)
143 
144         for result, weight in zip(results, weights):
145             for doc in result:
146                 scores_map[doc.id] += (doc.score if doc.score else 0) * weight
147 
148         return scores_map
149 
150     def _calculate_rrf(self, results):
151         """
152         Calculates the reciprocal rank fusion. The constant K is set to 61 (60 was suggested by the original paper,
153         plus 1 as python lists are 0-based and the paper used 1-based ranking).
154         """
155         K = 61
156 
157         scores_map = defaultdict(int)
158         for result in results:
159             for rank, doc in enumerate(result):
160                 scores_map[doc.id] += 1 / (K + rank)
161 
162         return scores_map
163
```
--- 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/nodes/other/join_docs.py b/haystack/nodes/other/join_docs.py
--- a/haystack/nodes/other/join_docs.py
+++ b/haystack/nodes/other/join_docs.py
@@ -14,8 +14,8 @@
     A node to join documents outputted by multiple retriever nodes.
 
     The node allows multiple join modes:
-    * concatenate: combine the documents from multiple nodes. Any duplicate documents are discarded.
-                   The score is only determined by the last node that outputs the document.
+    * concatenate: combine the documents from multiple nodes.
+                   In case of duplicate documents, the one with the highest score is kept.
     * merge: merge scores of documents from multiple nodes. Optionally, each input score can be given a different
              `weight` & a `top_k` limit can be set. This mode can also be used for "reranking" retrieved documents.
     * reciprocal_rank_fusion: combines the documents based on their rank in multiple nodes.
@@ -130,7 +130,7 @@
                 for doc in result:
                     if doc.id == idx:
                         tmp.append(doc)
-            item_best_score = max(tmp, key=lambda x: x.score)
+            item_best_score = max(tmp, key=lambda x: x.score if x.score is not None else -inf)
             scores_map.update({idx: item_best_score.score})
         return scores_map

{"golden_diff": "diff --git a/haystack/nodes/other/join_docs.py b/haystack/nodes/other/join_docs.py\n--- a/haystack/nodes/other/join_docs.py\n+++ b/haystack/nodes/other/join_docs.py\n@@ -14,8 +14,8 @@\n     A node to join documents outputted by multiple retriever nodes.\n \n     The node allows multiple join modes:\n-    * concatenate: combine the documents from multiple nodes. Any duplicate documents are discarded.\n-                   The score is only determined by the last node that outputs the document.\n+    * concatenate: combine the documents from multiple nodes.\n+                   In case of duplicate documents, the one with the highest score is kept.\n     * merge: merge scores of documents from multiple nodes. Optionally, each input score can be given a different\n              `weight` & a `top_k` limit can be set. This mode can also be used for \"reranking\" retrieved documents.\n     * reciprocal_rank_fusion: combines the documents based on their rank in multiple nodes.\n@@ -130,7 +130,7 @@\n                 for doc in result:\n                     if doc.id == idx:\n                         tmp.append(doc)\n-            item_best_score = max(tmp, key=lambda x: x.score)\n+            item_best_score = max(tmp, key=lambda x: x.score if x.score is not None else -inf)\n             scores_map.update({idx: item_best_score.score})\n         return scores_map\n", "issue": "`JoinDocuments` now fails if any document is without a score\n**Describe the bug**\r\nSince the latest release (more specifically this commit: https://github.com/deepset-ai/haystack/commit/32e87d37c153cc8e8c5de5c72e5f939b2efd1022) the `JoinDocument` node will error out if any of the documents joined has not score, even if the `JoinDocuments` is used with `sort_by_score=False`\r\n\r\nThis break the documented functionality of the `sort_by_score` param, which allows (well, allowed before the above commit) the use of JoinDocuments with documents without scores:\r\nhttps://github.com/deepset-ai/haystack/blob/main/haystack/nodes/other/join_docs.py#L40\r\n\r\nAlso this likely means that there is no unit test to verify that `sort_by_score=False` actually works, otherwise that should have caught this bug when it got introduced.\r\n\r\n**Error message**\r\nThe error you will be seeing if you have some documents without scores even if you have `sort_by_score=False` set :\r\n```\r\n  File \".../lib/python3.10/site-packages/haystack/nodes/base.py\", line 317, in _adispatch_run_general\r\n    output, stream = await run_method(**run_inputs, **run_params)\r\n                           \u2502            \u2502             \u2514 {}\r\n                           \u2502            \u2514 {'inputs': [{'documents': [<Document: {'content': \"Zoltan Fedor is the Director, AI Platform ...\r\n                           \u2514 <bound method JoinNode.run of <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>>\r\n  File \".../lib/python3.10/site-packages/haystack/nodes/other/join.py\", line 24, in run\r\n    results = self.run_accumulated(inputs, top_k_join=top_k_join)\r\n              \u2502    \u2502               \u2502                  \u2514 None\r\n              \u2502    \u2502               \u2514 [{'documents': [<Document: {'content': \"Zoltan Fedor is the Director, AI Platform ...\r\n              \u2502    \u2514 <function JoinDocuments.run_accumulated at 0x7f7af5d160e0>\r\n              \u2514 <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>\r\n  File \".../lib/python3.10/site-packages/haystack/nodes/other/join_docs.py\", line 66, in run_accumulated\r\n    scores_map = self._concatenate_results(results, document_map)\r\n                 \u2502    \u2502                    \u2502        \u2514 {'8a51d965-082b-055f-f486-a372fc7254e6': <Document: {'content': \"Zoltan Fedor is the Director, AI Platform ...\r\n                 \u2502    \u2502                    \u2514 [[<Document: {'content': \"Zoltan Fedor is the Director, AI Platform ...\r\n                 \u2502    \u2514 <function JoinDocuments._concatenate_results at 0x7f7af5d164d0>\r\n                 \u2514 <haystack.nodes.other.join_docs.JoinDocuments object at 0x7f7beae0df30>\r\n  File \".../lib/python3.10/site-packages/haystack/nodes/other/join_docs.py\", line 133, in _concatenate_results\r\n    item_best_score = max(tmp, key=lambda x: x.score)\r\n                          \u2514 [<Document: {'content': \"Zoltan Fedor is the Director, AI Platform ...\r\n\r\nTypeError: '>' not supported between instances of 'NoneType' and 'float'\r\n```\r\n\r\n**Expected behavior**\r\nThe document functionality of `sort_by_score=False` of `JoinDocuments` should not get broken by new functionality being introduced.\r\n\r\n**Additional context**\r\nAs mentioned above this bug could have not been introduced it there would have been a unit test for the `sort_by_score=False` functionality - so I would recommend adding one to avoid this happening again in the future.\r\n\r\n**To Reproduce**\r\nCreate a document set with some documents with no score and use `JoinDocuments` on them.\r\n\r\n**FAQ Check**\r\n- [ X ] Have you had a look at [our new FAQ page](https://docs.haystack.deepset.ai/docs/faq)?\r\n\r\n**System:**\r\n - OS: all\r\n - GPU/CPU: all\r\n - Haystack version (commit or version number): 1.22.0\r\n - DocumentStore: Any\r\n - Reader: Any\r\n - Retriever: Any that produces some documents with no score\r\n\n", "before_files": [{"content": "import logging\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\n\nfrom haystack.nodes.other.join import JoinNode\nfrom haystack.schema import Document\n\nlogger = logging.getLogger(__name__)\n\n\nclass JoinDocuments(JoinNode):\n    \"\"\"\n    A node to join documents outputted by multiple retriever nodes.\n\n    The node allows multiple join modes:\n    * concatenate: combine the documents from multiple nodes. Any duplicate documents are discarded.\n                   The score is only determined by the last node that outputs the document.\n    * merge: merge scores of documents from multiple nodes. Optionally, each input score can be given a different\n             `weight` & a `top_k` limit can be set. This mode can also be used for \"reranking\" retrieved documents.\n    * reciprocal_rank_fusion: combines the documents based on their rank in multiple nodes.\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def __init__(\n        self,\n        join_mode: str = \"concatenate\",\n        weights: Optional[List[float]] = None,\n        top_k_join: Optional[int] = None,\n        sort_by_score: bool = True,\n    ):\n        \"\"\"\n        :param join_mode: `concatenate` to combine documents from multiple retrievers `merge` to aggregate scores of\n                          individual documents, `reciprocal_rank_fusion` to apply rank based scoring.\n        :param weights: A node-wise list(length of list must be equal to the number of input nodes) 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_join: Limit documents to top_k based on the resulting scores of the join.\n        :param sort_by_score: Whether to sort the incoming documents by their score. Set this to True if all your\n                              Documents are coming with `score` values. Set to False if any of the Documents come\n                              from sources where the `score` is set to `None`, like `TfidfRetriever` on Elasticsearch.\n        \"\"\"\n        assert join_mode in [\n            \"concatenate\",\n            \"merge\",\n            \"reciprocal_rank_fusion\",\n        ], f\"JoinDocuments node does not support '{join_mode}' join_mode.\"\n\n        assert not (\n            weights is not None and join_mode == \"concatenate\"\n        ), \"Weights are not compatible with 'concatenate' join_mode.\"\n\n        super().__init__()\n\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_join = top_k_join\n        self.sort_by_score = sort_by_score\n\n    def run_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore\n        results = [inp[\"documents\"] for inp in inputs]\n        document_map = {doc.id: doc for result in results for doc in result}\n\n        if self.join_mode == \"concatenate\":\n            scores_map = self._concatenate_results(results, document_map)\n        elif self.join_mode == \"merge\":\n            scores_map = self._calculate_comb_sum(results)\n        elif self.join_mode == \"reciprocal_rank_fusion\":\n            scores_map = self._calculate_rrf(results)\n        else:\n            raise ValueError(f\"Invalid join_mode: {self.join_mode}\")\n\n        # only sort the docs if that was requested\n        if self.sort_by_score:\n            sorted_docs = sorted(scores_map.items(), key=lambda d: d[1] if d[1] is not None else -inf, reverse=True)\n            if any(s is None for s in scores_map.values()):\n                logger.info(\n                    \"The `JoinDocuments` node has received some documents with `score=None` - and was requested \"\n                    \"to sort the documents by score, so the `score=None` documents got sorted as if their \"\n                    \"score would be `-infinity`.\"\n                )\n        else:\n            sorted_docs = list(scores_map.items())\n\n        if not top_k_join:\n            top_k_join = self.top_k_join\n        if not top_k_join:\n            top_k_join = len(sorted_docs)\n\n        docs = []\n        for id, score in sorted_docs[:top_k_join]:\n            doc = document_map[id]\n            doc.score = score\n            docs.append(doc)\n\n        output = {\"documents\": docs, \"labels\": inputs[0].get(\"labels\", None)}\n\n        return output, \"output_1\"\n\n    def run_batch_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore\n        # Join single document lists\n        if isinstance(inputs[0][\"documents\"][0], Document):\n            return self.run(inputs=inputs, top_k_join=top_k_join)\n        # Join lists of document lists\n        else:\n            output_docs = []\n            incoming_edges = [inp[\"documents\"] for inp in inputs]\n            for idx in range(len(incoming_edges[0])):\n                cur_docs_to_join = []\n                for edge in incoming_edges:\n                    cur_docs_to_join.append({\"documents\": edge[idx]})\n                cur, _ = self.run(inputs=cur_docs_to_join, top_k_join=top_k_join)\n                output_docs.append(cur[\"documents\"])\n\n            output = {\"documents\": output_docs, \"labels\": inputs[0].get(\"labels\", None)}\n\n            return output, \"output_1\"\n\n    def _concatenate_results(self, results, document_map):\n        \"\"\"\n        Concatenates multiple document result lists.\n        Return the documents with the higher score.\n        \"\"\"\n        list_id = list(document_map.keys())\n        scores_map = {}\n        for idx in list_id:\n            tmp = []\n            for result in results:\n                for doc in result:\n                    if doc.id == idx:\n                        tmp.append(doc)\n            item_best_score = max(tmp, key=lambda x: x.score)\n            scores_map.update({idx: item_best_score.score})\n        return scores_map\n\n    def _calculate_comb_sum(self, results):\n        \"\"\"\n        Calculates a combination sum by multiplying each score by its weight.\n        \"\"\"\n        scores_map = defaultdict(int)\n        weights = self.weights if self.weights else [1 / len(results)] * len(results)\n\n        for result, weight in zip(results, weights):\n            for doc in result:\n                scores_map[doc.id] += (doc.score if doc.score else 0) * weight\n\n        return scores_map\n\n    def _calculate_rrf(self, results):\n        \"\"\"\n        Calculates the reciprocal rank fusion. 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        for result in results:\n            for rank, doc in enumerate(result):\n                scores_map[doc.id] += 1 / (K + rank)\n\n        return scores_map\n", "path": "haystack/nodes/other/join_docs.py"}], "after_files": [{"content": "import logging\nfrom collections import defaultdict\nfrom math import inf\nfrom typing import List, Optional\n\nfrom haystack.nodes.other.join import JoinNode\nfrom haystack.schema import Document\n\nlogger = logging.getLogger(__name__)\n\n\nclass JoinDocuments(JoinNode):\n    \"\"\"\n    A node to join documents outputted by multiple retriever nodes.\n\n    The node allows multiple join modes:\n    * concatenate: combine the documents from multiple nodes.\n                   In case of duplicate documents, the one with the highest score is kept.\n    * merge: merge scores of documents from multiple nodes. Optionally, each input score can be given a different\n             `weight` & a `top_k` limit can be set. This mode can also be used for \"reranking\" retrieved documents.\n    * reciprocal_rank_fusion: combines the documents based on their rank in multiple nodes.\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def __init__(\n        self,\n        join_mode: str = \"concatenate\",\n        weights: Optional[List[float]] = None,\n        top_k_join: Optional[int] = None,\n        sort_by_score: bool = True,\n    ):\n        \"\"\"\n        :param join_mode: `concatenate` to combine documents from multiple retrievers `merge` to aggregate scores of\n                          individual documents, `reciprocal_rank_fusion` to apply rank based scoring.\n        :param weights: A node-wise list(length of list must be equal to the number of input nodes) 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_join: Limit documents to top_k based on the resulting scores of the join.\n        :param sort_by_score: Whether to sort the incoming documents by their score. Set this to True if all your\n                              Documents are coming with `score` values. Set to False if any of the Documents come\n                              from sources where the `score` is set to `None`, like `TfidfRetriever` on Elasticsearch.\n        \"\"\"\n        assert join_mode in [\n            \"concatenate\",\n            \"merge\",\n            \"reciprocal_rank_fusion\",\n        ], f\"JoinDocuments node does not support '{join_mode}' join_mode.\"\n\n        assert not (\n            weights is not None and join_mode == \"concatenate\"\n        ), \"Weights are not compatible with 'concatenate' join_mode.\"\n\n        super().__init__()\n\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_join = top_k_join\n        self.sort_by_score = sort_by_score\n\n    def run_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore\n        results = [inp[\"documents\"] for inp in inputs]\n        document_map = {doc.id: doc for result in results for doc in result}\n\n        if self.join_mode == \"concatenate\":\n            scores_map = self._concatenate_results(results, document_map)\n        elif self.join_mode == \"merge\":\n            scores_map = self._calculate_comb_sum(results)\n        elif self.join_mode == \"reciprocal_rank_fusion\":\n            scores_map = self._calculate_rrf(results)\n        else:\n            raise ValueError(f\"Invalid join_mode: {self.join_mode}\")\n\n        # only sort the docs if that was requested\n        if self.sort_by_score:\n            sorted_docs = sorted(scores_map.items(), key=lambda d: d[1] if d[1] is not None else -inf, reverse=True)\n            if any(s is None for s in scores_map.values()):\n                logger.info(\n                    \"The `JoinDocuments` node has received some documents with `score=None` - and was requested \"\n                    \"to sort the documents by score, so the `score=None` documents got sorted as if their \"\n                    \"score would be `-infinity`.\"\n                )\n        else:\n            sorted_docs = list(scores_map.items())\n\n        if not top_k_join:\n            top_k_join = self.top_k_join\n        if not top_k_join:\n            top_k_join = len(sorted_docs)\n\n        docs = []\n        for id, score in sorted_docs[:top_k_join]:\n            doc = document_map[id]\n            doc.score = score\n            docs.append(doc)\n\n        output = {\"documents\": docs, \"labels\": inputs[0].get(\"labels\", None)}\n\n        return output, \"output_1\"\n\n    def run_batch_accumulated(self, inputs: List[dict], top_k_join: Optional[int] = None):  # type: ignore\n        # Join single document lists\n        if isinstance(inputs[0][\"documents\"][0], Document):\n            return self.run(inputs=inputs, top_k_join=top_k_join)\n        # Join lists of document lists\n        else:\n            output_docs = []\n            incoming_edges = [inp[\"documents\"] for inp in inputs]\n            for idx in range(len(incoming_edges[0])):\n                cur_docs_to_join = []\n                for edge in incoming_edges:\n                    cur_docs_to_join.append({\"documents\": edge[idx]})\n                cur, _ = self.run(inputs=cur_docs_to_join, top_k_join=top_k_join)\n                output_docs.append(cur[\"documents\"])\n\n            output = {\"documents\": output_docs, \"labels\": inputs[0].get(\"labels\", None)}\n\n            return output, \"output_1\"\n\n    def _concatenate_results(self, results, document_map):\n        \"\"\"\n        Concatenates multiple document result lists.\n        Return the documents with the higher score.\n        \"\"\"\n        list_id = list(document_map.keys())\n        scores_map = {}\n        for idx in list_id:\n            tmp = []\n            for result in results:\n                for doc in result:\n                    if doc.id == idx:\n                        tmp.append(doc)\n            item_best_score = max(tmp, key=lambda x: x.score if x.score is not None else -inf)\n            scores_map.update({idx: item_best_score.score})\n        return scores_map\n\n    def _calculate_comb_sum(self, results):\n        \"\"\"\n        Calculates a combination sum by multiplying each score by its weight.\n        \"\"\"\n        scores_map = defaultdict(int)\n        weights = self.weights if self.weights else [1 / len(results)] * len(results)\n\n        for result, weight in zip(results, weights):\n            for doc in result:\n                scores_map[doc.id] += (doc.score if doc.score else 0) * weight\n\n        return scores_map\n\n    def _calculate_rrf(self, results):\n        \"\"\"\n        Calculates the reciprocal rank fusion. 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        for result in results:\n            for rank, doc in enumerate(result):\n                scores_map[doc.id] += 1 / (K + rank)\n\n        return scores_map\n", "path": "haystack/nodes/other/join_docs.py"}]}