Repository: deepset-ai/haystack
Pull request: feat: Add `CsvTextConverter`
Fixes #3550, allow user to build full FAQ using YAML pipeline description and with CSV import and indexing.

### Related Issues
- fixes #3550

### Proposed Changes:
I've added ~2~1 node to be able to build a full FAQ pipeline with CSV import :

- **Csv2Documents** : Takes a file input, parse it an FAQ CSV file to output Documents.
- ~**EmbedDocuments** : Build using and embeder (actually a Retriever), takes Documents as input and run the `embeded.run_indexing`on it.~

And Also a sample of FAQ pipeline.

### How did you test it?
Run haystack with `pipelines.haystack-pipeline.faq.yml` then import the [covid FAQ CSV](https://s3.eu-central-1.amazonaws.com/deepset.ai-farm-qa/datasets/documents/small_faq_covid.csv.zip).

### Notes for the reviewer
I noticed this issue about splitting Embedder and  Retriever concept, I've mentioned it in a comment : #2403

If you want I can also update the tutorial "[Utilizing existing FAQs for Question Answering](https://haystack.deepset.ai/tutorials/04_faq_style_qa)" to mention the YAML pipeline sample.

### Checklist
- [X] I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- [X] I have updated the related issue with new insights and changes
- [X] I added tests that demonstrate the correct behavior of the change
- [X] I've used the [conventional commit convention](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title
- [X] I documented my code
- [X] I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
FAQ CSV Indexing using YAML description, adding missing nodes
**Is your feature request related to a problem? Please describe.**
Not really.

**Describe the solution you'd like**
As an Haystack user I would like to be able to import FAQ (as CSV files), index them though API endpoint using an indexing pipeline described as YAML.

**Describe alternatives you've considered**
I've added 2 nodes to do so :

- **Csv2Documents** : Takes a file input, parse it an FAQ CSV file to output Documents.
- **EmbedDocuments** : Build using and embeder (actually a Retriever), takes Documents as input and run the `embeded.run_indexing`on it.

Those 2 nodes are really basic but seems necessary to then build a FAQ with CSV indexing pipeline. I will push it today.

**Additional context**
Index pipeline description looks like this : 
```yaml
# To allow your IDE to autocomplete and validate your YAML pipelines, name them as <name of your choice>.haystack-pipeline.yml

version: ignore

components:    # define all the building-blocks for Pipeline
  - name: DocumentStore
    type: ElasticsearchDocumentStore
    params:
      host: localhost
      embedding_field: question_emb
      embedding_dim: 384
      excluded_meta_data:
        - question_emb
      similarity: cosine
  - name: Retriever
    type: EmbeddingRetriever
    params:
      document_store: DocumentStore    # params can reference other components defined in the YAML
      embedding_model: sentence-transformers/all-MiniLM-L6-v2
      scale_score: False
  - name: Doc2Answers       # custom-name for the component; helpful for visualization & debugging
    type: Docs2Answers    # Haystack Class name for the component
  - name: CsvToDocs
    type: CsvToDocuments
  - name: EmbedDocs
    type: EmbedDocuments
    params:
      embeder: Retriever

pipelines:
  - name: query    # a sample extractive-qa Pipeline
    nodes:
      - name: Retriever
        inputs: [Query]
      - name: Doc2Answers
        inputs: [Retriever]
  - name: indexing
    nodes:
      - name: CsvToDocs
        inputs: [File]
      - name: EmbedDocuments
        inputs: [ CsvToDocs ]
      - name: DocumentStore
        inputs: [ EmbedDocuments ]

```

Hints from issue discussion:
Pushed 👍, thanks in advance for the review of #3587 
@vblagoje can you double check if this is somehow related to your proposal https://github.com/deepset-ai/haystack/pull/3558 ? In case, I would merge the efforts.
Hi @masci and @vblagoje  I took a look at the proposal, the focus here is in having the ability to import CSV file (though rest API), embed them and use it for FAQ pipeline without having to develop any node. Something that any one willing to use FAQ will certainly need.

It introduces the idea of having an "embedding / indexing node" that simply calls the `run_indexing(docs)` of a retriever, clearly here it's only used as an "embeder" what is called **encorder** in the proposal. It's highlighting the design issue mentioned here : #2403 and could benefit from having the **encoder** concept mentioned in the  proposal that will be used by this "embeding / indexing" node.


Thanks for this effort @Benvii , no @masci this is unrelated to https://github.com/deepset-ai/haystack/pull/3558
> Thanks for this effort @Benvii , no @masci this is unrelated to #3558

@masci I take back this statement because at the time of writing the comment above, our proposal  https://github.com/deepset-ai/haystack/pull/3558 didn't have the changes regarding separating Embedder and EmbeddingRetriever outlined in #2403 that @ZanSara suggested. Now that Sara's recommendations have been accepted into the [proposal](https://github.com/deepset-ai/haystack/commit/394b1f08c213a23a8d5082c47cd55967e1b4df54) - this issue is indeed related to #3558 

As a matter of fact, #3558 is a prerequisite for the implementation of this issue, and we should follow @ZanSara recommendations [outlined](https://github.com/deepset-ai/haystack/pull/3638#issuecomment-1330408336) 
@ZanSara should we then wait for https://github.com/deepset-ai/haystack/pull/3638 to be merged first before proceeding with the merge of this PR? 
@vblagoje yes I propose to follow the procedure, so wait another 3 days for any last-minute commenter on the proposal, merge the proposal, and then merge the implementation.
Deal @ZanSara

Commit message: black

Patch to review:
diff --git a/haystack/nodes/__init__.py b/haystack/nodes/__init__.py
index ec3e6126d7..facbb3ce73 100644
--- a/haystack/nodes/__init__.py
+++ b/haystack/nodes/__init__.py
@@ -19,6 +19,7 @@
     TextConverter,
     AzureConverter,
     ParsrConverter,
+    CsvTextConverter,
 )
 from haystack.nodes.label_generator import PseudoLabelGenerator
 from haystack.nodes.other import Docs2Answers, JoinDocuments, RouteDocuments, JoinAnswers, DocumentMerger
diff --git a/haystack/nodes/file_converter/__init__.py b/haystack/nodes/file_converter/__init__.py
index db0b714fc6..74f9fbb4fa 100644
--- a/haystack/nodes/file_converter/__init__.py
+++ b/haystack/nodes/file_converter/__init__.py
@@ -2,6 +2,7 @@
 
 from haystack.utils.import_utils import safe_import
 
+from haystack.nodes.file_converter.csv import CsvTextConverter
 from haystack.nodes.file_converter.docx import DocxToTextConverter
 from haystack.nodes.file_converter.tika import TikaConverter, TikaXHTMLParser
 from haystack.nodes.file_converter.txt import TextConverter
diff --git a/haystack/nodes/file_converter/csv.py b/haystack/nodes/file_converter/csv.py
new file mode 100644
index 0000000000..93c31e9d1d
--- /dev/null
+++ b/haystack/nodes/file_converter/csv.py
@@ -0,0 +1,54 @@
+from typing import Union, List, Optional, Any, Dict
+
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+from haystack import Document
+from haystack.nodes.file_converter import BaseConverter
+
+
+logger = logging.getLogger(__name__)
+
+
+class CsvTextConverter(BaseConverter):
+    """
+    Converts Question & Answers CSV files to text Documents.
+    """
+
+    outgoing_edges = 1
+
+    def convert(
+        self,
+        file_path: Union[Path, List[Path], str, List[str], List[Union[Path, str]]],
+        meta: Optional[Dict[str, Any]],
+        remove_numeric_tables: Optional[bool] = None,
+        valid_languages: Optional[List[str]] = None,
+        encoding: Optional[str] = "UTF-8",
+        id_hash_keys: Optional[List[str]] = None,
+    ) -> List[Document]:
+        """
+        Load CVS file and convert it to documents.
+
+        :param file_path: Path to a CSV file containing two columns.
+            The first will be interpreted as a question, the second as content.
+        :returns: List of document, 1 document per line in the CSV.
+        """
+        df = pd.read_csv(file_path, encoding=encoding)
+
+        df.fillna(value="", inplace=True)
+        df["question"] = df["question"].apply(lambda x: x.strip())
+
+        df = df.rename(columns={"question": "content"})
+        docs_dicts = df.to_dict(orient="records")
+
+        docs = []
+        for dictionary in docs_dicts:
+            if meta:
+                dictionary["meta"] = meta
+            if id_hash_keys:
+                dictionary["id_hash_keys"] = id_hash_keys
+            docs.append(Document.from_dict(dictionary))
+
+        return docs
diff --git a/rest_api/rest_api/pipeline/pipelines_faq.haystack-pipeline.yml b/rest_api/rest_api/pipeline/pipelines_faq.haystack-pipeline.yml
new file mode 100644
index 0000000000..3473660ff1
--- /dev/null
+++ b/rest_api/rest_api/pipeline/pipelines_faq.haystack-pipeline.yml
@@ -0,0 +1,40 @@
+# To allow your IDE to autocomplete and validate your YAML pipelines, name them as <name of your choice>.haystack-pipeline.yml
+
+version: ignore
+
+components:    # define all the building-blocks for Pipeline
+  - name: DocumentStore
+    type: ElasticsearchDocumentStore
+    params:
+      host: localhost
+      embedding_field: question_emb
+      embedding_dim: 384
+      excluded_meta_data:
+        - question_emb
+      similarity: cosine
+  - name: Retriever
+    type: EmbeddingRetriever
+    params:
+      document_store: DocumentStore    # params can reference other components defined in the YAML
+      embedding_model: sentence-transformers/all-MiniLM-L6-v2
+      scale_score: False
+  - name: Doc2Answers       # custom-name for the component; helpful for visualization & debugging
+    type: Docs2Answers    # Haystack Class name for the component
+  - name: CsvTextConverter
+    type: CsvTextConverter
+
+pipelines:
+  - name: query    # a sample extractive-qa Pipeline
+    nodes:
+      - name: Retriever
+        inputs: [Query]
+      - name: Doc2Answers
+        inputs: [Retriever]
+  - name: indexing
+    nodes:
+      - name: CsvTextConverter
+        inputs: [File]
+      - name: Retriever
+        inputs: [ CsvTextConverter ]
+      - name: DocumentStore
+        inputs: [ Retriever ]
diff --git a/test/nodes/test_file_converter.py b/test/nodes/test_file_converter.py
index d515287b15..a47bb6fb5a 100644
--- a/test/nodes/test_file_converter.py
+++ b/test/nodes/test_file_converter.py
@@ -1,10 +1,14 @@
+from typing import List
+
 import os
 import sys
 from pathlib import Path
 import subprocess
+import csv
 
 import pytest
 
+from haystack import Document
 from haystack.nodes import (
     MarkdownConverter,
     DocxToTextConverter,
@@ -14,6 +18,7 @@
     AzureConverter,
     ParsrConverter,
     TextConverter,
+    CsvTextConverter,
 )
 
 from ..conftest import SAMPLES_PATH
@@ -265,3 +270,30 @@ def test_id_hash_keys_from_pipeline_params():
 
     assert len(documents) == 2
     assert len(unique_ids) == 2
+
+
+def write_as_csv(data: List[List[str]], file_path: Path):
+    with open(file_path, "w") as f:
+        writer = csv.writer(f)
+        writer.writerows(data)
+
+
+@pytest.mark.integration
+def test_csv_to_document_with_qa_headers(tmp_path):
+    node = CsvTextConverter()
+    csv_path = tmp_path / "csv_qa_with_headers.csv"
+    rows = [
+        ["question", "answer"],
+        ["What is Haystack ?", "Haystack is an NLP Framework to use transformers in your Applications."],
+    ]
+    write_as_csv(rows, csv_path)
+
+    output, edge = node.run(file_paths=csv_path)
+    assert edge == "output_1"
+    assert "documents" in output
+    assert len(output["documents"]) == 1
+
+    doc = output["documents"][0]
+    assert isinstance(doc, Document)
+    assert doc.content == "What is Haystack ?"
+    assert doc.meta["answer"] == "Haystack is an NLP Framework to use transformers in your Applications."


Code review comments:
diff_hunk: @@ -0,0 +1,54 @@
+from typing import Union, List, Optional, Any, Dict
+
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+from haystack import Document
+from haystack.nodes.file_converter import BaseConverter
+
+
+logger = logging.getLogger(__name__)
+
+
+class CsvTextConverter(BaseConverter):
+    """
+    Converts Question & Answers CSV files to text Documents.
+    """
+
+    outgoing_edges = 1
+
+    def convert(
+        self,
+        file_path: Union[Path, List[Path], str, List[str], List[Union[Path, str]]],
+        meta: Optional[Dict[str, Any]],
+        remove_numeric_tables: Optional[bool] = None,
+        valid_languages: Optional[List[str]] = None,
+        encoding: Optional[str] = "UTF-8",
+        id_hash_keys: Optional[List[str]] = None,
+    ) -> List[Document]:
+        """
+        Load CVS file and convert it to documents.
+
+        :param file_path: Path to a CSV file containing two columns.
+            The first will be interpreted as a question, the second as content.
+        :returns: List of document, 1 document per line in the CSV.
+        """
+        df = pd.read_csv(file_path, encoding=encoding)
path: haystack/nodes/file_converter/csv.py
text: @user1:
Can pandas really handle all these file options? 

@user2:
What do you mean? https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html

@user1:
All the options in the signature `Union[Path, List[Path], str, List[str], List[Union[Path, str]]]`

@user2:
Ahh I see! Sorry I misunderstood. Good point, we need to distinguish at least between lists and non-lists.

diff_hunk: @@ -0,0 +1,54 @@
+from typing import Union, List, Optional, Any, Dict
+
+import logging
+from pathlib import Path
+
+import pandas as pd
+
+from haystack import Document
+from haystack.nodes.file_converter import BaseConverter
+
+
+logger = logging.getLogger(__name__)
+
+
+class CsvTextConverter(BaseConverter):
+    """
+    Converts Question & Answers CSV files to text Documents.
+    """
+
+    outgoing_edges = 1
+
+    def convert(
+        self,
+        file_path: Union[Path, List[Path], str, List[str], List[Union[Path, str]]],
+        meta: Optional[Dict[str, Any]],
+        remove_numeric_tables: Optional[bool] = None,
+        valid_languages: Optional[List[str]] = None,
+        encoding: Optional[str] = "UTF-8",
+        id_hash_keys: Optional[List[str]] = None,
+    ) -> List[Document]:
+        """
+        Load CVS file and convert it to documents.
+
+        :param file_path: Path to a CSV file containing two columns.
+            The first will be interpreted as a question, the second as content.
+        :returns: List of document, 1 document per line in the CSV.
+        """
+        df = pd.read_csv(file_path, encoding=encoding)
+
+        df.fillna(value="", inplace=True)
+        df["question"] = df["question"].apply(lambda x: x.strip())
path: haystack/nodes/file_converter/csv.py
text: @user1:
Can we add some checks for malformed files?