We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
JoinDocuments should use highest score when multiple retrievers recall the same document
**Is your feature request related to a problem? Please describe.**
The JoinDocuments node currently uses the document from the last retriever if their are multiple retrievers and they recall the same document. The first retriever could have the highest score and be more useful. That's why I propose to use the document with the highest score if there are duplicate documents recalled


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

{"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@@ -1,11 +1,10 @@\n-from collections import defaultdict\n import logging\n+from collections import defaultdict\n from math import inf\n+from typing import List, Optional\n \n-from typing import Optional, List\n-\n-from haystack.schema import Document\n from haystack.nodes.other.join import JoinNode\n+from haystack.schema import Document\n \n logger = logging.getLogger(__name__)\n \n@@ -64,7 +63,7 @@\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)\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@@ -118,11 +117,22 @@\n \n             return output, \"output_1\"\n \n-    def _concatenate_results(self, results):\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-        return {doc.id: doc.score for result in results for doc in result}\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", "issue": "JoinDocuments should use highest score when multiple retrievers recall the same document\n**Is your feature request related to a problem? Please describe.**\r\nThe JoinDocuments node currently uses the document from the last retriever if their are multiple retrievers and they recall the same document. The first retriever could have the highest score and be more useful. That's why I propose to use the document with the highest score if there are duplicate documents recalled\r\n\n", "before_files": [{"content": "from collections import defaultdict\nimport logging\nfrom math import inf\n\nfrom typing import Optional, List\n\nfrom haystack.schema import Document\nfrom haystack.nodes.other.join import JoinNode\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)\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):\n        \"\"\"\n        Concatenates multiple document result lists.\n        \"\"\"\n        return {doc.id: doc.score for result in results for doc in result}\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. 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"}]}