fix: include components with consumed outputs in pipeline results (#10065)

* include components with consumed outputs in pipeline results

* update test

* update test

* release notes

* Add async pipeline test and update more checks in async pipeline

---------

Co-authored-by: Sebastian Husch Lee <sjrl423@gmail.com>

Author: mrchtr

haystack/core/pipeline/async_pipeline.py

@@ -281,11 +281,11 @@ async def _run_highest_in_isolation(component_name: str) -> AsyncIterator[dict[s
                     receivers=cached_receivers[component_name],
                     include_outputs_from=include_outputs_from,
                 )
-                if pruned:
+                if pruned or component_name in include_outputs_from:
                     pipeline_outputs[component_name] = pruned
 
                 scheduled_components.remove(component_name)
-                if pruned:
+                if pruned or component_name in include_outputs_from:
                     yield {component_name: _deepcopy_with_exceptions(pruned)}
 
             async def _schedule_task(component_name: str) -> None:
@@ -329,7 +329,7 @@ async def _runner():
                         receivers=cached_receivers[component_name],
                         include_outputs_from=include_outputs_from,
                     )
-                    if pruned:
+                    if pruned or component_name in include_outputs_from:
                         pipeline_outputs[component_name] = pruned
 
                     scheduled_components.remove(component_name)

haystack/core/pipeline/pipeline.py

@@ -434,7 +434,7 @@ def run(  # noqa: PLR0915, PLR0912, C901, pylint: disable=too-many-branches
                     include_outputs_from=include_outputs_from,
                 )
 
-                if component_pipeline_outputs:
+                if component_pipeline_outputs or component_name in include_outputs_from:
                     pipeline_outputs[component_name] = deepcopy(component_pipeline_outputs)
                 if self._is_queue_stale(priority_queue):
                     priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits)

releasenotes/notes/fix-include-outputs-from-empty-dict-f60bdc72f56ed435.yaml

@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    Fixed a bug where components explicitly listed in `include_outputs_from` would not appear
+    in the pipeline results if they returned an empty dictionary.
+    Now, any component specified in `include_outputs_from` will be included in the results
+    regardless of whether its output is empty.

test/core/pipeline/test_async_pipeline.py

@@ -3,10 +3,11 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import asyncio
+from typing import Optional
 
 import pytest
 
-from haystack import AsyncPipeline
+from haystack import AsyncPipeline, component
 
 
 def test_async_pipeline_reentrance(waiting_component, spying_tracer):
@@ -43,3 +44,77 @@ async def call_run():
 
     with pytest.raises(RuntimeError, match="Cannot call run\\(\\) from within an async context"):
         asyncio.run(call_run())
+
+
+def test_component_with_empty_dict_as_output_appears_in_results():
+    """Test that components that return an empty dict as output appear in results as an empty dict"""
+
+    @component
+    class Producer:
+        def __init__(self, prefix: str):
+            self.prefix = prefix
+
+        @component.output_types(value=Optional[str])
+        def run(self, text: Optional[str]):
+            return {"value": f"{self.prefix}: {text}"}
+
+        @component.output_types(value=Optional[str])
+        async def run_async(self, text: Optional[str]):
+            return {"value": f"{self.prefix}: {text}"}
+
+    @component
+    class EmptyProcessor:
+        @component.output_types()
+        def run(self, sources: list[str]):
+            # Returns empty dict when sources is empty
+            return {}
+
+        @component.output_types()
+        async def run_async(self, sources: list[str]):
+            # Returns empty dict when sources is empty
+            return {}
+
+    @component
+    class Combiner:
+        @component.output_types(combined=str)
+        def run(self, input_a: Optional[str], input_b: Optional[str]):
+            if input_a is None:
+                input_a = ""
+            if input_b is None:
+                input_b = ""
+            return {"combined": f"{input_a} | {input_b}"}
+
+        @component.output_types(combined=str)
+        async def run_async(self, input_a: Optional[str], input_b: Optional[str]):
+            if input_a is None:
+                input_a = ""
+            if input_b is None:
+                input_b = ""
+            return {"combined": f"{input_a} | {input_b}"}
+
+    pp = AsyncPipeline()
+    pp.add_component("producer_a", Producer("A"))
+    pp.add_component("producer_b", Producer("B"))
+    pp.add_component("empty_processor", EmptyProcessor())
+    pp.add_component("combiner", Combiner())
+
+    pp.connect("producer_a.value", "combiner.input_a")
+    pp.connect("producer_b.value", "combiner.input_b")
+
+    result = pp.run(
+        {"producer_a": {"text": "hello"}, "producer_b": {"text": "world"}, "empty_processor": {"sources": []}},
+        include_outputs_from={"producer_a", "empty_processor", "combiner"},
+    )
+
+    # Producer A should appear in results because it's in include_outputs_from
+    assert "producer_a" in result
+    assert result["producer_a"] == {"value": "A: hello"}
+    # Producer B should NOT appear since it's not in include_outputs_from
+    assert "producer_b" not in result
+    # Combiner should appear in results
+    assert "combiner" in result
+    assert result["combiner"] == {"combined": "A: hello | B: world"}
+    # Empty processor should appear in results even though it returns an empty dict
+    # because it's in include_outputs_from
+    assert "empty_processor" in result
+    assert result["empty_processor"] == {}

test/core/pipeline/test_pipeline.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
 
 import pytest
 
@@ -123,3 +124,59 @@ def run(self):
                 component_visits={"erroring_component": 0},
             )
         assert "Component name: 'erroring_component'" in str(exc_info.value)
+
+    def test_component_with_empty_dict_as_output_appears_in_results(self):
+        """Test that components that return an empty dict as output appear in results as an empty dict"""
+
+        @component
+        class Producer:
+            def __init__(self, prefix: str):
+                self.prefix = prefix
+
+            @component.output_types(value=Optional[str])
+            def run(self, text: Optional[str]):
+                return {"value": f"{self.prefix}: {text}"}
+
+        @component
+        class EmptyProcessor:
+            @component.output_types()
+            def run(self, sources: list[str]):
+                # Returns empty dict when sources is empty
+                return {}
+
+        @component
+        class Combiner:
+            @component.output_types(combined=str)
+            def run(self, input_a: Optional[str], input_b: Optional[str]):
+                if input_a is None:
+                    input_a = ""
+                if input_b is None:
+                    input_b = ""
+                return {"combined": f"{input_a} | {input_b}"}
+
+        pp = Pipeline()
+        pp.add_component("producer_a", Producer("A"))
+        pp.add_component("producer_b", Producer("B"))
+        pp.add_component("empty_processor", EmptyProcessor())
+        pp.add_component("combiner", Combiner())
+
+        pp.connect("producer_a.value", "combiner.input_a")
+        pp.connect("producer_b.value", "combiner.input_b")
+
+        result = pp.run(
+            {"producer_a": {"text": "hello"}, "producer_b": {"text": "world"}, "empty_processor": {"sources": []}},
+            include_outputs_from={"producer_a", "empty_processor", "combiner"},
+        )
+
+        # Producer A should appear in results because it's in include_outputs_from
+        assert "producer_a" in result
+        assert result["producer_a"] == {"value": "A: hello"}
+        # Producer B should NOT appear since it's not in include_outputs_from
+        assert "producer_b" not in result
+        # Combiner should appear in results
+        assert "combiner" in result
+        assert result["combiner"] == {"combined": "A: hello | B: world"}
+        # Empty processor should appear in results even though it returns an empty dict
+        # because it's in include_outputs_from
+        assert "empty_processor" in result
+        assert result["empty_processor"] == {}