Repository: deepset-ai/haystack

Issue:
Pipeline - `include_outputs_from` parameter of `run` method has unexpected behavior
**To Reproduce**
```python
from haystack.components.fetchers import LinkContentFetcher
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
from haystack.components.websearch.serper_dev import SerperDevWebSearch

from haystack.utils import Secret
from haystack import Pipeline

from haystack import Document, component, logging
import os
from typing import List, Dict, Any, Union

os.environ["SERPERDEV_API_KEY"] = "..."


web_rag_template = """
Answer the question using the provided snippets.
Question: {{question}}

Snippets:
{% for doc in documents %}
- snippet: "{{doc.content}}"
{% endfor %}

chosen option:
"""

web_rag_pipeline = Pipeline()
web_rag_pipeline.add_component("websearch", SerperDevWebSearch(top_k=3))
web_rag_pipeline.add_component("prompt_builder", PromptBuilder(template=web_rag_template))
web_rag_pipeline.add_component("generator", OpenAIGenerator())
web_rag_pipeline.connect("websearch.documents", "prompt_builder.documents")
web_rag_pipeline.connect("prompt_builder", "generator")

result = web_rag_pipeline.run(data={"websearch": {"query": "What did OpenAI yesterday?"}},
                                      include_outputs_from=["websearch", "generator", "prompt_builder"])

print(result)
```

```
{'websearch': {'links': ['https://abcnews.go.com/Business/openai-announces-new-version-ai-language-model-fuels/story?id=110167092', 'https://www.businessinsider.com/openai-event-live-updates-sam-altman-announcement-chatgpt-news-2024-5', 'https://qz.com/openai-chat-gpt-updates-sam-altman-magic-1851472690']}, 
'generator': {'replies': ['OpenAI announced its new flagship model ChatGPT-4o on Monday, an AI chatbot that can see, hear, and have real-time conversations.'], 'meta': [{'model': 'gpt-3.5-turbo-0125', 'index': 0, 'finish_reason': 'stop', 'usage': {'completion_tokens': 32, 'prompt_tokens': 137, 'total_tokens': 169}}]}, 
'prompt_builder': {'prompt': '\nAnswer the question using the provided snippets.\nQuestion: \n\nSnippets:\n\n- snippet: "OpenAI announced a desktop version of its conversation bot ChatGPT on Monday as well as the latest iteration of the AI language model that fuels ..."\n\n- snippet: "But the big news on Monday was OpenAI\'s new flagship AI model, GPT-4o, which will be free to all users and "can reason across audio, vision, and ..."\n\n- snippet: "OpenAI on Monday announced its new flagship model ChatGPT-4o — an AI chatbot that can see, hear, and have real-time conversations."\n\n\nchosen option:'}}
```

`SerperDevWebSearch` component has 2 outputs: `links` and `documents`.
`documents`, connected to the `PromptBuilder` is not returned in the Pipeline output.

**Expected behavior**
I would expect `documents` to be returned in the Pipeline output.

**System:**
 - Haystack version (commit or version number): 2.1.1


Discussion:
coveralls: ## Pull Request Test Coverage Report for [Build 9077086779](https://coveralls.io/builds/67475680)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.006%**) to **90.414%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/67475680/badge)](https://coveralls.io/builds/67475680) |
| :-- | --: |
| Change from base [Build 9074873557](https://coveralls.io/builds/67472294): |  0.006% |
| Covered Lines: | 6536 |
| Relevant Lines: | 7229 |

---
##### 💛  - [Coveralls](https://coveralls.io)

vblagoje: Shall we include this for 2.1.2? 

Patch:
diff --git a/haystack/components/evaluators/llm_evaluator.py b/haystack/components/evaluators/llm_evaluator.py
index 1b623403..39f8653b 100644
--- a/haystack/components/evaluators/llm_evaluator.py
+++ b/haystack/components/evaluators/llm_evaluator.py
@@ -87,7 +87,9 @@ class LLMEvaluator:
         self.api_key = api_key
 
         if api == "openai":
-            self.generator = OpenAIGenerator(api_key=api_key)
+            self.generator = OpenAIGenerator(
+                api_key=api_key, generation_kwargs={"response_format": {"type": "json_object"}}
+            )
         else:
             raise ValueError(f"Unsupported API: {api}")
 
diff --git a/haystack/components/extractors/named_entity_extractor.py b/haystack/components/extractors/named_entity_extractor.py
index a8c6c15b..93fe20ec 100644
--- a/haystack/components/extractors/named_entity_extractor.py
+++ b/haystack/components/extractors/named_entity_extractor.py
@@ -197,7 +197,7 @@ class NamedEntityExtractor:
         """
         return default_to_dict(
             self,
-            backend=self._backend.type,
+            backend=self._backend.type.name,
             model=self._backend.model_name,
             device=self._backend.device.to_dict(),
             pipeline_kwargs=self._backend._pipeline_kwargs,
@@ -217,6 +217,7 @@ class NamedEntityExtractor:
             init_params = data["init_parameters"]
             if init_params["device"] is not None:
                 init_params["device"] = ComponentDevice.from_dict(init_params["device"])
+            init_params["backend"] = NamedEntityExtractorBackend[init_params["backend"]]
             return default_from_dict(cls, data)
         except Exception as e:
             raise DeserializationError(f"Couldn't deserialize {cls.__name__} instance") from e
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index 3680faa3..2224f8dc 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -23,7 +23,7 @@ class Pipeline(PipelineBase):
     """
 
     # TODO: We're ignoring these linting rules for the time being, after we properly optimize this function we'll remove the noqa
-    def run(  # noqa: C901, PLR0912, PLR0915 pylint: disable=too-many-branches
+    def run(  # noqa: C901, PLR0912, PLR0915 pylint: disable=too-many-branches,too-many-locals
         self, data: Dict[str, Any], debug: bool = False, include_outputs_from: Optional[Set[str]] = None
     ) -> Dict[str, Any]:
         """
@@ -431,7 +431,16 @@ class Pipeline(PipelineBase):
 
             if len(include_outputs_from) > 0:
                 for name, output in extra_outputs.items():
-                    if name not in final_outputs:
+                    inner = final_outputs.get(name)
+                    if inner is None:
                         final_outputs[name] = output
+                    else:
+                        # Let's not override any keys that are already
+                        # in the final_outputs as they might be different
+                        # from what we cached in extra_outputs, e.g. when loops
+                        # are involved.
+                        for k, v in output.items():
+                            if k not in inner:
+                                inner[k] = v
 
             return final_outputs
diff --git a/releasenotes/notes/force-valid-JSON-OpeanAI-LLM-based-evaluators-64816e68f137739b.yaml b/releasenotes/notes/force-valid-JSON-OpeanAI-LLM-based-evaluators-64816e68f137739b.yaml
new file mode 100644
index 00000000..19e1ddb1
--- /dev/null
+++ b/releasenotes/notes/force-valid-JSON-OpeanAI-LLM-based-evaluators-64816e68f137739b.yaml
@@ -0,0 +1,6 @@
+---
+
+enhancements:
+  - |
+    Enforce JSON mode on OpenAI LLM-based evaluators so that the they always return valid JSON output.
+    This is to ensure that the output is always in a consistent format, regardless of the input.
diff --git a/releasenotes/notes/named-entity-extractor-serde-improvements-28b594be5a38f175.yaml b/releasenotes/notes/named-entity-extractor-serde-improvements-28b594be5a38f175.yaml
new file mode 100644
index 00000000..af37a9c3
--- /dev/null
+++ b/releasenotes/notes/named-entity-extractor-serde-improvements-28b594be5a38f175.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Fixed (de)serialization of NamedEntityExtractor. Includes updated tests verifying these fixes when NamedEntityExtractor is used in pipelines.
diff --git a/releasenotes/notes/pipeline-run-fix-extra-outputs-a6c750a91faaa8fd.yaml b/releasenotes/notes/pipeline-run-fix-extra-outputs-a6c750a91faaa8fd.yaml
new file mode 100644
index 00000000..00d23f64
--- /dev/null
+++ b/releasenotes/notes/pipeline-run-fix-extra-outputs-a6c750a91faaa8fd.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    The `include_outputs_from` parameter in `Pipeline.run` correctly returns outputs of components with multiple outputs.


Test patch:
diff --git a/test/components/extractors/test_named_entity_extractor.py b/test/components/extractors/test_named_entity_extractor.py
index d47ae69d..180bee41 100644
--- a/test/components/extractors/test_named_entity_extractor.py
+++ b/test/components/extractors/test_named_entity_extractor.py
@@ -3,12 +3,11 @@
 # SPDX-License-Identifier: Apache-2.0
 import pytest
 
-from haystack import ComponentError, DeserializationError
+from haystack import ComponentError, DeserializationError, Pipeline
 from haystack.components.extractors import NamedEntityExtractor, NamedEntityExtractorBackend
 from haystack.utils.device import ComponentDevice
 
 
-@pytest.mark.unit
 def test_named_entity_extractor_backend():
     _ = NamedEntityExtractor(backend=NamedEntityExtractorBackend.HUGGING_FACE, model="dslim/bert-base-NER")
 
@@ -22,7 +21,6 @@ def test_named_entity_extractor_backend():
         NamedEntityExtractor(backend="random_backend", model="dslim/bert-base-NER")
 
 
-@pytest.mark.unit
 def test_named_entity_extractor_serde():
     extractor = NamedEntityExtractor(
         backend=NamedEntityExtractorBackend.HUGGING_FACE,
@@ -42,7 +40,20 @@ def test_named_entity_extractor_serde():
         _ = NamedEntityExtractor.from_dict(serde_data)
 
 
-@pytest.mark.unit
+# tests for NamedEntityExtractor serialization/deserialization in a pipeline
+def test_named_entity_extractor_pipeline_serde(tmp_path):
+    extractor = NamedEntityExtractor(backend=NamedEntityExtractorBackend.HUGGING_FACE, model="dslim/bert-base-NER")
+    p = Pipeline()
+    p.add_component(instance=extractor, name="extractor")
+
+    with open(tmp_path / "test_pipeline.yaml", "w") as f:
+        p.dump(f)
+    with open(tmp_path / "test_pipeline.yaml", "r") as f:
+        q = Pipeline.load(f)
+
+    assert p.to_dict() == q.to_dict(), "Pipeline serialization/deserialization with NamedEntityExtractor failed."
+
+
 def test_named_entity_extractor_serde_none_device():
     extractor = NamedEntityExtractor(
         backend=NamedEntityExtractorBackend.HUGGING_FACE, model="dslim/bert-base-NER", device=None
diff --git a/test/core/pipeline/test_intermediate_outputs.py b/test/core/pipeline/test_intermediate_outputs.py
index a0a5a4e5..a241fad6 100644
--- a/test/core/pipeline/test_intermediate_outputs.py
+++ b/test/core/pipeline/test_intermediate_outputs.py
@@ -5,11 +5,23 @@ import logging
 
 from haystack.components.others import Multiplexer
 from haystack.core.pipeline import Pipeline
+from haystack.core.component import component
 from haystack.testing.sample_components import Accumulate, AddFixedValue, Double, Threshold
 
 logging.basicConfig(level=logging.DEBUG)
 
 
+@component
+class DoubleWithOriginal:
+    """
+    Doubles the input value and returns the original value as well.
+    """
+
+    @component.output_types(value=int, original=int)
+    def run(self, value: int):
+        return {"value": value * 2, "original": value}
+
+
 def test_pipeline_intermediate_outputs():
     pipeline = Pipeline()
     pipeline.add_component("first_addition", AddFixedValue(add=2))
@@ -62,3 +74,24 @@ def test_pipeline_with_loops_intermediate_outputs():
         "below_5": {"above": 8},
         "add_three": {"result": 11},
     }
+
+
+def test_pipeline_intermediate_outputs_multiple_output_sockets():
+    pipeline = Pipeline()
+    pipeline.add_component("first_addition", AddFixedValue(add=2))
+    pipeline.add_component("second_addition", AddFixedValue())
+    pipeline.add_component("double", DoubleWithOriginal())
+    pipeline.connect("first_addition", "double")
+    pipeline.connect("double.value", "second_addition")
+
+    results = pipeline.run(
+        {"first_addition": {"value": 1}}, include_outputs_from={"first_addition", "second_addition", "double"}
+    )
+    assert results == {
+        "second_addition": {"result": 7},
+        "first_addition": {"result": 3},
+        "double": {"value": 6, "original": 3},
+    }
+
+    results = pipeline.run({"first_addition": {"value": 1}}, include_outputs_from={"double"})
+    assert results == {"second_addition": {"result": 7}, "double": {"value": 6, "original": 3}}