We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Combination of multiple retrievers
Originating from a discussion in #104. 

**What**
Combine multiple scores during retrieval (e.g. embeddings similarity of multiple text fields, or BM25 + Embedding similarity ...) 

**Options**

a) Concatenation of Retrievers: Make the Finder accept a list of Retrievers and combine their scores (incl. weights) before feeding results to the Reader
Pro: most generic solution also allowing combination of BM25 + EmbeddingRetriever
Con: Two separate retriever queries (suboptimal efficiency in some cases)

b) New MultiEmbeddingRetriever: Extend the EmbeddingRetriever to the case of multiple Embeddings
Pro: simple implementation without side effects on the regular user interface
Con: won't allow the combination of other retriever methods (e.g. ElasticsearchRetriever + EmbeddingRetriver)

I am leaning towards a), but could see b) as a simpler short-term solution for the original use case discussed in #104 .



┆Issue is synchronized with this [Jira Task](https://deepset-ai.atlassian.net/browse/HS-42) by [Unito](https://www.unito.io/learn-more)


--- 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/pipeline.py`
Content:
```
1 from pathlib import Path
2 from typing import List, Optional, Dict
3 
4 import networkx as nx
5 from networkx import DiGraph
6 from networkx.drawing.nx_agraph import to_agraph
7 
8 from haystack.generator.base import BaseGenerator
9 from haystack.reader.base import BaseReader
10 from haystack.retriever.base import BaseRetriever
11 
12 
13 class Pipeline:
14     """
15     Pipeline brings together building blocks to build a complex search pipeline with Haystack & user-defined components.
16 
17     Under-the-hood, a pipeline is represented as a directed acyclic graph of component nodes. It enables custom query
18     flows with options to branch queries(eg, extractive qa vs keyword match query), merge candidate documents for a
19     Reader from multiple Retrievers, or re-ranking of candidate documents.
20     """
21 
22     def __init__(self):
23         self.graph = DiGraph()
24         self.root_node_id = "Query"
25         self.graph.add_node("Query", component=QueryNode())
26 
27     def add_node(self, component, name: str, inputs: List[str]):
28         """
29         Add a new node to the pipeline.
30 
31         :param component: The object to be called when the data is passed to the node. It can be a Haystack component
32                           (like Retriever, Reader, or Generator) or a user-defined object that implements a run()
33                           method to process incoming data from predecessor node.
34         :param name: The name for the node. It must not contain any dots.
35         :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name
36                        of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single
37                        edge with a list of documents. It can be represented as ["ElasticsearchRetriever"].
38 
39                        In cases when the predecessor node has multiple outputs, e.g., a "QueryClassifier", the output
40                        must be specified explicitly as "QueryClassifier.output_2".
41         """
42         self.graph.add_node(name, component=component)
43 
44         for i in inputs:
45             if "." in i:
46                 [input_node_name, input_edge_name] = i.split(".")
47                 assert "output_" in input_edge_name, f"'{input_edge_name}' is not a valid edge name."
48                 outgoing_edges_input_node = self.graph.nodes[input_node_name]["component"].outgoing_edges
49                 assert int(input_edge_name.split("_")[1]) <= outgoing_edges_input_node, (
50                     f"Cannot connect '{input_edge_name}' from '{input_node_name}' as it only has "
51                     f"{outgoing_edges_input_node} outgoing edge(s)."
52                 )
53             else:
54                 outgoing_edges_input_node = self.graph.nodes[i]["component"].outgoing_edges
55                 assert outgoing_edges_input_node == 1, (
56                     f"Adding an edge from {i} to {name} is ambiguous as {i} has {outgoing_edges_input_node} edges. "
57                     f"Please specify the output explicitly."
58                 )
59                 input_node_name = i
60                 input_edge_name = "output_1"
61             self.graph.add_edge(input_node_name, name, label=input_edge_name)
62 
63     def get_node(self, name: str):
64         """
65         Get a node from the Pipeline.
66 
67         :param name: The name of the node.
68         """
69         component = self.graph.nodes[name]["component"]
70         return component
71 
72     def set_node(self, name: str, component):
73         """
74         Set the component for a node in the Pipeline.
75 
76         :param name: The name of the node.
77         :param component: The component object to be set at the node.
78         """
79         self.graph.nodes[name]["component"] = component
80 
81     def run(self, **kwargs):
82         has_next_node = True
83         current_node_id = self.root_node_id
84         input_dict = kwargs
85         output_dict = None
86 
87         while has_next_node:
88             output_dict, stream_id = self.graph.nodes[current_node_id]["component"].run(**input_dict)
89             input_dict = output_dict
90             next_nodes = self._get_next_nodes(current_node_id, stream_id)
91 
92             if len(next_nodes) > 1:
93                 join_node_id = list(nx.neighbors(self.graph, next_nodes[0]))[0]
94                 if set(self.graph.predecessors(join_node_id)) != set(next_nodes):
95                     raise NotImplementedError(
96                         "The current pipeline does not support multiple levels of parallel nodes."
97                     )
98                 inputs_for_join_node = {"inputs": []}
99                 for n_id in next_nodes:
100                     output = self.graph.nodes[n_id]["component"].run(**input_dict)
101                     inputs_for_join_node["inputs"].append(output)
102                 input_dict = inputs_for_join_node
103                 current_node_id = join_node_id
104             elif len(next_nodes) == 1:
105                 current_node_id = next_nodes[0]
106             else:
107                 has_next_node = False
108 
109         return output_dict
110 
111     def _get_next_nodes(self, node_id: str, stream_id: str):
112         current_node_edges = self.graph.edges(node_id, data=True)
113         next_nodes = [
114             next_node
115             for _, next_node, data in current_node_edges
116             if not stream_id or data["label"] == stream_id
117         ]
118         return next_nodes
119 
120     def draw(self, path: Path = Path("pipeline.png")):
121         """
122         Create a Graphviz visualization of the pipeline.
123 
124         :param path: the path to save the image.
125         """
126         try:
127             import pygraphviz
128         except ImportError:
129             raise ImportError(f"Could not import `pygraphviz`. Please install via: \n"
130                               f"pip install pygraphviz\n"
131                               f"(You might need to run this first: apt install libgraphviz-dev graphviz )")
132 
133         graphviz = to_agraph(self.graph)
134         graphviz.layout("dot")
135         graphviz.draw(path)
136 
137 
138 class BaseStandardPipeline:
139     pipeline: Pipeline
140 
141     def add_node(self, component, name: str, inputs: List[str]):
142         """
143         Add a new node to the pipeline.
144 
145         :param component: The object to be called when the data is passed to the node. It can be a Haystack component
146                           (like Retriever, Reader, or Generator) or a user-defined object that implements a run()
147                           method to process incoming data from predecessor node.
148         :param name: The name for the node. It must not contain any dots.
149         :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name
150                        of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single
151                        edge with a list of documents. It can be represented as ["ElasticsearchRetriever"].
152 
153                        In cases when the predecessor node has multiple outputs, e.g., a "QueryClassifier", the output
154                        must be specified explicitly as "QueryClassifier.output_2".
155         """
156 
157         self.pipeline.add_node(component=component, name=name, inputs=inputs)
158 
159     def get_node(self, name: str):
160         """
161         Get a node from the Pipeline.
162 
163         :param name: The name of the node.
164         """
165         component = self.pipeline.get_node(name)
166         return component
167 
168     def set_node(self, name: str, component):
169         """
170         Set the component for a node in the Pipeline.
171 
172         :param name: The name of the node.
173         :param component: The component object to be set at the node.
174         """
175         self.pipeline.set_node(name, component)
176 
177     def draw(self, path: Path = Path("pipeline.png")):
178         """
179         Create a Graphviz visualization of the pipeline.
180 
181         :param path: the path to save the image.
182         """
183         self.pipeline.draw(path)
184 
185 
186 class ExtractiveQAPipeline(BaseStandardPipeline):
187     def __init__(self, reader: BaseReader, retriever: BaseRetriever):
188         """
189         Initialize a Pipeline for Extractive Question Answering.
190 
191         :param reader: Reader instance
192         :param retriever: Retriever instance
193         """
194         self.pipeline = Pipeline()
195         self.pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
196         self.pipeline.add_node(component=reader, name="Reader", inputs=["Retriever"])
197 
198     def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_reader: int = 10):
199         output = self.pipeline.run(
200             query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_reader=top_k_reader
201         )
202         return output
203 
204 
205 class DocumentSearchPipeline(BaseStandardPipeline):
206     def __init__(self, retriever: BaseRetriever):
207         """
208         Initialize a Pipeline for semantic document search.
209 
210         :param retriever: Retriever instance
211         """
212         self.pipeline = Pipeline()
213         self.pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
214 
215     def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):
216         output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)
217         document_dicts = [doc.to_dict() for doc in output["documents"]]
218         output["documents"] = document_dicts
219         return output
220 
221 
222 class GenerativeQAPipeline(BaseStandardPipeline):
223     def __init__(self, generator: BaseGenerator, retriever: BaseRetriever):
224         """
225         Initialize a Pipeline for Generative Question Answering.
226 
227         :param generator: Generator instance
228         :param retriever: Retriever instance
229         """
230         self.pipeline = Pipeline()
231         self.pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
232         self.pipeline.add_node(component=generator, name="Generator", inputs=["Retriever"])
233 
234     def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_generator: int = 10):
235         output = self.pipeline.run(
236             query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_generator=top_k_generator
237         )
238         return output
239 
240 
241 class FAQPipeline(BaseStandardPipeline):
242     def __init__(self, retriever: BaseRetriever):
243         """
244         Initialize a Pipeline for finding similar FAQs using semantic document search.
245 
246         :param retriever: Retriever instance
247         """
248         self.pipeline = Pipeline()
249         self.pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
250 
251     def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):
252         output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)
253         documents = output["documents"]
254 
255         results: Dict = {"query": query, "answers": []}
256         for doc in documents:
257             # TODO proper calibratation of pseudo probabilities
258             cur_answer = {
259                 "query": doc.text,
260                 "answer": doc.meta["answer"],
261                 "document_id": doc.id,
262                 "context": doc.meta["answer"],
263                 "score": doc.score,
264                 "probability": doc.probability,
265                 "offset_start": 0,
266                 "offset_end": len(doc.meta["answer"]),
267                 "meta": doc.meta,
268             }
269 
270             results["answers"].append(cur_answer)
271 
272         return results
273 
274 
275 class QueryNode:
276     outgoing_edges = 1
277 
278     def run(self, **kwargs):
279         return kwargs, "output_1"
280 
281 
282 class JoinDocuments:
283     outgoing_edges = 1
284 
285     def __init__(self, join_mode="concatenate"):
286         pass
287 
288     def run(self, **kwargs):
289         inputs = kwargs["inputs"]
290 
291         documents = []
292         for i, _ in inputs:
293             documents.extend(i["documents"])
294         output = {
295             "query": inputs[0][0]["query"],
296             "documents": documents
297         }
298         return output, "output_1"
299
```
--- 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/pipeline.py b/haystack/pipeline.py
--- a/haystack/pipeline.py
+++ b/haystack/pipeline.py
@@ -1,3 +1,4 @@
+from copy import deepcopy
 from pathlib import Path
 from typing import List, Optional, Dict
 
@@ -280,19 +281,63 @@
 
 
 class JoinDocuments:
+    """
+    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.
+    * 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.
+    """
+
     outgoing_edges = 1
 
-    def __init__(self, join_mode="concatenate"):
-        pass
+    def __init__(
+        self, join_mode: str = "concatenate", weights: Optional[List[float]] = None, top_k_join: Optional[int] = None
+    ):
+        """
+        :param join_mode: `concatenate` to combine documents from multiple retrievers or `merge` to aggregate scores of
+                          individual documents.
+        :param weights: A node-wise list(length of list must be equal to the number of input nodes) of weights for
+                        adjusting document scores when using the `merge` join_mode. By default, equal weight is given
+                        to each retriever score. This param is not compatible with the `concatenate` join_mode.
+        :param top_k_join: Limit documents to top_k based on the resulting scores of the join.
+        """
+        assert join_mode in ["concatenate", "merge"], f"JoinDocuments node does not support '{join_mode}' join_mode."
+
+        assert not (
+            weights is not None and join_mode == "concatenate"
+        ), "Weights are not compatible with 'concatenate' join_mode."
+        self.join_mode = join_mode
+        self.weights = weights
+        self.top_k = top_k_join
 
     def run(self, **kwargs):
         inputs = kwargs["inputs"]
 
-        documents = []
-        for i, _ in inputs:
-            documents.extend(i["documents"])
-        output = {
-            "query": inputs[0][0]["query"],
-            "documents": documents
-        }
+        if self.join_mode == "concatenate":
+            document_map = {}
+            for input_from_node, _ in inputs:
+                for doc in input_from_node["documents"]:
+                    document_map[doc.id] = doc
+        elif self.join_mode == "merge":
+            document_map = {}
+            if self.weights:
+                weights = self.weights
+            else:
+                weights = [1/len(inputs)] * len(inputs)
+            for (input_from_node, _), weight in zip(inputs, weights):
+                for doc in input_from_node["documents"]:
+                    if document_map.get(doc.id):  # document already exists; update score
+                        document_map[doc.id].score += doc.score * weight
+                    else:  # add the document in map
+                        document_map[doc.id] = deepcopy(doc)
+                        document_map[doc.id].score *= weight
+        else:
+            raise Exception(f"Invalid join_mode: {self.join_mode}")
+
+        documents = sorted(document_map.values(), key=lambda d: d.score, reverse=True)
+        if self.top_k:
+            documents = documents[: self.top_k]
+        output = {"query": inputs[0][0]["query"], "documents": documents}
         return output, "output_1"

{"golden_diff": "diff --git a/haystack/pipeline.py b/haystack/pipeline.py\n--- a/haystack/pipeline.py\n+++ b/haystack/pipeline.py\n@@ -1,3 +1,4 @@\n+from copy import deepcopy\n from pathlib import Path\n from typing import List, Optional, Dict\n \n@@ -280,19 +281,63 @@\n \n \n class JoinDocuments:\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+    * 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+    \"\"\"\n+\n     outgoing_edges = 1\n \n-    def __init__(self, join_mode=\"concatenate\"):\n-        pass\n+    def __init__(\n+        self, join_mode: str = \"concatenate\", weights: Optional[List[float]] = None, top_k_join: Optional[int] = None\n+    ):\n+        \"\"\"\n+        :param join_mode: `concatenate` to combine documents from multiple retrievers or `merge` to aggregate scores of\n+                          individual documents.\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+        \"\"\"\n+        assert join_mode in [\"concatenate\", \"merge\"], 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+        self.join_mode = join_mode\n+        self.weights = weights\n+        self.top_k = top_k_join\n \n     def run(self, **kwargs):\n         inputs = kwargs[\"inputs\"]\n \n-        documents = []\n-        for i, _ in inputs:\n-            documents.extend(i[\"documents\"])\n-        output = {\n-            \"query\": inputs[0][0][\"query\"],\n-            \"documents\": documents\n-        }\n+        if self.join_mode == \"concatenate\":\n+            document_map = {}\n+            for input_from_node, _ in inputs:\n+                for doc in input_from_node[\"documents\"]:\n+                    document_map[doc.id] = doc\n+        elif self.join_mode == \"merge\":\n+            document_map = {}\n+            if self.weights:\n+                weights = self.weights\n+            else:\n+                weights = [1/len(inputs)] * len(inputs)\n+            for (input_from_node, _), weight in zip(inputs, weights):\n+                for doc in input_from_node[\"documents\"]:\n+                    if document_map.get(doc.id):  # document already exists; update score\n+                        document_map[doc.id].score += doc.score * weight\n+                    else:  # add the document in map\n+                        document_map[doc.id] = deepcopy(doc)\n+                        document_map[doc.id].score *= weight\n+        else:\n+            raise Exception(f\"Invalid join_mode: {self.join_mode}\")\n+\n+        documents = sorted(document_map.values(), key=lambda d: d.score, reverse=True)\n+        if self.top_k:\n+            documents = documents[: self.top_k]\n+        output = {\"query\": inputs[0][0][\"query\"], \"documents\": documents}\n         return output, \"output_1\"\n", "issue": "Combination of multiple retrievers\nOriginating from a discussion in #104. \n\n**What**\nCombine multiple scores during retrieval (e.g. embeddings similarity of multiple text fields, or BM25 + Embedding similarity ...) \n\n**Options**\n\na) Concatenation of Retrievers: Make the Finder accept a list of Retrievers and combine their scores (incl. weights) before feeding results to the Reader\nPro: most generic solution also allowing combination of BM25 + EmbeddingRetriever\nCon: Two separate retriever queries (suboptimal efficiency in some cases)\n\nb) New MultiEmbeddingRetriever: Extend the EmbeddingRetriever to the case of multiple Embeddings\nPro: simple implementation without side effects on the regular user interface\nCon: won't allow the combination of other retriever methods (e.g. ElasticsearchRetriever + EmbeddingRetriver)\n\nI am leaning towards a), but could see b) as a simpler short-term solution for the original use case discussed in #104 .\n\n\n\n\u2506Issue is synchronized with this [Jira Task](https://deepset-ai.atlassian.net/browse/HS-42) by [Unito](https://www.unito.io/learn-more)\n\n", "before_files": [{"content": "from pathlib import Path\nfrom typing import List, Optional, Dict\n\nimport networkx as nx\nfrom networkx import DiGraph\nfrom networkx.drawing.nx_agraph import to_agraph\n\nfrom haystack.generator.base import BaseGenerator\nfrom haystack.reader.base import BaseReader\nfrom haystack.retriever.base import BaseRetriever\n\n\nclass Pipeline:\n    \"\"\"\n    Pipeline brings together building blocks to build a complex search pipeline with Haystack & user-defined components.\n\n    Under-the-hood, a pipeline is represented as a directed acyclic graph of component nodes. It enables custom query\n    flows with options to branch queries(eg, extractive qa vs keyword match query), merge candidate documents for a\n    Reader from multiple Retrievers, or re-ranking of candidate documents.\n    \"\"\"\n\n    def __init__(self):\n        self.graph = DiGraph()\n        self.root_node_id = \"Query\"\n        self.graph.add_node(\"Query\", component=QueryNode())\n\n    def add_node(self, component, name: str, inputs: List[str]):\n        \"\"\"\n        Add a new node to the pipeline.\n\n        :param component: The object to be called when the data is passed to the node. It can be a Haystack component\n                          (like Retriever, Reader, or Generator) or a user-defined object that implements a run()\n                          method to process incoming data from predecessor node.\n        :param name: The name for the node. It must not contain any dots.\n        :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name\n                       of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single\n                       edge with a list of documents. It can be represented as [\"ElasticsearchRetriever\"].\n\n                       In cases when the predecessor node has multiple outputs, e.g., a \"QueryClassifier\", the output\n                       must be specified explicitly as \"QueryClassifier.output_2\".\n        \"\"\"\n        self.graph.add_node(name, component=component)\n\n        for i in inputs:\n            if \".\" in i:\n                [input_node_name, input_edge_name] = i.split(\".\")\n                assert \"output_\" in input_edge_name, f\"'{input_edge_name}' is not a valid edge name.\"\n                outgoing_edges_input_node = self.graph.nodes[input_node_name][\"component\"].outgoing_edges\n                assert int(input_edge_name.split(\"_\")[1]) <= outgoing_edges_input_node, (\n                    f\"Cannot connect '{input_edge_name}' from '{input_node_name}' as it only has \"\n                    f\"{outgoing_edges_input_node} outgoing edge(s).\"\n                )\n            else:\n                outgoing_edges_input_node = self.graph.nodes[i][\"component\"].outgoing_edges\n                assert outgoing_edges_input_node == 1, (\n                    f\"Adding an edge from {i} to {name} is ambiguous as {i} has {outgoing_edges_input_node} edges. \"\n                    f\"Please specify the output explicitly.\"\n                )\n                input_node_name = i\n                input_edge_name = \"output_1\"\n            self.graph.add_edge(input_node_name, name, label=input_edge_name)\n\n    def get_node(self, name: str):\n        \"\"\"\n        Get a node from the Pipeline.\n\n        :param name: The name of the node.\n        \"\"\"\n        component = self.graph.nodes[name][\"component\"]\n        return component\n\n    def set_node(self, name: str, component):\n        \"\"\"\n        Set the component for a node in the Pipeline.\n\n        :param name: The name of the node.\n        :param component: The component object to be set at the node.\n        \"\"\"\n        self.graph.nodes[name][\"component\"] = component\n\n    def run(self, **kwargs):\n        has_next_node = True\n        current_node_id = self.root_node_id\n        input_dict = kwargs\n        output_dict = None\n\n        while has_next_node:\n            output_dict, stream_id = self.graph.nodes[current_node_id][\"component\"].run(**input_dict)\n            input_dict = output_dict\n            next_nodes = self._get_next_nodes(current_node_id, stream_id)\n\n            if len(next_nodes) > 1:\n                join_node_id = list(nx.neighbors(self.graph, next_nodes[0]))[0]\n                if set(self.graph.predecessors(join_node_id)) != set(next_nodes):\n                    raise NotImplementedError(\n                        \"The current pipeline does not support multiple levels of parallel nodes.\"\n                    )\n                inputs_for_join_node = {\"inputs\": []}\n                for n_id in next_nodes:\n                    output = self.graph.nodes[n_id][\"component\"].run(**input_dict)\n                    inputs_for_join_node[\"inputs\"].append(output)\n                input_dict = inputs_for_join_node\n                current_node_id = join_node_id\n            elif len(next_nodes) == 1:\n                current_node_id = next_nodes[0]\n            else:\n                has_next_node = False\n\n        return output_dict\n\n    def _get_next_nodes(self, node_id: str, stream_id: str):\n        current_node_edges = self.graph.edges(node_id, data=True)\n        next_nodes = [\n            next_node\n            for _, next_node, data in current_node_edges\n            if not stream_id or data[\"label\"] == stream_id\n        ]\n        return next_nodes\n\n    def draw(self, path: Path = Path(\"pipeline.png\")):\n        \"\"\"\n        Create a Graphviz visualization of the pipeline.\n\n        :param path: the path to save the image.\n        \"\"\"\n        try:\n            import pygraphviz\n        except ImportError:\n            raise ImportError(f\"Could not import `pygraphviz`. Please install via: \\n\"\n                              f\"pip install pygraphviz\\n\"\n                              f\"(You might need to run this first: apt install libgraphviz-dev graphviz )\")\n\n        graphviz = to_agraph(self.graph)\n        graphviz.layout(\"dot\")\n        graphviz.draw(path)\n\n\nclass BaseStandardPipeline:\n    pipeline: Pipeline\n\n    def add_node(self, component, name: str, inputs: List[str]):\n        \"\"\"\n        Add a new node to the pipeline.\n\n        :param component: The object to be called when the data is passed to the node. It can be a Haystack component\n                          (like Retriever, Reader, or Generator) or a user-defined object that implements a run()\n                          method to process incoming data from predecessor node.\n        :param name: The name for the node. It must not contain any dots.\n        :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name\n                       of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single\n                       edge with a list of documents. It can be represented as [\"ElasticsearchRetriever\"].\n\n                       In cases when the predecessor node has multiple outputs, e.g., a \"QueryClassifier\", the output\n                       must be specified explicitly as \"QueryClassifier.output_2\".\n        \"\"\"\n\n        self.pipeline.add_node(component=component, name=name, inputs=inputs)\n\n    def get_node(self, name: str):\n        \"\"\"\n        Get a node from the Pipeline.\n\n        :param name: The name of the node.\n        \"\"\"\n        component = self.pipeline.get_node(name)\n        return component\n\n    def set_node(self, name: str, component):\n        \"\"\"\n        Set the component for a node in the Pipeline.\n\n        :param name: The name of the node.\n        :param component: The component object to be set at the node.\n        \"\"\"\n        self.pipeline.set_node(name, component)\n\n    def draw(self, path: Path = Path(\"pipeline.png\")):\n        \"\"\"\n        Create a Graphviz visualization of the pipeline.\n\n        :param path: the path to save the image.\n        \"\"\"\n        self.pipeline.draw(path)\n\n\nclass ExtractiveQAPipeline(BaseStandardPipeline):\n    def __init__(self, reader: BaseReader, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for Extractive Question Answering.\n\n        :param reader: Reader instance\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n        self.pipeline.add_node(component=reader, name=\"Reader\", inputs=[\"Retriever\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_reader: int = 10):\n        output = self.pipeline.run(\n            query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_reader=top_k_reader\n        )\n        return output\n\n\nclass DocumentSearchPipeline(BaseStandardPipeline):\n    def __init__(self, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for semantic document search.\n\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):\n        output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)\n        document_dicts = [doc.to_dict() for doc in output[\"documents\"]]\n        output[\"documents\"] = document_dicts\n        return output\n\n\nclass GenerativeQAPipeline(BaseStandardPipeline):\n    def __init__(self, generator: BaseGenerator, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for Generative Question Answering.\n\n        :param generator: Generator instance\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n        self.pipeline.add_node(component=generator, name=\"Generator\", inputs=[\"Retriever\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_generator: int = 10):\n        output = self.pipeline.run(\n            query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_generator=top_k_generator\n        )\n        return output\n\n\nclass FAQPipeline(BaseStandardPipeline):\n    def __init__(self, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for finding similar FAQs using semantic document search.\n\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):\n        output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)\n        documents = output[\"documents\"]\n\n        results: Dict = {\"query\": query, \"answers\": []}\n        for doc in documents:\n            # TODO proper calibratation of pseudo probabilities\n            cur_answer = {\n                \"query\": doc.text,\n                \"answer\": doc.meta[\"answer\"],\n                \"document_id\": doc.id,\n                \"context\": doc.meta[\"answer\"],\n                \"score\": doc.score,\n                \"probability\": doc.probability,\n                \"offset_start\": 0,\n                \"offset_end\": len(doc.meta[\"answer\"]),\n                \"meta\": doc.meta,\n            }\n\n            results[\"answers\"].append(cur_answer)\n\n        return results\n\n\nclass QueryNode:\n    outgoing_edges = 1\n\n    def run(self, **kwargs):\n        return kwargs, \"output_1\"\n\n\nclass JoinDocuments:\n    outgoing_edges = 1\n\n    def __init__(self, join_mode=\"concatenate\"):\n        pass\n\n    def run(self, **kwargs):\n        inputs = kwargs[\"inputs\"]\n\n        documents = []\n        for i, _ in inputs:\n            documents.extend(i[\"documents\"])\n        output = {\n            \"query\": inputs[0][0][\"query\"],\n            \"documents\": documents\n        }\n        return output, \"output_1\"\n", "path": "haystack/pipeline.py"}], "after_files": [{"content": "from copy import deepcopy\nfrom pathlib import Path\nfrom typing import List, Optional, Dict\n\nimport networkx as nx\nfrom networkx import DiGraph\nfrom networkx.drawing.nx_agraph import to_agraph\n\nfrom haystack.generator.base import BaseGenerator\nfrom haystack.reader.base import BaseReader\nfrom haystack.retriever.base import BaseRetriever\n\n\nclass Pipeline:\n    \"\"\"\n    Pipeline brings together building blocks to build a complex search pipeline with Haystack & user-defined components.\n\n    Under-the-hood, a pipeline is represented as a directed acyclic graph of component nodes. It enables custom query\n    flows with options to branch queries(eg, extractive qa vs keyword match query), merge candidate documents for a\n    Reader from multiple Retrievers, or re-ranking of candidate documents.\n    \"\"\"\n\n    def __init__(self):\n        self.graph = DiGraph()\n        self.root_node_id = \"Query\"\n        self.graph.add_node(\"Query\", component=QueryNode())\n\n    def add_node(self, component, name: str, inputs: List[str]):\n        \"\"\"\n        Add a new node to the pipeline.\n\n        :param component: The object to be called when the data is passed to the node. It can be a Haystack component\n                          (like Retriever, Reader, or Generator) or a user-defined object that implements a run()\n                          method to process incoming data from predecessor node.\n        :param name: The name for the node. It must not contain any dots.\n        :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name\n                       of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single\n                       edge with a list of documents. It can be represented as [\"ElasticsearchRetriever\"].\n\n                       In cases when the predecessor node has multiple outputs, e.g., a \"QueryClassifier\", the output\n                       must be specified explicitly as \"QueryClassifier.output_2\".\n        \"\"\"\n        self.graph.add_node(name, component=component)\n\n        for i in inputs:\n            if \".\" in i:\n                [input_node_name, input_edge_name] = i.split(\".\")\n                assert \"output_\" in input_edge_name, f\"'{input_edge_name}' is not a valid edge name.\"\n                outgoing_edges_input_node = self.graph.nodes[input_node_name][\"component\"].outgoing_edges\n                assert int(input_edge_name.split(\"_\")[1]) <= outgoing_edges_input_node, (\n                    f\"Cannot connect '{input_edge_name}' from '{input_node_name}' as it only has \"\n                    f\"{outgoing_edges_input_node} outgoing edge(s).\"\n                )\n            else:\n                outgoing_edges_input_node = self.graph.nodes[i][\"component\"].outgoing_edges\n                assert outgoing_edges_input_node == 1, (\n                    f\"Adding an edge from {i} to {name} is ambiguous as {i} has {outgoing_edges_input_node} edges. \"\n                    f\"Please specify the output explicitly.\"\n                )\n                input_node_name = i\n                input_edge_name = \"output_1\"\n            self.graph.add_edge(input_node_name, name, label=input_edge_name)\n\n    def get_node(self, name: str):\n        \"\"\"\n        Get a node from the Pipeline.\n\n        :param name: The name of the node.\n        \"\"\"\n        component = self.graph.nodes[name][\"component\"]\n        return component\n\n    def set_node(self, name: str, component):\n        \"\"\"\n        Set the component for a node in the Pipeline.\n\n        :param name: The name of the node.\n        :param component: The component object to be set at the node.\n        \"\"\"\n        self.graph.nodes[name][\"component\"] = component\n\n    def run(self, **kwargs):\n        has_next_node = True\n        current_node_id = self.root_node_id\n        input_dict = kwargs\n        output_dict = None\n\n        while has_next_node:\n            output_dict, stream_id = self.graph.nodes[current_node_id][\"component\"].run(**input_dict)\n            input_dict = output_dict\n            next_nodes = self._get_next_nodes(current_node_id, stream_id)\n\n            if len(next_nodes) > 1:\n                join_node_id = list(nx.neighbors(self.graph, next_nodes[0]))[0]\n                if set(self.graph.predecessors(join_node_id)) != set(next_nodes):\n                    raise NotImplementedError(\n                        \"The current pipeline does not support multiple levels of parallel nodes.\"\n                    )\n                inputs_for_join_node = {\"inputs\": []}\n                for n_id in next_nodes:\n                    output = self.graph.nodes[n_id][\"component\"].run(**input_dict)\n                    inputs_for_join_node[\"inputs\"].append(output)\n                input_dict = inputs_for_join_node\n                current_node_id = join_node_id\n            elif len(next_nodes) == 1:\n                current_node_id = next_nodes[0]\n            else:\n                has_next_node = False\n\n        return output_dict\n\n    def _get_next_nodes(self, node_id: str, stream_id: str):\n        current_node_edges = self.graph.edges(node_id, data=True)\n        next_nodes = [\n            next_node\n            for _, next_node, data in current_node_edges\n            if not stream_id or data[\"label\"] == stream_id\n        ]\n        return next_nodes\n\n    def draw(self, path: Path = Path(\"pipeline.png\")):\n        \"\"\"\n        Create a Graphviz visualization of the pipeline.\n\n        :param path: the path to save the image.\n        \"\"\"\n        try:\n            import pygraphviz\n        except ImportError:\n            raise ImportError(f\"Could not import `pygraphviz`. Please install via: \\n\"\n                              f\"pip install pygraphviz\\n\"\n                              f\"(You might need to run this first: apt install libgraphviz-dev graphviz )\")\n\n        graphviz = to_agraph(self.graph)\n        graphviz.layout(\"dot\")\n        graphviz.draw(path)\n\n\nclass BaseStandardPipeline:\n    pipeline: Pipeline\n\n    def add_node(self, component, name: str, inputs: List[str]):\n        \"\"\"\n        Add a new node to the pipeline.\n\n        :param component: The object to be called when the data is passed to the node. It can be a Haystack component\n                          (like Retriever, Reader, or Generator) or a user-defined object that implements a run()\n                          method to process incoming data from predecessor node.\n        :param name: The name for the node. It must not contain any dots.\n        :param inputs: A list of inputs to the node. If the predecessor node has a single outgoing edge, just the name\n                       of node is sufficient. For instance, a 'ElasticsearchRetriever' node would always output a single\n                       edge with a list of documents. It can be represented as [\"ElasticsearchRetriever\"].\n\n                       In cases when the predecessor node has multiple outputs, e.g., a \"QueryClassifier\", the output\n                       must be specified explicitly as \"QueryClassifier.output_2\".\n        \"\"\"\n\n        self.pipeline.add_node(component=component, name=name, inputs=inputs)\n\n    def get_node(self, name: str):\n        \"\"\"\n        Get a node from the Pipeline.\n\n        :param name: The name of the node.\n        \"\"\"\n        component = self.pipeline.get_node(name)\n        return component\n\n    def set_node(self, name: str, component):\n        \"\"\"\n        Set the component for a node in the Pipeline.\n\n        :param name: The name of the node.\n        :param component: The component object to be set at the node.\n        \"\"\"\n        self.pipeline.set_node(name, component)\n\n    def draw(self, path: Path = Path(\"pipeline.png\")):\n        \"\"\"\n        Create a Graphviz visualization of the pipeline.\n\n        :param path: the path to save the image.\n        \"\"\"\n        self.pipeline.draw(path)\n\n\nclass ExtractiveQAPipeline(BaseStandardPipeline):\n    def __init__(self, reader: BaseReader, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for Extractive Question Answering.\n\n        :param reader: Reader instance\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n        self.pipeline.add_node(component=reader, name=\"Reader\", inputs=[\"Retriever\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_reader: int = 10):\n        output = self.pipeline.run(\n            query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_reader=top_k_reader\n        )\n        return output\n\n\nclass DocumentSearchPipeline(BaseStandardPipeline):\n    def __init__(self, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for semantic document search.\n\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):\n        output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)\n        document_dicts = [doc.to_dict() for doc in output[\"documents\"]]\n        output[\"documents\"] = document_dicts\n        return output\n\n\nclass GenerativeQAPipeline(BaseStandardPipeline):\n    def __init__(self, generator: BaseGenerator, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for Generative Question Answering.\n\n        :param generator: Generator instance\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n        self.pipeline.add_node(component=generator, name=\"Generator\", inputs=[\"Retriever\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10, top_k_generator: int = 10):\n        output = self.pipeline.run(\n            query=query, filters=filters, top_k_retriever=top_k_retriever, top_k_generator=top_k_generator\n        )\n        return output\n\n\nclass FAQPipeline(BaseStandardPipeline):\n    def __init__(self, retriever: BaseRetriever):\n        \"\"\"\n        Initialize a Pipeline for finding similar FAQs using semantic document search.\n\n        :param retriever: Retriever instance\n        \"\"\"\n        self.pipeline = Pipeline()\n        self.pipeline.add_node(component=retriever, name=\"Retriever\", inputs=[\"Query\"])\n\n    def run(self, query: str, filters: Optional[Dict] = None, top_k_retriever: int = 10):\n        output = self.pipeline.run(query=query, filters=filters, top_k_retriever=top_k_retriever)\n        documents = output[\"documents\"]\n\n        results: Dict = {\"query\": query, \"answers\": []}\n        for doc in documents:\n            # TODO proper calibratation of pseudo probabilities\n            cur_answer = {\n                \"query\": doc.text,\n                \"answer\": doc.meta[\"answer\"],\n                \"document_id\": doc.id,\n                \"context\": doc.meta[\"answer\"],\n                \"score\": doc.score,\n                \"probability\": doc.probability,\n                \"offset_start\": 0,\n                \"offset_end\": len(doc.meta[\"answer\"]),\n                \"meta\": doc.meta,\n            }\n\n            results[\"answers\"].append(cur_answer)\n\n        return results\n\n\nclass QueryNode:\n    outgoing_edges = 1\n\n    def run(self, **kwargs):\n        return kwargs, \"output_1\"\n\n\nclass JoinDocuments:\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    * 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    \"\"\"\n\n    outgoing_edges = 1\n\n    def __init__(\n        self, join_mode: str = \"concatenate\", weights: Optional[List[float]] = None, top_k_join: Optional[int] = None\n    ):\n        \"\"\"\n        :param join_mode: `concatenate` to combine documents from multiple retrievers or `merge` to aggregate scores of\n                          individual documents.\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        \"\"\"\n        assert join_mode in [\"concatenate\", \"merge\"], 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        self.join_mode = join_mode\n        self.weights = weights\n        self.top_k = top_k_join\n\n    def run(self, **kwargs):\n        inputs = kwargs[\"inputs\"]\n\n        if self.join_mode == \"concatenate\":\n            document_map = {}\n            for input_from_node, _ in inputs:\n                for doc in input_from_node[\"documents\"]:\n                    document_map[doc.id] = doc\n        elif self.join_mode == \"merge\":\n            document_map = {}\n            if self.weights:\n                weights = self.weights\n            else:\n                weights = [1/len(inputs)] * len(inputs)\n            for (input_from_node, _), weight in zip(inputs, weights):\n                for doc in input_from_node[\"documents\"]:\n                    if document_map.get(doc.id):  # document already exists; update score\n                        document_map[doc.id].score += doc.score * weight\n                    else:  # add the document in map\n                        document_map[doc.id] = deepcopy(doc)\n                        document_map[doc.id].score *= weight\n        else:\n            raise Exception(f\"Invalid join_mode: {self.join_mode}\")\n\n        documents = sorted(document_map.values(), key=lambda d: d.score, reverse=True)\n        if self.top_k:\n            documents = documents[: self.top_k]\n        output = {\"query\": inputs[0][0][\"query\"], \"documents\": documents}\n        return output, \"output_1\"\n", "path": "haystack/pipeline.py"}]}