Repository: deepset-ai/haystack
Pull request: feat: prompt at query time
### Related Issues
- fixes https://github.com/deepset-ai/haystack/issues/4394

### Proposed Changes:
- add `prompt_template` param to `PromptTemplate.run` which can be used in pipelines like this
```python
# p is the pipeline
# PROMPT_TEMPLATE is one of the following
# - PromptTemplate: use the given prompt template object
# - str: parses the string depending on its content:
#    - prompt template name: uses the prompt template registered with the given name
#    - prompt template yaml: uses the prompt template specified by the given yaml
#    - prompt text: uses a copy of the default prompt template with the given prompt text
p.run(query=QUERY, params={"prompt_template": PROMPT_TEMPLATE})
```
If you want to specify a specific prompt node only you can target it via params as usual:
```python
p.run(query=QUERY, params={"node_3". {"prompt_template": PROMPT_TEMPLATE}})
```
A prompt template YAML would look like this:
```yaml
name: "question-answering-with-references-custom"
prompt_text: 'Create a concise and informative answer (no more than 50 words) for
  a given question based solely on the given documents. Cite the documents using Document[number] notation.


  {join(documents, delimiter=new_line+new_line, pattern=''Document[$idx]: $content'')}


  Question: {query}


  Answer: '
output_parser:
  # note that we use the same notation as for defining components in pipeline YAMLs. This enables us to reuse our instanciation mechanism and should be a nice balance between flexibility and intuitivity
  type: RegexAnswerParser # for now the only output parser we support, we can add additional ones, e.g. JSONAnswerParser, or DocumentParsers if we need them
  params:
    reference_pattern: Document\\[([^\\]]+)\\] # group captures the document index which is used to resolve the source document to populate document_ids
    pattern: [^\\n]+$ # matches only the last line
```

### How did you test it?
- added tests / existing tests

### :warning: Notes for the reviewer
- This is the PR to merge into `main` (this one will be merged)
- First we need to merge https://github.com/deepset-ai/haystack/pull/4378/ as this PR builds on top of it
- For easier reviewing the review should be done on https://github.com/deepset-ai/haystack/pull/4458 as it merges into the branch of https://github.com/deepset-ai/haystack/pull/4378/ and thus has the correct diffs. 

### 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 one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- [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:
Change prompt at query time for pipelines
**Is your feature request related to a problem? Please describe.**
Currently it's already possible to change the prompt at query time when `PromptNode` is used outside of pipelines via `prompt_template_name` kwarg.
We also want to enable this when `PromptNode` is used in a pipeline to enable faster iterations on prompt engineering.

**Describe the solution you'd like**
Add a parameter to `pipeline.run` and `pipeline.run_batch` to override the used prompt or PromptTemplate. With the new syntax it would be most powerful to only pass the prompt while reusing the rest of the PromptTemplate (e.g. output_shapers)

**Describe alternatives you've considered**
As PromptTemplates are fully serializable, passing a (serialized) PromptTemplate would also be an option, but probably harder to use (Shaper syntax will be required).

**Additional context**
Add any other context or screenshots about the feature request here.

Hints from issue discussion:

Commit message: Merge branch 'main' into feat/prompt_at_query_time

Patch to review:
diff --git a/haystack/agents/base.py b/haystack/agents/base.py
index 572346886d..e6af407fc5 100644
--- a/haystack/agents/base.py
+++ b/haystack/agents/base.py
@@ -163,7 +163,12 @@ def __init__(
             )
         )
         self.prompt_node = prompt_node
-        self.prompt_template = prompt_node.get_prompt_template(prompt_template)
+        resolved_prompt_template = prompt_node.get_prompt_template(prompt_template)
+        if not resolved_prompt_template:
+            raise ValueError(
+                f"Prompt template '{prompt_template}' not found. Please check the spelling of the template name."
+            )
+        self.prompt_template = resolved_prompt_template
         self.tools = {tool.name: tool for tool in tools} if tools else {}
         self.tool_names = ", ".join(self.tools.keys())
         self.tool_names_with_descriptions = "\n".join(
diff --git a/haystack/nodes/prompt/prompt_node.py b/haystack/nodes/prompt/prompt_node.py
index 956713666a..1ac7235b8c 100644
--- a/haystack/nodes/prompt/prompt_node.py
+++ b/haystack/nodes/prompt/prompt_node.py
@@ -4,10 +4,12 @@
 import logging
 from abc import ABC
 import os
+import re
 from typing import Dict, List, Optional, Tuple, Union, Any, Iterator, Type, overload
 from uuid import uuid4
 
 import torch
+import yaml
 
 from haystack import MultiLabel
 from haystack.environment import HAYSTACK_PROMPT_TEMPLATE_ALLOWED_FUNCTIONS
@@ -239,7 +241,9 @@ class PromptTemplate(BasePromptTemplate, ABC):
     [PromptNode](https://docs.haystack.deepset.ai/docs/prompt_node).
     """
 
-    def __init__(self, name: str, prompt_text: str, output_parser: Optional[BaseOutputParser] = None):
+    def __init__(
+        self, name: str, prompt_text: str, output_parser: Optional[Union[BaseOutputParser, Dict[str, Any]]] = None
+    ):
         """
          Creates a PromptTemplate instance.
 
@@ -247,6 +251,12 @@ def __init__(self, name: str, prompt_text: str, output_parser: Optional[BaseOutp
         :param prompt_text: The prompt text, including prompt parameters.
         :param output_parser: A parser that will be applied to the model output.
                 For example, if you want to convert the model output to an Answer object, you can use `AnswerParser`.
+                Instead of BaseOutputParser instances, you can also pass dictionaries defining the output parsers. For example:
+                ```
+                output_parser=[
+                    {"type": "AnswerParser", "params": {"pattern": "Answer: (.*)"}},
+                ]
+                ```
         """
         super().__init__()
 
@@ -279,7 +289,13 @@ def __init__(self, name: str, prompt_text: str, output_parser: Optional[BaseOutp
             **{k: v for k, v in globals().items() if k in PROMPT_TEMPLATE_ALLOWED_FUNCTIONS},
             **PROMPT_TEMPLATE_SPECIAL_CHAR_ALIAS,
         }
-        self.output_parser = output_parser
+        self.output_parser: Optional[BaseOutputParser] = None
+        if isinstance(output_parser, BaseOutputParser):
+            self.output_parser = output_parser
+        elif isinstance(output_parser, dict):
+            output_parser_type = output_parser["type"]
+            output_parser_params = output_parser.get("params", {})
+            self.output_parser = BaseComponent._create_instance(output_parser_type, output_parser_params)
 
     @property
     def output_variable(self) -> Optional[str]:
@@ -724,10 +740,10 @@ def __call__(self, *args, **kwargs) -> List[Any]:
             sa(documents=[Document("I am in love and I feel great!")])
         ```
         """
-        if "prompt_template_name" in kwargs:
-            prompt_template_name = kwargs["prompt_template_name"]
-            kwargs.pop("prompt_template_name")
-            return self.prompt(prompt_template_name, *args, **kwargs)
+        if "prompt_template" in kwargs:
+            prompt_template = kwargs["prompt_template"]
+            kwargs.pop("prompt_template")
+            return self.prompt(prompt_template, *args, **kwargs)
         else:
             return self.prompt(self.default_prompt_template, *args, **kwargs)
 
@@ -745,18 +761,11 @@ def prompt(self, prompt_template: Optional[Union[str, PromptTemplate]], *args, *
         results = []
         # we pop the prompt_collector kwarg to avoid passing it to the model
         prompt_collector: List[Union[str, List[Dict[str, str]]]] = kwargs.pop("prompt_collector", [])
-        if isinstance(prompt_template, str) and not self.is_supported_template(prompt_template):
-            raise ValueError(
-                f"{prompt_template} not supported, please select one of: {self.get_prompt_template_names()} "
-                f"or pass a PromptTemplate instance for prompting."
-            )
 
         # kwargs override model kwargs
         kwargs = {**self._prepare_model_kwargs(), **kwargs}
-        prompt_template_used = prompt_template or self.default_prompt_template
-        if prompt_template_used:
-            template_to_fill = self.get_prompt_template(prompt_template_used)
-
+        template_to_fill = self.get_prompt_template(prompt_template)
+        if template_to_fill:
             # prompt template used, yield prompts from inputs args
             for prompt in template_to_fill.fill(*args, **kwargs):
                 kwargs_copy = copy.copy(kwargs)
@@ -840,19 +849,48 @@ def is_supported_template(self, prompt_template: Union[str, PromptTemplate]) ->
         template_name = prompt_template if isinstance(prompt_template, str) else prompt_template.name
         return template_name in self.prompt_templates
 
-    def get_prompt_template(self, prompt_template: Union[str, PromptTemplate, None]) -> PromptTemplate:
+    def get_prompt_template(self, prompt_template: Union[str, PromptTemplate, None] = None) -> Optional[PromptTemplate]:
         """
-        Returns a prompt template by name.
-        :param prompt_template_name: The name of the prompt template to be returned.
-        :return: The prompt template object.
+        Resolves a prompt template.
+
+        :param prompt_template: The prompt template to be resolved. You can choose between the following types:
+            - None: returns the default prompt template
+            - PromptTemplate: returns the given prompt template object
+            - str: parses the string depending on its content:
+                - prompt template name: returns the prompt template registered with the given name
+                - prompt template yaml: returns a prompt template specified by the given yaml
+                - prompt text: returns a copy of the default prompt template with the given prompt text
+
+            :return: The prompt template object.
         """
+        prompt_template = prompt_template or self.default_prompt_template
+        if prompt_template is None:
+            return None
+
         if isinstance(prompt_template, PromptTemplate):
             return prompt_template
 
-        if not isinstance(prompt_template, str) or prompt_template not in self.prompt_templates:
-            raise ValueError(f"Prompt template {prompt_template} not supported")
+        if isinstance(prompt_template, str) and prompt_template in self.prompt_templates:
+            return self.prompt_templates[prompt_template]
 
-        return self.prompt_templates[prompt_template]
+        # if it's not a string or looks like a prompt template name
+        if not isinstance(prompt_template, str) or re.fullmatch(r"[-a-zA-Z0-9_]+", prompt_template):
+            raise ValueError(
+                f"{prompt_template} not supported, please select one of: {self.get_prompt_template_names()} or pass a PromptTemplate instance for prompting."
+            )
+
+        if "prompt_text:" in prompt_template:
+            prompt_template_parsed = yaml.safe_load(prompt_template)
+            if isinstance(prompt_template_parsed, dict):
+                return PromptTemplate(**prompt_template_parsed)
+
+        # it's a prompt_text
+        prompt_text = prompt_template
+        output_parser: Optional[BaseOutputParser] = None
+        default_prompt_template = self.get_prompt_template()
+        if default_prompt_template:
+            output_parser = default_prompt_template.output_parser
+        return PromptTemplate(name="custom-at-query-time", prompt_text=prompt_text, output_parser=output_parser)
 
     def prompt_template_params(self, prompt_template: str) -> List[str]:
         """
@@ -873,6 +911,7 @@ def run(
         documents: Optional[List[Document]] = None,
         meta: Optional[dict] = None,
         invocation_context: Optional[Dict[str, Any]] = None,
+        prompt_template: Optional[Union[str, PromptTemplate]] = None,
     ) -> Tuple[Dict, str]:
         """
         Runs the PromptNode on these inputs parameters. Returns the output of the prompt model.
@@ -890,6 +929,13 @@ def run(
         :param meta: PromptNode usually ignores meta information, unless it's used as a parameter in the
         PromptTemplate.
         :param invocation_context: The invocation context to be used for the prompt.
+        :param prompt_template: The prompt template to use. You can choose between the following types:
+            - None: use default prompt template
+            - PromptTemplate: use the given prompt template object
+            - str: parses the string depending on its content:
+                - prompt template name: uses the prompt template registered with the given name
+                - prompt template yaml: uses the prompt template specified by the given yaml
+                - prompt text: uses a copy of the default prompt template with the given prompt text
         """
         # prompt_collector is an empty list, it's passed to the PromptNode that will fill it with the rendered prompts,
         # so that they can be returned by `run()` as part of the pipeline's debug output.
@@ -911,11 +957,13 @@ def run(
         if meta and "meta" not in invocation_context.keys():
             invocation_context["meta"] = meta
 
-        results = self(prompt_collector=prompt_collector, **invocation_context)
+        if "prompt_template" not in invocation_context.keys():
+            invocation_context["prompt_template"] = self.get_prompt_template(prompt_template)
 
-        prompt_template = self.get_prompt_template(self.default_prompt_template)
-        output_variable = self.output_variable or prompt_template.output_variable or "results"
+        results = self(prompt_collector=prompt_collector, **invocation_context)
 
+        prompt_template_resolved: PromptTemplate = invocation_context.pop("prompt_template")
+        output_variable = self.output_variable or prompt_template_resolved.output_variable or "results"
         invocation_context[output_variable] = results
         invocation_context["prompts"] = prompt_collector
         final_result: Dict[str, Any] = {output_variable: results, "invocation_context": invocation_context}
@@ -930,6 +978,7 @@ def run_batch(  # type: ignore
         queries: Optional[List[str]] = None,
         documents: Optional[Union[List[Document], List[List[Document]]]] = None,
         invocation_contexts: Optional[List[Dict[str, Any]]] = None,
+        prompt_templates: Optional[List[Union[str, PromptTemplate]]] = None,
     ):
         """
         Runs PromptNode in batch mode.
@@ -949,16 +998,24 @@ def run_batch(  # type: ignore
         :param queries: List of queries.
         :param documents: Single list of Documents or list of lists of Documents in which to search for the answers.
         :param invocation_contexts: List of invocation contexts.
+        :param prompt_templates: The prompt templates to use. You can choose between the following types:
+            - None: use default prompt template
+            - PromptTemplate: use the given prompt template object
+            - str: parses the string depending on its content:
+                - prompt template name: uses the prompt template registered with the given name
+                - prompt template yaml: uses the prompt template specified by the given yaml
+                - prompt text: uses a copy of the default prompt template with the given prompt text
         """
-        prompt_template = self.get_prompt_template(self.default_prompt_template)
-        output_variable = self.output_variable or prompt_template.output_variable or "results"
-
-        inputs = PromptNode._flatten_inputs(queries, documents, invocation_contexts)
+        inputs = PromptNode._flatten_inputs(queries, documents, invocation_contexts, prompt_templates)
         all_results: Dict[str, List] = defaultdict(list)
-        for query, docs, invocation_context in zip(
-            inputs["queries"], inputs["documents"], inputs["invocation_contexts"]
+        for query, docs, invocation_context, prompt_template in zip(
+            inputs["queries"], inputs["documents"], inputs["invocation_contexts"], inputs["prompt_templates"]
         ):
-            results = self.run(query=query, documents=docs, invocation_context=invocation_context)[0]
+            prompt_template = self.get_prompt_template(self.default_prompt_template)
+            output_variable = self.output_variable or prompt_template.output_variable or "results"
+            results = self.run(
+                query=query, documents=docs, invocation_context=invocation_context, prompt_template=prompt_template
+            )[0]
             all_results[output_variable].append(results[output_variable])
             all_results["invocation_contexts"].append(results["invocation_context"])
             if self.debug:
@@ -975,6 +1032,7 @@ def _flatten_inputs(
         queries: Optional[List[str]] = None,
         documents: Optional[Union[List[Document], List[List[Document]]]] = None,
         invocation_contexts: Optional[List[Dict[str, Any]]] = None,
+        prompt_templates: Optional[List[Union[str, PromptTemplate]]] = None,
     ) -> Dict[str, List]:
         """Flatten and copy the queries, documents, and invocation contexts into lists of equal length.
 
@@ -997,6 +1055,7 @@ def _flatten_inputs(
         # Check that queries, and invocation_contexts are of the same length if provided
         input_queries: List[Any]
         input_invocation_contexts: List[Any]
+        input_prompt_templates: List[Any]
         if queries is not None and invocation_contexts is not None:
             if len(queries) != len(invocation_contexts):
                 raise ValueError("The input variables queries and invocation_contexts should have the same length.")
@@ -1012,37 +1071,59 @@ def _flatten_inputs(
             input_queries = [None]
             input_invocation_contexts = [None]
 
+        if prompt_templates is not None:
+            if len(prompt_templates) != len(input_queries):
+                raise ValueError("The input variables prompt_templates and queries should have the same length.")
+            input_prompt_templates = prompt_templates
+        else:
+            input_prompt_templates = [None] * len(input_queries)
+
         multi_docs_list = isinstance(documents, list) and len(documents) > 0 and isinstance(documents[0], list)
         single_docs_list = isinstance(documents, list) and len(documents) > 0 and isinstance(documents[0], Document)
 
         # Docs case 1: single list of Documents
         # -> apply each query (and invocation_contexts) to all Documents
-        inputs: Dict[str, List] = {"queries": [], "invocation_contexts": [], "documents": []}
+        inputs: Dict[str, List] = defaultdict(list)
         if documents is not None:
             if single_docs_list:
-                for query, invocation_context in zip(input_queries, input_invocation_contexts):
+                for query, invocation_context, prompt_template in zip(
+                    input_queries, input_invocation_contexts, input_prompt_templates
+                ):
                     for doc in documents:
                         inputs["queries"].append(query)
                         inputs["invocation_contexts"].append(invocation_context)
                         inputs["documents"].append([doc])
+                        inputs["prompt_templates"].append(prompt_template)
             # Docs case 2: list of lists of Documents
             # -> apply each query (and invocation_context) to corresponding list of Documents,
             # if queries contains only one query, apply it to each list of Documents
             elif multi_docs_list:
                 total_queries = input_queries.copy()
                 total_invocation_contexts = input_invocation_contexts.copy()
-                if len(total_queries) == 1 and len(total_invocation_contexts) == 1:
+                total_prompt_templates = input_prompt_templates.copy()
+                if len(total_queries) == 1 and len(total_invocation_contexts) == 1 and len(total_prompt_templates) == 1:
                     total_queries = input_queries * len(documents)
                     total_invocation_contexts = input_invocation_contexts * len(documents)
-                if len(total_queries) != len(documents) or len(total_invocation_contexts) != len(documents):
+                    total_prompt_templates = input_prompt_templates * len(documents)
+                if (
+                    len(total_queries) != len(documents)
+                    or len(total_invocation_contexts) != len(documents)
+                    or len(total_prompt_templates) != len(documents)
+                ):
                     raise ValueError("Number of queries must be equal to number of provided Document lists.")
-                for query, invocation_context, cur_docs in zip(total_queries, total_invocation_contexts, documents):
+                for query, invocation_context, prompt_template, cur_docs in zip(
+                    total_queries, total_invocation_contexts, total_prompt_templates, documents
+                ):
                     inputs["queries"].append(query)
                     inputs["invocation_contexts"].append(invocation_context)
                     inputs["documents"].append(cur_docs)
-        elif queries is not None or invocation_contexts is not None:
-            for query, invocation_context in zip(input_queries, input_invocation_contexts):
+                    inputs["prompt_templates"].append(prompt_template)
+        elif queries is not None or invocation_contexts is not None or prompt_templates is not None:
+            for query, invocation_context, prompt_template in zip(
+                input_queries, input_invocation_contexts, input_prompt_templates
+            ):
                 inputs["queries"].append(query)
                 inputs["invocation_contexts"].append(invocation_context)
                 inputs["documents"].append([None])
+                inputs["prompt_templates"].append(prompt_template)
         return inputs
diff --git a/test/conftest.py b/test/conftest.py
index caf5d8dbd3..e1b451243c 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -378,7 +378,7 @@ def __init__(self):
     def prompt(self, prompt_template: Optional[Union[str, PromptTemplate]], *args, **kwargs) -> List[str]:
         return [""]
 
-    def get_prompt_template(self, prompt_template: Union[str, PromptTemplate, None]) -> PromptTemplate:
+    def get_prompt_template(self, prompt_template: Union[str, PromptTemplate, None]) -> Optional[PromptTemplate]:
         if prompt_template == "think-step-by-step":
             return PromptTemplate(
                 name="think-step-by-step",
diff --git a/test/prompt/test_prompt_node.py b/test/prompt/test_prompt_node.py
index c8632b43a7..d4d28cab86 100644
--- a/test/prompt/test_prompt_node.py
+++ b/test/prompt/test_prompt_node.py
@@ -260,15 +260,17 @@ def test_run_invalid_template(prompt_node):
 
 @pytest.mark.integration
 def test_invalid_prompting(prompt_node):
-    with pytest.raises(ValueError, match="Hey there, what is the best city in the worl"):
-        prompt_node.prompt(
-            "Hey there, what is the best city in the world?" "Hey there, what is the best city in the world?"
-        )
-
     with pytest.raises(ValueError, match="Hey there, what is the best city in the"):
         prompt_node.prompt(["Hey there, what is the best city in the world?", "Hey, answer me!"])
 
 
+@pytest.mark.integration
+def test_prompt_at_query_time(prompt_node: PromptNode):
+    results = prompt_node.prompt("Hey there, what is the best city in the world?")
+    assert len(results) == 1
+    assert isinstance(results[0], str)
+
+
 @pytest.mark.integration
 def test_invalid_state_ops(prompt_node):
     with pytest.raises(ValueError, match="Prompt template no_such_task_exists"):
@@ -499,6 +501,122 @@ def test_pipeline_with_qa_with_references(prompt_model):
     )
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("prompt_model", ["openai", "azure"], indirect=True)
+def test_pipeline_with_prompt_text_at_query_time(prompt_model):
+    skip_test_for_invalid_key(prompt_model)
+    node = PromptNode(prompt_model, default_prompt_template="question-answering-with-references", top_k=1)
+
+    pipe = Pipeline()
+    pipe.add_node(component=node, name="prompt_node", inputs=["Query"])
+    result = pipe.run(
+        query="Who lives in Berlin?",  # this being a string instead of a list what is being tested
+        documents=[
+            Document("My name is Carla and I live in Berlin", id="1"),
+            Document("My name is Christelle and I live in Paris", id="2"),
+        ],
+        params={
+            "prompt_template": "Create a concise and informative answer (no more than 50 words) for a given question based solely on the given documents. Cite the documents using Document[number] notation.\n\n{join(documents, delimiter=new_line+new_line, pattern='Document[$idx]: $content')}\n\nQuestion: {query}\n\nAnswer: "
+        },
+    )
+
+    assert len(result["answers"]) == 1
+    assert "carla" in result["answers"][0].answer.casefold()
+
+    assert result["answers"][0].document_ids == ["1"]
+    assert (
+        result["answers"][0].meta["prompt"]
+        == "Create a concise and informative answer (no more than 50 words) for a given question based solely on the given documents. Cite the documents using Document[number] notation.\n\n"
+        "Document[1]: My name is Carla and I live in Berlin\n\nDocument[2]: My name is Christelle and I live in Paris\n\n"
+        "Question: Who lives in Berlin?\n\nAnswer: "
+    )
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("prompt_model", ["openai", "azure"], indirect=True)
+def test_pipeline_with_prompt_template_at_query_time(prompt_model):
+    skip_test_for_invalid_key(prompt_model)
+    node = PromptNode(prompt_model, default_prompt_template="question-answering-with-references", top_k=1)
+
+    prompt_template_yaml = """
+            name: "question-answering-with-references-custom"
+            prompt_text: 'Create a concise and informative answer (no more than 50 words) for
+                a given question based solely on the given documents. Cite the documents using Doc[number] notation.
+
+
+                {join(documents, delimiter=new_line+new_line, pattern=''Doc[$idx]: $content'')}
+
+
+                Question: {query}
+
+
+                Answer: '
+            output_parser:
+                type: AnswerParser
+                params:
+                    reference_pattern: Doc\\[([^\\]]+)\\]
+        """
+
+    pipe = Pipeline()
+    pipe.add_node(component=node, name="prompt_node", inputs=["Query"])
+    result = pipe.run(
+        query="Who lives in Berlin?",  # this being a string instead of a list what is being tested
+        documents=[
+            Document("My name is Carla and I live in Berlin", id="doc-1"),
+            Document("My name is Christelle and I live in Paris", id="doc-2"),
+        ],
+        params={"prompt_template": prompt_template_yaml},
+    )
+
+    assert len(result["answers"]) == 1
+    assert "carla" in result["answers"][0].answer.casefold()
+
+    assert result["answers"][0].document_ids == ["doc-1"]
+    assert (
+        result["answers"][0].meta["prompt"]
+        == "Create a concise and informative answer (no more than 50 words) for a given question based solely on the given documents. Cite the documents using Doc[number] notation.\n\n"
+        "Doc[1]: My name is Carla and I live in Berlin\n\nDoc[2]: My name is Christelle and I live in Paris\n\n"
+        "Question: Who lives in Berlin?\n\nAnswer: "
+    )
+
+
+@pytest.mark.integration
+def test_pipeline_with_prompt_template_and_nested_shaper_yaml(tmp_path):
+    with open(tmp_path / "tmp_config_with_prompt_template.yml", "w") as tmp_file:
+        tmp_file.write(
+            f"""
+            version: ignore
+            components:
+            - name: template_with_nested_shaper
+              type: PromptTemplate
+              params:
+                name: custom-template-with-nested-shaper
+                prompt_text: "Given the context please answer the question. Context: {{documents}}; Question: {{query}}; Answer: "
+                output_parser:
+                  type: AnswerParser
+            - name: p1
+              params:
+                model_name_or_path: google/flan-t5-small
+                default_prompt_template: template_with_nested_shaper
+              type: PromptNode
+            pipelines:
+            - name: query
+              nodes:
+              - name: p1
+                inputs:
+                - Query
+        """
+        )
+    pipeline = Pipeline.load_from_yaml(path=tmp_path / "tmp_config_with_prompt_template.yml")
+    result = pipeline.run(query="What is an amazing city?", documents=[Document("Berlin is an amazing city.")])
+    answer = result["answers"][0].answer
+    assert any(word for word in ["berlin", "germany", "population", "city", "amazing"] if word in answer.casefold())
+    assert (
+        result["answers"][0].meta["prompt"]
+        == "Given the context please answer the question. Context: Berlin is an amazing city.; Question: What is an amazing city?; Answer: "
+    )
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("prompt_model", ["hf"], indirect=True)
 def test_prompt_node_no_debug(prompt_model):


Code review comments:
diff_hunk: @@ -239,14 +241,22 @@ class PromptTemplate(BasePromptTemplate, ABC):
     [PromptNode](https://docs.haystack.deepset.ai/docs/prompt_node).
     """
 
-    def __init__(self, name: str, prompt_text: str, output_parser: Optional[BaseOutputParser] = None):
+    def __init__(
+        self, name: str, prompt_text: str, output_parser: Optional[Union[BaseOutputParser, Dict[str, Any]]] = None
+    ):
         """
          Creates a PromptTemplate instance.
 
         :param name: The name of the prompt template (for example, sentiment-analysis, question-generation). You can specify your own name but it must be unique.
         :param prompt_text: The prompt text, including prompt parameters.
         :param output_parser: A parser that will be applied to the model output.
                 For example, if you want to convert the model output to an Answer object, you can use `AnswerParser`.
+                Instead of BaseOutputParser instances, you can also pass dictionaries defining the output parsers. For example:
+                ```
+                output_parser=[
+                    {"type": "AnswerParser", "params": {"pattern": "Answer: (.*)"}},
+                ]
path: haystack/nodes/prompt/prompt_node.py
text: @author:
```suggestion
                output_parser={"type": "AnswerParser", "params": {"pattern": "Answer: (.*)"}},
```