Repository: deepset-ai/haystack
Pull request: refactor: simplify Summarizer, add Document Merger
### Related Issues
- fixes #3403

### Proposed Changes:
As discussed in #3403
- make the summarizer suitable for indexing pipelines
- write the summarization results in meta instead of altering document content (current behavior)
- remove `generate_single_summary` parameter: currently it transforms several documents into one
(it is better to design a dedicated node for the purpose of merging documents)

### How did you test it?
- Adapted some tests
- Other tests to be added?

### Notes for the reviewer
Just a first draft to understand what this change breaks

### 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
- [ ] 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
- [ ] 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:
Make Summarizer work in indexing pipeline
Currently, when the Summarizer is run at query time, its primary output is the summarized text, and the original, full text is stored in metadata as "context". 

There should be support for the Summarizer to be used in an indexing pipeline. When run, the Summarizer should not change `Document.content` but should rather populate metadata with a `summary` field.

Hints from issue discussion:
@brandenchan it seems fair and reasonable to me!

How to distinguish if the Summarizer is used for queryies or for indexing?
Any suggestions/resources?
So `BaseDocumentClassifier.run()` takes `root_node` as an argument. When `root_node=="File"` it's indexing, and when `root_node=="Query"` it's querying. I'm sure that we could implement the same for `BaseSummarizer.run()`. 
To add my two cents, I believe it's better if Summarizer works consistently in both pipeline types though. In a query pipeline the fact that content is unchanged, and simply a `summary` field appears in the metadata, would be already great (and in  my opinion generally better). 
> To add my two cents, I believe it's better if Summarizer works consistently in both pipeline types though. In a query pipeline the fact that content is unchanged, and simply a `summary` field appears in the metadata, would be already great (and in my opinion generally better).

Yes I agree with this. It would bring it in line with the `DocumentClassifier` and also the `EntityExtractor`
Count on me for this challenge! :smiley: 
While working on this, I noticed that currently there is a `generate_single_summary` parameter (default value is False).

Based on my experiments and the [docs](https://docs.haystack.deepset.ai/docs/pipelines#example-creating-a-retriever-ranker-summarizer-pipeline), I can say that if `generate_single_summary` is True, the result is only one document.
Its text is the summary, while the context is the sum of all the documents' texts.

It can make sense in querying pipelines.

**But now we want to leave the text unchanged and add the summary in a meta field**.
This seems appropriate both for querying and indexing pipelines.

An option could be:
if `generate_single_summary` is True, we leave the document texts unchanged
and add the same single summary to every single document.

@ZanSara @brandenchan WDYT? Any better ideas?


Wow... Good point. You make a nice proposal, but I think we can do better :grin: 

I propose to remove the `generate_single_summary` feature altogether.

If users wants a single summary from all the retrieved documents, fine, I'd rather give them a node that merges small documents into bigger ones. We have the `PreProcessor` for splitting, let's make a small `DocumentMerger` for merging. It might come handy for other usecases as well. This makes `Summarizer` simpler and smaller, and gives us a new node with a simple and clear task that we can reuse.

You might need to update the documentation to clarify this change. I'm not sure there are tutorials involving this node, have a look.
Generally speaking, I like your idea of ​​making the `Summarizer` simpler.

However, I think that most of the `Summarizer` use cases presented in the documentation for now, leverage this parameter: they use the `Summarizer` to do a sanity check of the search results.

In any case, if you think it is the best thing, I proceed to the simplification proposed by you.
No tutorials involved, but several docs.
> they use the Summarizer to do a sanity check of the search results.

Ok this is interesting! I'm not too familiar with the Summarizer docs actually. If that's too much work to remove `generate_single_summary`, your original plan works too! Have a look and feel free to choose your way. Either is good for me :slightly_smiling_face: 
So funnily enough I raised this issue in the first place because we wanted the docs to have real world use cases that are more realistic! The use case that we are aiming for is one where summaries are attached to all documents at index time so that after retrieval, documents can be displayed with their text or their summary.

Commit message: added test that will fail in 1.12

Patch to review:
diff --git a/haystack/nodes/other/document_merger.py b/haystack/nodes/other/document_merger.py
new file mode 100644
index 0000000000..87d37d2009
--- /dev/null
+++ b/haystack/nodes/other/document_merger.py
@@ -0,0 +1,110 @@
+import logging
+from copy import deepcopy
+from typing import Optional, List, Dict, Union
+
+from haystack.schema import Document
+from haystack.nodes.base import BaseComponent
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentMerger(BaseComponent):
+    """
+    A node to merge the texts of the documents.
+    """
+
+    outgoing_edges = 1
+
+    def __init__(self, separator: str = " "):
+        """
+        :param separator: The separator that appears between subsequent merged documents.
+        """
+        super().__init__()
+        self.separator = separator
+
+    def merge(self, documents: List[Document], separator: Optional[str] = None) -> List[Document]:
+        """
+        Produce a list made up of a single document, which contains all the texts of the documents provided.
+
+        :param separator: The separator that appears between subsequent merged documents.
+        :return: List of Documents
+        """
+        if len(documents) == 0:
+            raise AttributeError("Document Merger needs at least one document to merge.")
+        if not all(doc.content_type == "text" for doc in documents):
+            raise AttributeError(
+                "Some of the documents provided are non-textual. Document Merger only works on textual documents."
+            )
+
+        if separator is None:
+            separator = self.separator
+
+        merged_content = separator.join([doc.content for doc in documents])
+        common_meta = self._extract_common_meta_dict(documents)
+
+        merged_document = Document(content=merged_content, meta=common_meta)
+        return [merged_document]
+
+    def run(self, documents: List[Document], separator: Optional[str] = None):  # type: ignore
+        results: Dict = {"documents": []}
+        if documents:
+            results["documents"] = self.merge(documents=documents, separator=separator)
+        return results, "output_1"
+
+    def run_batch(  # type: ignore
+        self, documents: Union[List[Document], List[List[Document]]], separator: Optional[str] = None
+    ):
+        is_doclist_flat = isinstance(documents[0], Document)
+        if is_doclist_flat:
+            flat_result: List[Document] = self.merge(
+                documents=[doc for doc in documents if isinstance(doc, Document)], separator=separator
+            )
+            return {"documents": flat_result}, "output_1"
+        else:
+            nested_result: List[List[Document]] = [
+                self.merge(documents=docs_lst, separator=separator)
+                for docs_lst in documents
+                if isinstance(docs_lst, list)
+            ]
+            return {"documents": nested_result}, "output_1"
+
+    def _extract_common_meta_dict(self, documents: List[Document]) -> dict:
+        """
+        Given a list of documents, extract a dictionary containing the meta fields
+        that are common to all the documents
+        """
+        flattened_meta = [self._flatten_dict(d.meta) for d in documents]
+        common_meta_flat_dict = deepcopy(flattened_meta[0])
+        for doc in flattened_meta[1:]:
+            if len(common_meta_flat_dict) == 0:
+                break
+            for k, v in doc.items():
+                if k in common_meta_flat_dict:
+                    if common_meta_flat_dict[k] != v:
+                        del common_meta_flat_dict[k]
+        common_meta_nested_dict = self._nest_dict(common_meta_flat_dict)
+        return common_meta_nested_dict
+
+    def _flatten_dict(self, d: dict, parent_key="") -> dict:
+        items: List = []
+        for k, v in d.items():
+            new_key = (parent_key, k) if parent_key else k
+            if isinstance(v, dict):
+                items.extend(self._flatten_dict(v, new_key).items())
+            else:
+                items.append((new_key, v))
+        return dict(items)
+
+    def _nest_dict(self, d: dict) -> dict:
+        nested_dict: dict = {}
+        for key, value in d.items():
+            target = nested_dict
+            if isinstance(key, tuple):
+                for k in key[:-1]:  # traverse all keys but the last
+                    target = target.setdefault(k, {})
+                target[key[-1]] = value
+            else:
+                target[key] = value
+        while any(isinstance(k, tuple) for k in nested_dict.keys()):
+            nested_dict = self._nest_dict(nested_dict)
+        return nested_dict
diff --git a/haystack/nodes/summarizer/base.py b/haystack/nodes/summarizer/base.py
index d2402c4d13..17198cdadd 100644
--- a/haystack/nodes/summarizer/base.py
+++ b/haystack/nodes/summarizer/base.py
@@ -19,12 +19,8 @@ def predict(self, documents: List[Document], generate_single_summary: Optional[b
         Abstract method for creating a summary.
 
         :param documents: Related documents (e.g. coming from a retriever) that the answer shall be conditioned on.
-        :param generate_single_summary: Whether to generate a single summary for all documents or one summary per document.
-                                        If set to "True", all docs will be joined to a single string that will then
-                                        be summarized.
-                                        Important: The summary will depend on the order of the supplied documents!
-        :return: List of Documents, where Document.content contains the summarization and Document.meta["context"]
-                 the original, not summarized text
+        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
+        :return: List of Documents, where Document.meta["summary"] contains the summarization
         """
         pass
 
@@ -54,7 +50,7 @@ def run_batch(  # type: ignore
     ):
 
         results = self.predict_batch(
-            documents=documents, generate_single_summary=generate_single_summary, batch_size=batch_size
+            documents=documents, batch_size=batch_size, generate_single_summary=generate_single_summary
         )
 
         return {"documents": results}, "output_1"
diff --git a/haystack/nodes/summarizer/transformers.py b/haystack/nodes/summarizer/transformers.py
index 531c246b26..322a5f3b3b 100644
--- a/haystack/nodes/summarizer/transformers.py
+++ b/haystack/nodes/summarizer/transformers.py
@@ -6,7 +6,6 @@
 import torch
 from tqdm.auto import tqdm
 from transformers import pipeline
-from transformers.models.auto.modeling_auto import AutoModelForSeq2SeqLM
 
 from haystack.schema import Document
 from haystack.nodes.summarizer.base import BaseSummarizer
@@ -34,19 +33,18 @@ class TransformersSummarizer(BaseSummarizer):
     |
     |     # Summarize
     |     summary = summarizer.predict(
-    |        documents=docs,
-    |        generate_single_summary=True
-    |     )
+    |        documents=docs)
     |
-    |     # Show results (List of Documents, containing summary and original text)
+    |     # Show results (List of Documents, containing summary and original content)
     |     print(summary)
     |
     |    [
     |      {
-    |        "text": "California's largest electricity provider has turned off power to hundreds of thousands of customers.",
+    |        "content": "PGE stated it scheduled the blackouts in response to forecasts for high winds amid dry conditions. ...",
     |        ...
     |        "meta": {
-    |          "context": "PGE stated it scheduled the blackouts in response to forecasts for high winds amid dry conditions. ..."
+    |                   "summary": "California's largest electricity provider has turned off power to hundreds of thousands of customers.",
+    |                   ...
     |              },
     |        ...
     |      },
@@ -83,12 +81,8 @@ def __init__(
         :param min_length: Minimum length of summarized text
         :param use_gpu: Whether to use GPU (if available).
         :param clean_up_tokenization_spaces: Whether or not to clean up the potential extra spaces in the text output
-        :param separator_for_single_summary: If `generate_single_summary=True` in `predict()`, we need to join all docs
-                                             into a single text. This separator appears between those subsequent docs.
-        :param generate_single_summary: Whether to generate a single summary for all documents or one summary per document.
-                                        If set to "True", all docs will be joined to a single string that will then
-                                        be summarized.
-                                        Important: The summary will depend on the order of the supplied documents!
+        :param separator_for_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
+        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
         :param batch_size: Number of documents to process at a time.
         :param progress_bar: Whether to show a progress bar.
         :param use_auth_token: The API token used to download private models from Huggingface.
@@ -103,6 +97,13 @@ def __init__(
         """
         super().__init__()
 
+        if generate_single_summary is True:
+            raise ValueError(
+                "'generate_single_summary' has been removed. Instead, you can use the Document Merger to merge documents before applying the Summarizer."
+            )
+        self.separator_for_single_summary = separator_for_single_summary
+        self.generate_single_summary = generate_single_summary
+
         self.devices, _ = initialize_device_settings(devices=devices, use_cuda=use_gpu, multi_gpu=False)
         if len(self.devices) > 1:
             logger.warning(
@@ -110,20 +111,20 @@ def __init__(
                 f"using the first device {self.devices[0]}."
             )
 
-        # TODO AutoModelForSeq2SeqLM is only necessary with transformers==4.1.1, with newer versions use the pipeline directly
         if tokenizer is None:
             tokenizer = model_name_or_path
-        model = AutoModelForSeq2SeqLM.from_pretrained(
-            pretrained_model_name_or_path=model_name_or_path, revision=model_version, use_auth_token=use_auth_token
-        )
+
         self.summarizer = pipeline(
-            "summarization", model=model, tokenizer=tokenizer, device=self.devices[0], use_auth_token=use_auth_token
+            task="summarization",
+            model=model_name_or_path,
+            tokenizer=tokenizer,
+            revision=model_version,
+            device=self.devices[0],
+            use_auth_token=use_auth_token,
         )
         self.max_length = max_length
         self.min_length = min_length
         self.clean_up_tokenization_spaces = clean_up_tokenization_spaces
-        self.separator_for_single_summary = separator_for_single_summary
-        self.generate_single_summary = generate_single_summary
         self.print_log: Set[str] = set()
         self.batch_size = batch_size
         self.progress_bar = progress_bar
@@ -134,29 +135,22 @@ def predict(self, documents: List[Document], generate_single_summary: Optional[b
         These document can for example be retrieved via the Retriever.
 
         :param documents: Related documents (e.g. coming from a retriever) that the answer shall be conditioned on.
-        :param generate_single_summary: Whether to generate a single summary for all documents or one summary per document.
-                                        If set to "True", all docs will be joined to a single string that will then
-                                        be summarized.
-                                        Important: The summary will depend on the order of the supplied documents!
-        :return: List of Documents, where Document.text contains the summarization and Document.meta["context"]
-                 the original, not summarized text
+        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
+        :return: List of Documents, where Document.meta["summary"] contains the summarization
         """
+        if generate_single_summary is True:
+            raise ValueError(
+                "'generate_single_summary' has been removed. Instead, you can use the Document Merger to merge documents before applying the Summarizer."
+            )
+
         if self.min_length > self.max_length:
             raise AttributeError("min_length cannot be greater than max_length")
 
         if len(documents) == 0:
             raise AttributeError("Summarizer needs at least one document to produce a summary.")
 
-        if generate_single_summary is None:
-            generate_single_summary = self.generate_single_summary
-
         contexts: List[str] = [doc.content for doc in documents]
 
-        if generate_single_summary:
-            # Documents order is very important to produce summary.
-            # Different order of same documents produce different summary.
-            contexts = [self.separator_for_single_summary.join(contexts)]
-
         encoded_input = self.summarizer.tokenizer(contexts, verbose=False)
         for input_id in encoded_input["input_ids"]:
             tokens_count: int = len(input_id)
@@ -182,15 +176,9 @@ def predict(self, documents: List[Document], generate_single_summary: Optional[b
 
         result: List[Document] = []
 
-        if generate_single_summary:
-            for context, summarized_answer in zip(contexts, summaries):
-                cur_doc = Document(content=summarized_answer["summary_text"], meta={"context": context})
-                result.append(cur_doc)
-        else:
-            for context, summarized_answer, document in zip(contexts, summaries, documents):
-                cur_doc = Document(content=summarized_answer["summary_text"], meta=document.meta)
-                cur_doc.meta.update({"context": context})
-                result.append(cur_doc)
+        for summary, document in zip(summaries, documents):
+            document.meta.update({"summary": summary["summary_text"]})
+            result.append(document)
 
         return result
 
@@ -206,13 +194,13 @@ def predict_batch(
 
         :param documents: Single list of related documents or list of lists of related documents
                           (e.g. coming from a retriever) that the answer shall be conditioned on.
-        :param generate_single_summary: Whether to generate a single summary for each provided document list or
-                                        one summary per document.
-                                        If set to "True", all docs of a document list will be joined to a single string
-                                        that will then be summarized.
-                                        Important: The summary will depend on the order of the supplied documents!
+        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
         :param batch_size: Number of Documents to process at a time.
         """
+        if generate_single_summary is True:
+            raise ValueError(
+                "'generate_single_summary' has been removed. Instead, you can use the Document Merger to merge documents before applying the Summarizer."
+            )
 
         if self.min_length > self.max_length:
             raise AttributeError("min_length cannot be greater than max_length")
@@ -225,14 +213,8 @@ def predict_batch(
         if batch_size is None:
             batch_size = self.batch_size
 
-        if generate_single_summary is None:
-            generate_single_summary = self.generate_single_summary
-
-        single_doc_list = False
-        if isinstance(documents[0], Document):
-            single_doc_list = True
-
-        if single_doc_list:
+        is_doclist_flat = isinstance(documents[0], Document)
+        if is_doclist_flat:
             contexts = [doc.content for doc in documents if isinstance(doc, Document)]
         else:
             contexts = [
@@ -240,19 +222,8 @@ def predict_batch(
                 for docs in documents
                 if isinstance(docs, list)
             ]
-
-        if generate_single_summary:
-            if single_doc_list:
-                contexts = [self.separator_for_single_summary.join(contexts)]
-            else:
-                contexts = [self.separator_for_single_summary.join(context_group) for context_group in contexts]
-            number_of_docs = [1 for _ in contexts]
-        else:
-            if single_doc_list:
-                number_of_docs = [1 for _ in contexts]
-            else:
-                number_of_docs = [len(context_group) for context_group in contexts]
-                contexts = list(itertools.chain.from_iterable(contexts))
+            number_of_docs = [len(context_group) for context_group in contexts]
+            contexts = list(itertools.chain.from_iterable(contexts))
 
         encoded_input = self.summarizer.tokenizer(contexts, verbose=False)
         for input_id in encoded_input["input_ids"]:
@@ -286,26 +257,30 @@ def predict_batch(
         ):
             summaries.extend(summary_batch)
 
-        # Group summaries together
-        grouped_summaries = []
-        grouped_contexts = []
-        left_idx = 0
-        right_idx = 0
-        for number in number_of_docs:
-            right_idx = left_idx + number
-            grouped_summaries.append(summaries[left_idx:right_idx])
-            grouped_contexts.append(contexts[left_idx:right_idx])
-            left_idx = right_idx
-
-        result = []
-        for summary_group, context_group in zip(grouped_summaries, grouped_contexts):
-            cur_summaries = [
-                Document(content=summary["summary_text"], meta={"context": context})
-                for summary, context in zip(summary_group, context_group)
-            ]
-            if single_doc_list:
-                result.append(cur_summaries[0])
-            else:
-                result.append(cur_summaries)  # type: ignore
-
-        return result
+        if is_doclist_flat:
+            flat_result: List[Document] = []
+            flat_doc_list: List[Document] = [doc for doc in documents if isinstance(doc, Document)]
+            for summary, document in zip(summaries, flat_doc_list):
+                document.meta.update({"summary": summary["summary_text"]})
+                flat_result.append(document)
+            return flat_result
+        else:
+            nested_result: List[List[Document]] = []
+            nested_doc_list: List[List[Document]] = [lst for lst in documents if isinstance(lst, list)]
+
+            # Group summaries together
+            grouped_summaries = []
+            left_idx = 0
+            right_idx = 0
+            for number in number_of_docs:
+                right_idx = left_idx + number
+                grouped_summaries.append(summaries[left_idx:right_idx])
+                left_idx = right_idx
+
+            for summary_group, docs_group in zip(grouped_summaries, nested_doc_list):
+                cur_summaries = []
+                for summary, document in zip(summary_group, docs_group):
+                    document.meta.update({"summary": summary["summary_text"]})
+                    cur_summaries.append(document)
+                nested_result.append(cur_summaries)
+            return nested_result
diff --git a/haystack/pipelines/standard_pipelines.py b/haystack/pipelines/standard_pipelines.py
index de18bd8ac5..85b9520084 100644
--- a/haystack/pipelines/standard_pipelines.py
+++ b/haystack/pipelines/standard_pipelines.py
@@ -1,26 +1,26 @@
 import logging
 from abc import ABC
 from copy import deepcopy
-from pathlib import Path
 from functools import wraps
-from typing import List, Optional, Dict, Any, Union
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
 
 try:
     from typing import Literal
 except ImportError:
     from typing_extensions import Literal  # type: ignore
 
-from haystack.schema import Document, EvaluationResult, MultiLabel
+from haystack.document_stores.base import BaseDocumentStore
 from haystack.nodes.answer_generator.base import BaseGenerator
 from haystack.nodes.other.docs2answers import Docs2Answers
+from haystack.nodes.other.document_merger import DocumentMerger
+from haystack.nodes.question_generator.question_generator import QuestionGenerator
 from haystack.nodes.reader.base import BaseReader
 from haystack.nodes.retriever.base import BaseRetriever
 from haystack.nodes.summarizer.base import BaseSummarizer
 from haystack.nodes.translator.base import BaseTranslator
-from haystack.nodes.question_generator.question_generator import QuestionGenerator
-from haystack.document_stores.base import BaseDocumentStore
 from haystack.pipelines.base import Pipeline
-
+from haystack.schema import Document, EvaluationResult, MultiLabel
 
 logger = logging.getLogger(__name__)
 
@@ -456,17 +456,29 @@ class SearchSummarizationPipeline(BaseStandardPipeline):
     Pipeline that retrieves documents for a query and then summarizes those documents.
     """
 
-    def __init__(self, summarizer: BaseSummarizer, retriever: BaseRetriever, return_in_answer_format: bool = False):
+    def __init__(
+        self,
+        summarizer: BaseSummarizer,
+        retriever: BaseRetriever,
+        generate_single_summary: bool = False,
+        return_in_answer_format: bool = False,
+    ):
         """
         :param summarizer: Summarizer instance
         :param retriever: Retriever instance
+        :param generate_single_summary: Whether to generate a single summary for all documents or one summary per document.
         :param return_in_answer_format: Whether the results should be returned as documents (False) or in the answer
                                         format used in other QA pipelines (True). With the latter, you can use this
                                         pipeline as a "drop-in replacement" for other QA pipelines.
         """
         self.pipeline = Pipeline()
         self.pipeline.add_node(component=retriever, name="Retriever", inputs=["Query"])
-        self.pipeline.add_node(component=summarizer, name="Summarizer", inputs=["Retriever"])
+        if generate_single_summary is True:
+            document_merger = DocumentMerger()
+            self.pipeline.add_node(component=document_merger, name="Document Merger", inputs=["Retriever"])
+            self.pipeline.add_node(component=summarizer, name="Summarizer", inputs=["Document Merger"])
+        else:
+            self.pipeline.add_node(component=summarizer, name="Summarizer", inputs=["Retriever"])
         self.return_in_answer_format = return_in_answer_format
 
     def run(self, query: str, params: Optional[dict] = None, debug: Optional[bool] = None):
@@ -489,9 +501,9 @@ def run(self, query: str, params: Optional[dict] = None, debug: Optional[bool] =
             for doc in docs:
                 cur_answer = {
                     "query": query,
-                    "answer": doc.content,
+                    "answer": doc.meta.pop("summary"),
                     "document_id": doc.id,
-                    "context": doc.meta.pop("context"),
+                    "context": doc.content,
                     "score": None,
                     "offset_start": None,
                     "offset_end": None,
@@ -527,9 +539,9 @@ def run_batch(self, queries: List[str], params: Optional[dict] = None, debug: Op
                 for doc in cur_docs:
                     cur_answer = {
                         "query": query,
-                        "answer": doc.content,
+                        "answer": doc.meta.pop("summary"),
                         "document_id": doc.id,
-                        "context": doc.meta.pop("context"),
+                        "context": doc.content,
                         "score": None,
                         "offset_start": None,
                         "offset_end": None,
diff --git a/test/nodes/test_document_merger.py b/test/nodes/test_document_merger.py
new file mode 100644
index 0000000000..9a8ad78988
--- /dev/null
+++ b/test/nodes/test_document_merger.py
@@ -0,0 +1,97 @@
+from haystack import Document
+from haystack.nodes.other.document_merger import DocumentMerger
+
+doc_dicts = [
+    {
+        "meta": {
+            "name": "name_1",
+            "year": "2020",
+            "month": "01",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
+        },
+        "content": "text_1",
+    },
+    {
+        "meta": {
+            "name": "name_2",
+            "year": "2020",
+            "month": "02",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
+        },
+        "content": "text_2",
+    },
+    {
+        "meta": {
+            "name": "name_3",
+            "year": "2020",
+            "month": "03",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 7, "c": {"3": 3}},
+        },
+        "content": "text_3",
+    },
+    {
+        "meta": {
+            "name": "name_4",
+            "year": "2021",
+            "month": "01",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
+        },
+        "content": "text_4",
+    },
+    {
+        "meta": {
+            "name": "name_5",
+            "year": "2021",
+            "month": "02",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
+        },
+        "content": "text_5",
+    },
+    {
+        "meta": {
+            "name": "name_6",
+            "year": "2021",
+            "month": "03",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
+        },
+        "content": "text_6",
+    },
+]
+
+documents = [Document.from_dict(doc) for doc in doc_dicts]
+
+
+def test_document_merger_merge():
+    separator = "|"
+    dm = DocumentMerger(separator=separator)
+    merged_list = dm.merge(documents)
+
+    assert len(merged_list) == 1
+    assert merged_list[0].content == separator.join([doc["content"] for doc in doc_dicts])
+    assert merged_list[0].meta == {"flat_field": 1, "nested_field": {1: 2, "c": {"3": 3}}}
+
+
+def test_document_merger_run():
+    separator = "|"
+    dm = DocumentMerger(separator=separator)
+    result = dm.run(documents)
+
+    assert len(result[0]["documents"]) == 1
+    assert result[0]["documents"][0].content == separator.join([doc["content"] for doc in doc_dicts])
+    assert result[0]["documents"][0].meta == {"flat_field": 1, "nested_field": {1: 2, "c": {"3": 3}}}
+
+
+def test_document_merger_run_batch():
+    separator = "|"
+    dm = DocumentMerger(separator=separator)
+    batch_result = dm.run_batch([documents, documents])
+
+    assert len(batch_result[0]["documents"]) == 2
+    assert batch_result[0]["documents"][0][0].content == separator.join([doc["content"] for doc in doc_dicts])
+    assert batch_result[0]["documents"][0][0].meta == {"flat_field": 1, "nested_field": {1: 2, "c": {"3": 3}}}
diff --git a/test/nodes/test_extractor_translation.py b/test/nodes/test_extractor_translation.py
new file mode 100644
index 0000000000..5d028f5e41
--- /dev/null
+++ b/test/nodes/test_extractor_translation.py
@@ -0,0 +1,33 @@
+import pytest
+
+from haystack.pipelines import TranslationWrapperPipeline, ExtractiveQAPipeline
+from haystack.nodes import DensePassageRetriever, EmbeddingRetriever
+from .test_summarizer import SPLIT_DOCS
+
+# Keeping few (retriever,document_store,reader) combination to reduce test time
+@pytest.mark.integration
+@pytest.mark.elasticsearch
+@pytest.mark.summarizer
+@pytest.mark.parametrize(
+    "retriever,document_store,reader",
+    [("embedding", "memory", "farm"), ("elasticsearch", "elasticsearch", "farm")],
+    indirect=True,
+)
+def test_extractive_qa_pipeline_with_translator(
+    document_store, retriever, reader, en_to_de_translator, de_to_en_translator
+):
+    document_store.write_documents(SPLIT_DOCS)
+
+    if isinstance(retriever, EmbeddingRetriever) or isinstance(retriever, DensePassageRetriever):
+        document_store.update_embeddings(retriever=retriever)
+
+    query = "Wo steht der Eiffelturm?"
+    base_pipeline = ExtractiveQAPipeline(retriever=retriever, reader=reader)
+    pipeline = TranslationWrapperPipeline(
+        input_translator=de_to_en_translator, output_translator=en_to_de_translator, pipeline=base_pipeline
+    )
+    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}})
+    assert len(output["documents"]) == 2
+    answers_texts = [el.answer for el in output["answers"]]
+
+    assert "Frankreich" in answers_texts
diff --git a/test/nodes/test_summarizer.py b/test/nodes/test_summarizer.py
index 239bb71074..ad7cb5a0da 100644
--- a/test/nodes/test_summarizer.py
+++ b/test/nodes/test_summarizer.py
@@ -1,10 +1,12 @@
-from copy import deepcopy
-
+from ensurepip import version
 import pytest
 
+import haystack
 from haystack.schema import Document
 from haystack.pipelines import SearchSummarizationPipeline
-from haystack.nodes import DensePassageRetriever, EmbeddingRetriever, TransformersSummarizer
+from haystack.nodes import DensePassageRetriever, EmbeddingRetriever
+from haystack.nodes.other.document_merger import DocumentMerger
+
 
 DOCS = [
     Document(
@@ -43,15 +45,7 @@ def test_summarization(summarizer):
     summarized_docs = summarizer.predict(documents=DOCS)
     assert len(summarized_docs) == len(DOCS)
     for expected_summary, summary in zip(EXPECTED_SUMMARIES, summarized_docs):
-        assert expected_summary == summary.content
-
-
-@pytest.mark.integration
-@pytest.mark.summarizer
-def test_summarization_one_summary(summarizer):
-    summarized_docs = summarizer.predict(documents=SPLIT_DOCS, generate_single_summary=True)
-    assert len(summarized_docs) == 1
-    assert EXPECTED_ONE_SUMMARIES[0] == summarized_docs[0].content
+        assert expected_summary == summary.meta["summary"]
 
 
 @pytest.mark.integration
@@ -60,7 +54,7 @@ def test_summarization_batch_single_doc_list(summarizer):
     summarized_docs = summarizer.predict_batch(documents=DOCS)
     assert len(summarized_docs) == len(DOCS)
     for expected_summary, summary in zip(EXPECTED_SUMMARIES, summarized_docs):
-        assert expected_summary == summary.content
+        assert expected_summary == summary.meta["summary"]
 
 
 @pytest.mark.integration
@@ -70,7 +64,7 @@ def test_summarization_batch_multiple_doc_lists(summarizer):
     assert len(summarized_docs) == 2  # Number of document lists
     assert len(summarized_docs[0]) == len(DOCS)
     for expected_summary, summary in zip(EXPECTED_SUMMARIES, summarized_docs[0]):
-        assert expected_summary == summary.content
+        assert expected_summary == summary.meta["summary"]
 
 
 @pytest.mark.integration
@@ -92,6 +86,35 @@ def test_summarization_pipeline(document_store, retriever, summarizer):
     assert " The Eiffel Tower in Paris has officially opened its doors to the public." == answers[0]["answer"]
 
 
+haystack_version = tuple(int(num) for num in haystack.__version__.split(".")[:2])
+fail_in_v1_12 = pytest.mark.xfail(
+    haystack_version >= (1, 12),
+    reason="'generate_single_summary' should be removed in v1.12, as it was deprecated in v1.10",
+)
+
+
+@fail_in_v1_12
+def test_generate_single_summary_deprecated(summarizer, documents=DOCS):
+    with pytest.raises(ValueError):
+        summarizer.predict(documents, generate_single_summary=True)
+
+
+#
+# Document Merger + Summarizer tests
+#
+
+
+@pytest.mark.integration
+@pytest.mark.summarizer
+def test_summarization_one_summary(summarizer):
+    dm = DocumentMerger()
+    merged_document = dm.merge(documents=SPLIT_DOCS)
+    print(merged_document)
+    summarized_docs = summarizer.predict(documents=merged_document)
+    assert len(summarized_docs) == 1
+    assert EXPECTED_ONE_SUMMARIES[0] == summarized_docs[0].meta["summary"]
+
+
 @pytest.mark.integration
 @pytest.mark.summarizer
 @pytest.mark.parametrize(
@@ -104,45 +127,10 @@ def test_summarization_pipeline_one_summary(document_store, retriever, summarize
         document_store.update_embeddings(retriever=retriever)
 
     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}, "Summarizer": {"generate_single_summary": True}}
+    pipeline = SearchSummarizationPipeline(
+        retriever=retriever, summarizer=summarizer, generate_single_summary=True, return_in_answer_format=True
     )
+    output = pipeline.run(query=query, params={"Retriever": {"top_k": 2}})
     answers = output["answers"]
     assert len(answers) == 1
     assert answers[0]["answer"] in EXPECTED_ONE_SUMMARIES
-
-
-@pytest.mark.summarizer
-def test_metadata_summarizer(summarizer):
-    docs = [
-        Document(
-            content="""PG&E stated it scheduled the blackouts in response to forecasts for high winds amid dry conditions. The aim is to reduce the risk of wildfires. Nearly 800 thousand customers were scheduled to be affected by the shutoffs which were expected to last through at least midday tomorrow.""",
-            meta={
-                "sub_content": "Pegasus Example",
-                "topic": "California's Electricity",
-                "context": "Dummy - PG&E stated it scheduled the blackouts in response to forecasts for high winds amid dry conditions. The aim is to reduce the risk of wildfires.",
-            },
-        ),
-        Document(
-            content="""The tower is 324 metres (1,063 ft) tall, about the same height as an 81-storey building, and the tallest structure in Paris. Its base is square, measuring 125 metres (410 ft) on each side. During its construction, the Eiffel Tower surpassed the Washington Monument to become the tallest man-made structure in the world, a title it held for 41 years until the Chrysler Building in New York City was finished in 1930. It was the first structure to reach a height of 300 metres. Due to the addition of a broadcasting aerial at the top of the tower in 1957, it is now taller than the Chrysler Building by 5.2 metres (17 ft). Excluding transmitters, the Eiffel Tower is the second tallest free-standing structure in France after the Millau Viaduct.""",
-            meta={"sub_content": "Paris best tour best tour", "topic": "Eiffel tower"},
-        ),
-    ]
-    # Original input is overwritten after the "predict". So adding the same input as check_output to assess the output
-    check_output = deepcopy(docs)
-
-    summary = summarizer.predict(documents=docs)
-
-    assert len(summary[0].meta) == len(check_output[0].meta)
-    assert len(summary[1].meta) - 1 == len(check_output[1].meta)
-    assert (
-        summary[0].meta["context"]
-        == """PG&E stated it scheduled the blackouts in response to forecasts for high winds amid dry conditions. The aim is to reduce the risk of wildfires. Nearly 800 thousand customers were scheduled to be affected by the shutoffs which were expected to last through at least midday tomorrow."""
-    )
-
-    summary = summarizer.predict(documents=docs, generate_single_summary=True)
-
-    assert len(summary) == 1
-    summary[0].meta.pop("context")
-    assert not summary[0].meta  # Remaining metadata is not returned in case of a single summary
diff --git a/test/nodes/test_summarizer_translation.py b/test/nodes/test_summarizer_translation.py
deleted file mode 100644
index bc5d8402f0..0000000000
--- a/test/nodes/test_summarizer_translation.py
+++ /dev/null
@@ -1,36 +0,0 @@
-import pytest
-
-from haystack.pipelines import TranslationWrapperPipeline, SearchSummarizationPipeline
-from haystack.nodes import DensePassageRetriever, EmbeddingRetriever
-from .test_summarizer import SPLIT_DOCS
-
-# Keeping few (retriever,document_store) combination to reduce test time
-@pytest.mark.integration
-@pytest.mark.elasticsearch
-@pytest.mark.summarizer
-@pytest.mark.parametrize(
-    "retriever,document_store", [("embedding", "memory"), ("elasticsearch", "elasticsearch")], indirect=True
-)
-def test_summarization_pipeline_with_translator(
-    document_store, retriever, summarizer, en_to_de_translator, de_to_en_translator
-):
-    document_store.write_documents(SPLIT_DOCS)
-
-    if isinstance(retriever, EmbeddingRetriever) or isinstance(retriever, DensePassageRetriever):
-        document_store.update_embeddings(retriever=retriever)
-
-    query = "Wo steht der Eiffelturm?"
-    base_pipeline = SearchSummarizationPipeline(retriever=retriever, summarizer=summarizer)
-    pipeline = TranslationWrapperPipeline(
-        input_translator=de_to_en_translator, output_translator=en_to_de_translator, pipeline=base_pipeline
-    )
-    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
-    assert documents[0].content in [
-        "Der Eiffelturm in Paris ist die höchste von Menschen geschaffene Struktur der Welt geworden.",
-        "Der Eiffelturm in Paris hat offiziell seine Türen für die Öffentlichkeit geöffnet.",
-    ]
diff --git a/test/pipelines/test_eval.py b/test/pipelines/test_eval.py
index 0011f54a5d..037c165cef 100644
--- a/test/pipelines/test_eval.py
+++ b/test/pipelines/test_eval.py
@@ -75,11 +75,11 @@ def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDoc
     assert metrics["Retriever"]["precision"] == 1.0
     assert metrics["Retriever"]["ndcg"] == pytest.approx(0.9461, 1e-4)
     assert metrics["Summarizer"]["mrr"] == 1.0
-    assert metrics["Summarizer"]["map"] == 0.735
-    assert metrics["Summarizer"]["recall_multi_hit"] == 0.8
+    assert metrics["Summarizer"]["map"] == pytest.approx(0.9167, 1e-4)
+    assert metrics["Summarizer"]["recall_multi_hit"] == pytest.approx(0.9167, 1e-4)
     assert metrics["Summarizer"]["recall_single_hit"] == 1.0
-    assert metrics["Summarizer"]["precision"] == 0.8
-    assert metrics["Summarizer"]["ndcg"] == pytest.approx(0.8422, 1e-4)
+    assert metrics["Summarizer"]["precision"] == 1.0
+    assert metrics["Summarizer"]["ndcg"] == pytest.approx(0.9461, 1e-4)
 
 
 @pytest.mark.parametrize("document_store", ["elasticsearch", "faiss", "memory", "milvus1"], indirect=True)
diff --git a/test/pipelines/test_eval_batch.py b/test/pipelines/test_eval_batch.py
index cb713a9e8d..66927a67dc 100644
--- a/test/pipelines/test_eval_batch.py
+++ b/test/pipelines/test_eval_batch.py
@@ -54,6 +54,7 @@ def test_generativeqa_calculate_metrics(
 @pytest.mark.parametrize("retriever_with_docs", ["embedding"], indirect=True)
 def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDocumentStore, retriever_with_docs):
     document_store_with_docs.update_embeddings(retriever=retriever_with_docs)
+    print(document_store_with_docs.get_all_documents())
     summarizer = TransformersSummarizer(model_name_or_path="sshleifer/distill-pegasus-xsum-16-4", use_gpu=False)
     pipeline = SearchSummarizationPipeline(
         retriever=retriever_with_docs, summarizer=summarizer, return_in_answer_format=True
@@ -62,6 +63,8 @@ def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDoc
         labels=EVAL_LABELS, params={"Retriever": {"top_k": 5}}, context_matching_min_length=10
     )
 
+    print(eval_result)
+
     metrics = eval_result.calculate_metrics(document_scope="context")
 
     assert "Retriever" in eval_result
@@ -75,11 +78,11 @@ def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDoc
     assert metrics["Retriever"]["precision"] == 1.0
     assert metrics["Retriever"]["ndcg"] == pytest.approx(0.9461, 1e-4)
     assert metrics["Summarizer"]["mrr"] == 1.0
-    assert metrics["Summarizer"]["map"] == 0.735
-    assert metrics["Summarizer"]["recall_multi_hit"] == 0.8
+    assert metrics["Summarizer"]["map"] == pytest.approx(0.9167, 1e-4)
+    assert metrics["Summarizer"]["recall_multi_hit"] == pytest.approx(0.9167, 1e-4)
     assert metrics["Summarizer"]["recall_single_hit"] == 1.0
-    assert metrics["Summarizer"]["precision"] == 0.8
-    assert metrics["Summarizer"]["ndcg"] == pytest.approx(0.8422, 1e-4)
+    assert metrics["Summarizer"]["precision"] == 1.0
+    assert metrics["Summarizer"]["ndcg"] == pytest.approx(0.9461, 1e-4)
 
 
 EVAL_LABELS = [


Code review comments:
diff_hunk: @@ -0,0 +1,110 @@
+import logging
+from copy import deepcopy
+from typing import Optional, List, Dict, Union
+
+from haystack.schema import Document
+from haystack.nodes.base import BaseComponent
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentMerger(BaseComponent):
+    """
+    A node to merge the texts of the documents.
+    """
+
+    outgoing_edges = 1
+
+    def __init__(self, separator: str = " "):
+        """
+        :param separator: The separator that appears between subsequent merged documents.
+        """
+        super().__init__()
+        self.separator = separator
+
+    def merge(self, documents: List[Document], separator: Optional[str] = None) -> List[Document]:
+        """
+        Produce a list made up of a single document, which contains all the texts of the documents provided.
+
+        :param separator: The separator that appears between subsequent merged documents.
+        :return: List of Documents
+        """
+        if len(documents) == 0:
+            raise AttributeError("Document Merger needs at least one document to merge.")
+        if not all(doc.content_type == "text" for doc in documents):
+            raise AttributeError(
+                "Some of the documents provided are non-textual. Document Merger only works on textual documents."
+            )
+
+        if separator is None:
+            separator = self.separator
+
+        merged_content = separator.join([doc.content for doc in documents])
+        common_meta = self._extract_common_meta_dict(documents)
+
+        merged_document = Document(content=merged_content, meta=common_meta)
+        return [merged_document]
+
+    def run(self, documents: List[Document], separator: Optional[str] = None):  # type: ignore
+        results: Dict = {"documents": []}
+        if documents:
+            results["documents"] = self.merge(documents=documents, separator=separator)
+        return results, "output_1"
+
+    def run_batch(  # type: ignore
+        self, documents: Union[List[Document], List[List[Document]]], separator: Optional[str] = None
+    ):
+        is_doclist_flat = isinstance(documents[0], Document)
+        if is_doclist_flat:
+            flat_result: List[Document] = self.merge(
+                documents=[doc for doc in documents if isinstance(doc, Document)], separator=separator
+            )
+            return {"documents": flat_result}, "output_1"
+        else:
+            nested_result: List[List[Document]] = [
+                self.merge(documents=docs_lst, separator=separator)
+                for docs_lst in documents
+                if isinstance(docs_lst, list)
+            ]
+            return {"documents": nested_result}, "output_1"
+
+    def _extract_common_meta_dict(self, documents: List[Document]) -> dict:
+        """
+        Given a list of documents, extract a dictionary containing the meta fields
+        that are common to all the documents
+        """
+        flattened_meta = [self._flatten_dict(d.meta) for d in documents]
+        common_meta_flat_dict = deepcopy(flattened_meta[0])
+        for doc in flattened_meta[1:]:
+            if len(common_meta_flat_dict) == 0:
+                break
+            for k, v in doc.items():
+                if k in common_meta_flat_dict:
+                    if common_meta_flat_dict[k] != v:
+                        del common_meta_flat_dict[k]
+        common_meta_nested_dict = self._nest_dict(common_meta_flat_dict)
+        return common_meta_nested_dict
+
+    def _flatten_dict(self, d: dict, parent_key="") -> dict:
+        items: List = []
+        for k, v in d.items():
+            new_key = (parent_key, k) if parent_key else k
+            if isinstance(v, dict):
+                items.extend(self._flatten_dict(v, new_key).items())
+            else:
+                items.append((new_key, v))
+        return dict(items)
+
+    def _nest_dict(self, d: dict) -> dict:
+        nested_dict: dict = {}
+        for key, value in d.items():
+            target = nested_dict
+            if isinstance(key, tuple):
+                for k in key[:-1]:  # traverse all keys but the last
+                    target = target.setdefault(k, {})
+                target[key[-1]] = value
+            else:
+                target[key] = value
+        while any(isinstance(k, tuple) for k in nested_dict.keys()):
+            nested_dict = self._nest_dict(nested_dict)
+        return nested_dict
path: haystack/nodes/other/document_merger.py
text: @user1:
Ok I believe this process can be heavily simplified. I don't think the flattening and re-nesting is useful.

An alternative method could look like this (test code, suggestion is below):

```python
import json
from copy import deepcopy
from typing import List, Dict, Any

list_of_dicts = [
    {
        'a': 1,
        'b': 2,
        'c': {
            'aa': 11,
            'bb': 22,
            'cc': 33
        },
        'd': [0, 9, 8],
        'dd': [0, 9, 8, 7],
        'e': {
            'ee': {
                'eee': 444,
                'fff': 555 
            },
            'gg': {
                'eee': 666,
                'fff': 777 
            }
        },
    },
    {
        'a': 1,
        'b': 222222222,
        'c': {
            'aa': 11,
            'bb': 22,
            'cc': 3333333333
        },
        'd': [0, 9, 8],
        'dd': [0, 9, 8, 7777777777],
        'e': {
            'ee': {
                'eee': 444,
                'fff': 555555555 
            },
            'gg': {
                'eeeeeeee': 666,
                'fff': 777 
            }
        }
    }
]

expected_result = {
        'a': 1,
        'c': {
            'aa': 11,
            'bb': 22,
        },
        'd': [0, 9, 8],
        'e': {
            'ee': {
                'eee': 444, 
            },
            'gg': {
                'fff': 777 
            }
        }
    }


def keep_common_keys(list_of_dicts: List[Dict[str, Any]]):
    merge_dictionary = deepcopy(list_of_dicts[0])
    for key, value in list_of_dicts[0].items():

        # if not all other dicts have this key, delete directly
        if not all(key in dict.keys() for dict in list_of_dicts):
            del merge_dictionary[key]

        # if they all have it and it's a dictionary, merge recursively
        elif isinstance(value, dict):
            # Get all the subkeys to merge in a new list
            list_of_subdicts = [dictionary[key] for dictionary in list_of_dicts]
            merge_dictionary[key] = keep_common_keys(list_of_subdicts)

        # If all dicts have this key and it's not a dictionary, delete only if the values differ
        elif not all(value == dict[key] for dict in list_of_dicts):
            del merge_dictionary[key]

    return merge_dictionary
    

print(json.dumps(keep_common_keys(list_of_dicts), indent=4))
assert keep_common_keys(list_of_dicts) == expected_result
```

If you find no issues with the code above (please test and challenge it if it looks suspicious!) then this is how it could look like in the node. Of course feel free to remix this to your liking.

Note that I changed the signature! It now takes the meta fields only, so you will need the caller to pass them.

```suggestion
    def _extract_common_meta_dict(self, metas: List[Dict[str, Any]) -> dict:
        """
        Given a list of document meta fields, extract a dictionary containing the meta fields
        that are common to all the documents.
        """
        merge_dictionary = deepcopy(list_of_dicts[0])
        for key, value in list_of_dicts[0].items():
        
            if not all(key in dict.keys() for dict in list_of_dicts):
                del merge_dictionary[key]
    
            elif isinstance(value, dict):
                list_of_subdicts = [dictionary[key] for dictionary in list_of_dicts]
                merge_dictionary[key] = self._extract_common_meta_dict(list_of_subdicts)
    
            elif not all(value == dict[key] for dict in list_of_dicts):
                del merge_dictionary[key]
    
        return merge_dictionary
```




@author:
Great simplification!

diff_hunk: @@ -206,13 +194,13 @@ def predict_batch(
 
         :param documents: Single list of related documents or list of lists of related documents
                           (e.g. coming from a retriever) that the answer shall be conditioned on.
-        :param generate_single_summary: Whether to generate a single summary for each provided document list or
-                                        one summary per document.
-                                        If set to "True", all docs of a document list will be joined to a single string
-                                        that will then be summarized.
-                                        Important: The summary will depend on the order of the supplied documents!
+        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12
path: haystack/nodes/summarizer/transformers.py
text: @user1:
Let's add a mention to the fact that the DocumentMerger is taking over this task: 

```suggestion
        :param generate_single_summary: This parameter is deprecated and will be removed in Haystack 1.12. To obtain single summaries from multiple documents, consider using the [DocumentMerger](https://docs.haystack.deepset.ai/docs/document_merger)
```

diff_hunk: @@ -92,6 +86,35 @@ def test_summarization_pipeline(document_store, retriever, summarizer):
     assert " The Eiffel Tower in Paris has officially opened its doors to the public." == answers[0]["answer"]
 
 
+haystack_version = tuple(int(num) for num in haystack.__version__.split(".")[:2])
+fail_in_v1_12 = pytest.mark.xfail(
+    haystack_version >= (1, 12),
+    reason="'generate_single_summary' should be removed in v1.12, as it was deprecated in v1.10",
+)
+
+
+@fail_in_v1_12
+def test_generate_single_summary_deprecated(summarizer, documents=DOCS):
+    with pytest.raises(ValueError):
+        summarizer.predict(documents, generate_single_summary=True)
+
+
+#
+# Document Merger + Summarizer tests
+#
+
+
+@pytest.mark.integration
+@pytest.mark.summarizer
+def test_summarization_one_summary(summarizer):
+    dm = DocumentMerger()
+    merged_document = dm.merge(documents=SPLIT_DOCS)
+    print(merged_document)
path: test/nodes/test_summarizer.py
text: @user1:
leftover

```suggestion
```

diff_hunk: @@ -54,6 +54,7 @@ def test_generativeqa_calculate_metrics(
 @pytest.mark.parametrize("retriever_with_docs", ["embedding"], indirect=True)
 def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDocumentStore, retriever_with_docs):
     document_store_with_docs.update_embeddings(retriever=retriever_with_docs)
+    print(document_store_with_docs.get_all_documents())
path: test/pipelines/test_eval_batch.py
text: @user1:
```suggestion
```

diff_hunk: @@ -0,0 +1,97 @@
+from haystack import Document
+from haystack.nodes.other.document_merger import DocumentMerger
+
+doc_dicts = [
+    {
+        "meta": {
+            "name": "name_1",
+            "year": "2020",
+            "month": "01",
+            "flat_field": 1,
+            "nested_field": {1: 2, "a": 5, "c": {"3": 3}},
path: test/nodes/test_document_merger.py
text: @user1:
I see only one small case left, which is the case of additional nested keys.

```suggestion
            "nested_field": {1: 2, "a": 5, "c": {"3": 3}, "d": "I will be dropped by the meta merge algorithm"},
```

diff_hunk: @@ -62,6 +63,8 @@ def test_summarizer_calculate_metrics(document_store_with_docs: ElasticsearchDoc
         labels=EVAL_LABELS, params={"Retriever": {"top_k": 5}}, context_matching_min_length=10
     )
 
+    print(eval_result)
+
path: test/pipelines/test_eval_batch.py
text: @user1:
```suggestion
```

diff_hunk: @@ -103,27 +97,34 @@ def __init__(
         """
         super().__init__()
 
+        if generate_single_summary is True:
+            raise ValueError(
+                "'generate_single_summary' has been removed. Instead, you can use the Document Merger to merge documents before applying the Summarizer."
+            )
+        self.separator_for_single_summary = separator_for_single_summary
+        self.generate_single_summary = generate_single_summary
path: haystack/nodes/summarizer/transformers.py
text: @user1:
I think this line can be removed

diff_hunk: @@ -0,0 +1,110 @@
+import logging
+from copy import deepcopy
+from typing import Optional, List, Dict, Union
+
+from haystack.schema import Document
+from haystack.nodes.base import BaseComponent
+
+logger = logging.getLogger(__name__)
+
+
+class DocumentMerger(BaseComponent):
+    """
+    A node to merge the texts of the documents.
+    """
+
+    outgoing_edges = 1
+
+    def __init__(self, separator: str = " "):
+        """
+        :param separator: The separator that appears between subsequent merged documents.
+        """
+        super().__init__()
+        self.separator = separator
+
+    def merge(self, documents: List[Document], separator: Optional[str] = None) -> List[Document]:
+        """
+        Produce a list made up of a single document, which contains all the texts of the documents provided.
+
+        :param separator: The separator that appears between subsequent merged documents.
+        :return: List of Documents
+        """
+        if len(documents) == 0:
+            raise AttributeError("Document Merger needs at least one document to merge.")
+        if not all(doc.content_type == "text" for doc in documents):
+            raise AttributeError(
+                "Some of the documents provided are non-textual. Document Merger only works on textual documents."
+            )
+
+        if separator is None:
+            separator = self.separator
path: haystack/nodes/other/document_merger.py
text: @user1:
Nitpick:

```suggestion
        separator = separator if separator is not None else self.separator
```