Repository: deepset-ai/haystack
Pull request: Pipeline node names validation
Related to #1568

The `params` dictionary passed to `Pipelines.run()` does not check that a node corresponding to each node name exists. This means that a typo in the node name makes all the parameters to be silently discarded. It would be better to throw an error if no node with a given name could be found in the pipeline.

Note: this **could not be done in** `BasePipeline`, as it has no `run()` method. To make it available in `BasePipeline` (and be able to use it in Ray), then we should consider pushing the `run` method implementation to `BasePipeline`.

Problem statement:
Add validation on the nodes names in the `params` dictionary passed to `Pipeline.run()`
The `params` dictionary passed to `Pipelines.run()` now validates the key-value pairs passed for each node. However, it does not check that a node corresponding to each node name exists. This means that a typo in the node name makes all the parameters to be silently discarded. 

It would be better to throw an error if no node with a given name could be found in the pipeline.

Hints from issue discussion:
Let's implement it in BasePipeline so that we can leverage it in Ray as well

Commit message: Use roberta model for test_pipeline.yaml

Patch to review:
diff --git a/haystack/pipeline.py b/haystack/pipeline.py
index 9ea3832426..92726582a1 100644
--- a/haystack/pipeline.py
+++ b/haystack/pipeline.py
@@ -284,6 +284,12 @@ def run(  # type: ignore
             :param debug_logs: Whether all the logs of the node should be printed in the console,
                                regardless of their severity and of the existing logger's settings.
         """
+        # validate the node names
+        if params:
+            if not all(node in self.graph.nodes for node in params.keys()):
+                non_existing_node = set(params.keys()) - set(self.graph.nodes)
+                raise ValueError(f"No node(s) named {', '.join(non_existing_node)} found in pipeline. Defined nodes: {self.graph.nodes}")
+
         node_output = None
         queue = {
             self.root_node: {"root_node": self.root_node, "params": params}
diff --git a/rest_api/controller/file_upload.py b/rest_api/controller/file_upload.py
index 56735cba24..e77b6e2a2a 100644
--- a/rest_api/controller/file_upload.py
+++ b/rest_api/controller/file_upload.py
@@ -41,12 +41,16 @@
 
 
 @as_form
-class FileUploadParams(BaseModel):
+class FileConverterParams(BaseModel):    
     remove_numeric_tables: Optional[bool] = None
-    remove_whitespace: Optional[bool] = None
-    remove_empty_lines: Optional[bool] = None
-    remove_header_footer: Optional[bool] = None
     valid_languages: Optional[List[str]] = None
+
+
+@as_form
+class PreprocessorParams(BaseModel):
+    clean_whitespace: Optional[bool] = None
+    clean_empty_lines: Optional[bool] = None
+    clean_header_footer: Optional[bool] = None
     split_by: Optional[str] = None
     split_length: Optional[int] = None
     split_overlap: Optional[int] = None
@@ -61,7 +65,8 @@ class Response(BaseModel):
 def file_upload(
     files: List[UploadFile] = File(...),
     meta: Optional[str] = Form("null"),  # JSON serialized string
-    params: FileUploadParams = Depends(FileUploadParams.as_form)
+    fileconverter_params: FileConverterParams = Depends(FileConverterParams.as_form),
+    preprocessor_params: PreprocessorParams = Depends(PreprocessorParams.as_form)
 ):
     if not INDEXING_PIPELINE:
         raise HTTPException(status_code=501, detail="Indexing Pipeline is not configured.")
@@ -85,5 +90,9 @@ def file_upload(
     INDEXING_PIPELINE.run(
             file_paths=file_paths,
             meta=file_metas,
-            params=params.dict(),
+            params={
+                "TextFileConverter": fileconverter_params.dict(), 
+                "PDFFileConverter": fileconverter_params.dict(),
+                "Preprocessor": preprocessor_params.dict()
+            },
     )
diff --git a/rest_api/controller/search.py b/rest_api/controller/search.py
index 0e365d680a..7d57884a97 100644
--- a/rest_api/controller/search.py
+++ b/rest_api/controller/search.py
@@ -52,18 +52,21 @@ def query(request: QueryRequest):
 def _process_request(pipeline, request) -> QueryResponse:
     start_time = time.time()
 
-    params = request.params or {}
-    params["filters"] = params.get("filters") or {}
+    reader_params = request.reader_params or {}
+    retriever_params = request.retriever_params or {}
+    retriever_params["filters"] = retriever_params.get("filters") or {}
     filters = {}
-    if "filters" in params:  # put filter values into a list and remove filters with null value
-        for key, values in params["filters"].items():
+    if "filters" in retriever_params:  # put filter values into a list and remove filters with null value
+        for key, values in retriever_params["filters"].items():
             if values is None:
                 continue
             if not isinstance(values, list):
                 values = [values]
             filters[key] = values
-    params["filters"] = filters
-    result = pipeline.run(query=request.query, params=params)
+    retriever_params["filters"] = filters
+
+    result = pipeline.run(query=request.query, params={"Retriever": retriever_params, "Reader": reader_params})
+    
     end_time = time.time()
     logger.info({"request": request.dict(), "response": result, "time": f"{(end_time - start_time):.2f}"})
 
diff --git a/rest_api/schema.py b/rest_api/schema.py
index b974bcc58c..b276fc97d3 100644
--- a/rest_api/schema.py
+++ b/rest_api/schema.py
@@ -8,7 +8,8 @@
 
 class QueryRequest(BaseModel):
     query: str
-    params: Optional[dict] = None
+    retriever_params: Optional[dict] = None
+    reader_params: Optional[dict] = None
 
 
 class FilterRequest(BaseModel):
diff --git a/test/test_eval.py b/test/test_eval.py
index 1850a82dd0..b0952e1cc7 100644
--- a/test/test_eval.py
+++ b/test/test_eval.py
@@ -126,7 +126,7 @@ def test_eval_pipeline(document_store: BaseDocumentStore, reader, retriever):
         res = p.run(
             query=l.query,
             labels=l,
-            params={"index": "haystack_test_eval_document"}
+            params={"ESRetriever":{"index": "haystack_test_eval_document"}}
         )
     assert eval_retriever.recall == 1.0
     assert round(eval_reader.top_k_f1, 4) == 0.8333
diff --git a/test/test_faiss_and_milvus.py b/test/test_faiss_and_milvus.py
index daf3ca2834..bd8406015d 100644
--- a/test/test_faiss_and_milvus.py
+++ b/test/test_faiss_and_milvus.py
@@ -189,7 +189,7 @@ def test_finding(document_store, retriever):
     document_store.write_documents(DOCUMENTS)
     pipe = DocumentSearchPipeline(retriever=retriever)
 
-    prediction = pipe.run(query="How to test this?", params={"top_k": 1})
+    prediction = pipe.run(query="How to test this?", params={"Retriever": {"top_k": 1}})
 
     assert len(prediction.get('documents', [])) == 1
 
@@ -222,7 +222,7 @@ def test_pipeline(document_store, retriever):
     document_store.write_documents(documents)
     pipeline = Pipeline()
     pipeline.add_node(component=retriever, name="FAISS", inputs=["Query"])
-    output = pipeline.run(query="How to test this?", params={"top_k": 3})
+    output = pipeline.run(query="How to test this?", params={"FAISS": {"top_k": 3}})
     assert len(output["documents"]) == 3
 
 
diff --git a/test/test_pipeline.py b/test/test_pipeline.py
index bedc12699e..4a546896aa 100644
--- a/test/test_pipeline.py
+++ b/test/test_pipeline.py
@@ -29,15 +29,14 @@ def test_load_and_save_yaml(document_store, tmp_path):
         Path(__file__).parent/"samples"/"pipeline"/"test_pipeline.yaml", pipeline_name="indexing_pipeline"
     )
     pipeline.run(
-        file_paths=Path(__file__).parent/"samples"/"pdf"/"sample_pdf_1.pdf",
-        params={"Retriever": {"top_k": 10}, "Reader": {"top_k": 3}},
+        file_paths=Path(__file__).parent/"samples"/"pdf"/"sample_pdf_1.pdf"
     )
     # test correct load of query pipeline from yaml
     pipeline = Pipeline.load_from_yaml(
         Path(__file__).parent/"samples"/"pipeline"/"test_pipeline.yaml", pipeline_name="query_pipeline"
     )
     prediction = pipeline.run(
-        query="Who made the PDF specification?", params={"Retriever": {"top_k": 10}, "Reader": {"top_k": 3}}
+        query="Who made the PDF specification?", params={"ESRetriever": {"top_k": 10}, "Reader": {"top_k": 3}}
     )
     assert prediction["query"] == "Who made the PDF specification?"
     assert prediction["answers"][0].answer == "Adobe Systems"
@@ -85,12 +84,46 @@ def test_load_and_save_yaml(document_store, tmp_path):
     ).replace("\n", "")
 
 
+@pytest.mark.elasticsearch
+@pytest.mark.parametrize("document_store_with_docs", ["elasticsearch"], indirect=True)
+def test_node_names_validation(document_store_with_docs, tmp_path):
+    pipeline = Pipeline()
+    pipeline.add_node(
+        component=ElasticsearchRetriever(document_store=document_store_with_docs), 
+        name="Retriever", 
+        inputs=["Query"])
+    pipeline.add_node(
+        component=FARMReader(model_name_or_path="deepset/minilm-uncased-squad2"), 
+        name="Reader", 
+        inputs=["Retriever"])
+
+    with pytest.raises(ValueError) as exc_info:
+        pipeline.run(
+            query="Who lives in Berlin?",
+            params={
+                "Reader": {"top_k": 3}, 
+                "non-existing-node": {"top_k": 10}, 
+                "non-existing-node-2": {"random_param": "rnd"}
+            },
+            debug=True,
+            debug_logs=True
+        )
+    # The error message looks like "Node X not found. Defined nodes: A, B".
+    # This asserts make sure that X comes before "Defined nodes", 
+    # while A comes after it.
+    exception_raised = str(exc_info.value)
+    delimiter = exception_raised.index("Defined nodes")
+    assert exception_raised.index("non-existing-node") < delimiter
+    assert exception_raised.index("non-existing-node-2") < delimiter
+    assert exception_raised.index("Reader") > delimiter
+
+
 @pytest.mark.elasticsearch
 @pytest.mark.parametrize("document_store_with_docs", ["elasticsearch"], indirect=True)
 def test_debug_attributes_global(document_store_with_docs, tmp_path):
 
     es_retriever = ElasticsearchRetriever(document_store=document_store_with_docs)
-    reader = FARMReader(model_name_or_path="deepset/roberta-base-squad2")
+    reader = FARMReader(model_name_or_path="deepset/minilm-uncased-squad2")
 
     pipeline = Pipeline()
     pipeline.add_node(component=es_retriever, name="ESRetriever", inputs=["Query"])
@@ -122,7 +155,7 @@ def test_debug_attributes_global(document_store_with_docs, tmp_path):
 def test_debug_attributes_per_node(document_store_with_docs, tmp_path):
 
     es_retriever = ElasticsearchRetriever(document_store=document_store_with_docs)
-    reader = FARMReader(model_name_or_path="deepset/roberta-base-squad2")
+    reader = FARMReader(model_name_or_path="deepset/minilm-uncased-squad2")
 
     pipeline = Pipeline()
     pipeline.add_node(component=es_retriever, name="ESRetriever", inputs=["Query"])
@@ -152,7 +185,7 @@ def test_debug_attributes_per_node(document_store_with_docs, tmp_path):
 def test_global_debug_attributes_override_node_ones(document_store_with_docs, tmp_path):
 
     es_retriever = ElasticsearchRetriever(document_store=document_store_with_docs)
-    reader = FARMReader(model_name_or_path="deepset/roberta-base-squad2")
+    reader = FARMReader(model_name_or_path="deepset/minilm-uncased-squad2")
 
     pipeline = Pipeline()
     pipeline.add_node(component=es_retriever, name="ESRetriever", inputs=["Query"])
@@ -296,13 +329,13 @@ def test_faq_pipeline(retriever, document_store):
 
     pipeline = FAQPipeline(retriever=retriever)
 
-    output = pipeline.run(query="How to test this?", params={"top_k": 3})
+    output = pipeline.run(query="How to test this?", params={"Retriever": {"top_k": 3}})
     assert len(output["answers"]) == 3
     assert output["query"].startswith("How to")
     assert output["answers"][0].answer.startswith("Using tests")
 
     if isinstance(document_store, ElasticsearchDocumentStore):
-        output = pipeline.run(query="How to test this?", params={"filters": {"source": ["wiki2"]}, "top_k": 5})
+        output = pipeline.run(query="How to test this?", params={"Retriever": {"filters": {"source": ["wiki2"]}, "top_k": 5}})
         assert len(output["answers"]) == 1
 
 
diff --git a/test/test_pipeline_extractive_qa.py b/test/test_pipeline_extractive_qa.py
index ff5cced4a9..48910f70fe 100644
--- a/test/test_pipeline_extractive_qa.py
+++ b/test/test_pipeline_extractive_qa.py
@@ -71,7 +71,7 @@ def test_extractive_qa_offsets(reader, retriever_with_docs):
 def test_extractive_qa_answers_single_result(reader, retriever_with_docs):
     pipeline = ExtractiveQAPipeline(reader=reader, retriever=retriever_with_docs)
     query = "testing finder"
-    prediction = pipeline.run(query=query, params={"top_k": 1})
+    prediction = pipeline.run(query=query, params={"Retriever": {"top_k": 1}, "Reader": {"top_k": 1}})
     assert prediction is not None
     assert len(prediction["answers"]) == 1
 
diff --git a/test/test_rest_api.py b/test/test_rest_api.py
index 4df1a092fa..d322413796 100644
--- a/test/test_rest_api.py
+++ b/test/test_rest_api.py
@@ -133,6 +133,7 @@ def test_delete_documents():
     response_json = response.json()
     assert len(response_json) == 1
 
+
 def test_file_upload(client: TestClient):
     file_to_upload = {'files': (Path(__file__).parent / "samples"/"pdf"/"sample_pdf_1.pdf").open('rb')}
     response = client.post(url="/file-upload", files=file_to_upload, data={"meta": '{"meta_key": "meta_value"}'})
@@ -149,7 +150,7 @@ def test_query_with_no_filter(populated_client: TestClient):
 
 
 def test_query_with_one_filter(populated_client: TestClient):
-    query_with_filter = {"query": "Who made the PDF specification?", "params": {"filters": {"meta_key": "meta_value"}}}
+    query_with_filter = {"query": "Who made the PDF specification?", "retriever_params": {"filters": {"meta_key": "meta_value"}}}
     response = populated_client.post(url="/query", json=query_with_filter)
     assert 200 == response.status_code
     response_json = response.json()
@@ -159,7 +160,7 @@ def test_query_with_one_filter(populated_client: TestClient):
 def test_query_with_filter_list(populated_client: TestClient):
     query_with_filter_list = {
         "query": "Who made the PDF specification?",
-        "params": {"filters": {"meta_key": ["meta_value", "another_value"]}}
+        "retriever_params": {"filters": {"meta_key": ["meta_value", "another_value"]}}
     }
     response = populated_client.post(url="/query", json=query_with_filter_list)
     assert 200 == response.status_code
@@ -169,7 +170,7 @@ def test_query_with_filter_list(populated_client: TestClient):
 
 def test_query_with_invalid_filter(populated_client: TestClient):
     query_with_invalid_filter = {
-        "query": "Who made the PDF specification?", "params": {"filters": {"meta_key": "invalid_value"}}
+        "query": "Who made the PDF specification?", "retriever_params": {"filters": {"meta_key": "invalid_value"}}
     }
     response = populated_client.post(url="/query", json=query_with_invalid_filter)
     assert 200 == response.status_code
diff --git a/test/test_summarizer.py b/test/test_summarizer.py
index 698131c173..36e8d0a453 100644
--- a/test/test_summarizer.py
+++ b/test/test_summarizer.py
@@ -67,7 +67,7 @@ def test_summarization_pipeline(document_store, retriever, summarizer):
 
     query = "Where is Eiffel Tower?"
     pipeline = SearchSummarizationPipeline(retriever=retriever, summarizer=summarizer, return_in_answer_format=True)
-    output = pipeline.run(query=query, params={"top_k": 1})
+    output = pipeline.run(query=query, params={"Retriever": {"top_k": 1}})
     answers = output["answers"]
     assert len(answers) == 1
     assert "The Eiffel Tower is a landmark in Paris, France." == answers[0]["answer"]
@@ -88,7 +88,7 @@ def test_summarization_pipeline_one_summary(document_store, retriever, summarize
 
     query = "Where is Eiffel Tower?"
     pipeline = SearchSummarizationPipeline(retriever=retriever, summarizer=summarizer, return_in_answer_format=True)
-    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}, "generate_single_summary": True})
+    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}, "Summarizer": {"generate_single_summary": True}})
     answers = output["answers"]
     assert len(answers) == 1
     assert answers[0]["answer"] in EXPECTED_ONE_SUMMARIES
@@ -122,7 +122,7 @@ def test_summarization_pipeline_with_translator(
         output_translator=en_to_de_translator,
         pipeline=base_pipeline
     )
-    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}, "generate_single_summary": True})
+    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}, "Summarizer": {"generate_single_summary": True}})
     # SearchSummarizationPipeline return answers but Summarizer return documents
     documents = output["documents"]
     assert len(documents) == 1


Code review comments:
diff_hunk: @@ -8,7 +8,8 @@
 
 class QueryRequest(BaseModel):
     query: str
-    params: Optional[dict] = None
+    retriever_params: Optional[dict] = None
path: rest_api/schema.py
text: @user1:
I see that this cleans up the code a bit but wondering if we would like to restrict the endpoint to an "extractive-qa" pipeline? With untyped generic `params`, this endpoint can be re-used for other pipelines.

@author:
On this I totally agree. If we want to reuse `QueryRequest` for different types of pipeline, I can also think of another solution: ask the user to provide the parameters by node name, ready to be given straight to `Pipeline.run()`. This way there would be no need for the two fields. What do you think?

@user1:
Yes, that's a good idea!