diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py index 84753b3191c..101ba682705 100644 --- a/haystack/core/component/types.py +++ b/haystack/core/component/types.py @@ -73,7 +73,8 @@ def is_variadic(self) -> bool: @property def is_mandatory(self) -> bool: """Check if the input is mandatory.""" - return self.default_value == _empty + # Identity, so a default with a custom `__eq__`, such as a DataFrame, is never compared to the sentinel. + return self.default_value is _empty def __post_init__(self) -> None: try: diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index 0784771745c..f2e24f96065 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -27,7 +27,7 @@ PipelineValidationError, ) from haystack.core.pipeline.component_checks import ( - _NO_OUTPUT_PRODUCED, + _NoOutputProduced, all_predecessors_executed, are_all_sockets_ready, can_component_run, @@ -1223,7 +1223,9 @@ def _consume_component_inputs( greedy_inputs_to_remove = set() for socket_name, socket in component["input_sockets"].items(): socket_inputs = component_inputs.get(socket_name, []) - socket_inputs_values = [sock["value"] for sock in socket_inputs if sock["value"] is not _NO_OUTPUT_PRODUCED] + socket_inputs_values = [ + sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _NoOutputProduced) + ] # if we are resuming a component, the inputs are already consumed, so we just return the first input if is_resume: @@ -1546,12 +1548,12 @@ def _write_component_outputs( :param include_outputs_from: Set of component names that should always return an output from the pipeline. """ for receiver_name, sender_socket, receiver_socket, conversion_strategy in receivers: - # We either get the value that was produced by the actor or we use the _NO_OUTPUT_PRODUCED class to indicate + # We either get the value that was produced by the actor or we use a _NoOutputProduced marker to indicate # that the sender did not produce an output for this socket. # This allows us to track if a predecessor already ran but did not produce an output. - value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED) + value = component_outputs.get(sender_socket.name, _NoOutputProduced()) - if value is not _NO_OUTPUT_PRODUCED and conversion_strategy: + if not isinstance(value, _NoOutputProduced) and conversion_strategy: try: value = _convert_value(value=value, conversion_strategy=conversion_strategy) except Exception as e: @@ -1588,7 +1590,8 @@ def _write_component_outputs( ) else: # If the receiver socket is not lazy variadic, it is greedy variadic or non-variadic. - # We overwrite with the new input if it's not _NO_OUTPUT_PRODUCED or if the current value is None. + # We overwrite with the new input if it's not a _NoOutputProduced marker, or if the current value + # is None. _write_to_standard_socket( inputs=inputs, receiver_name=receiver_name, @@ -1861,5 +1864,5 @@ def _write_to_standard_socket( current_value = inputs[receiver_name].get(receiver_socket_name) # Only overwrite if there's no existing value, or we have a new value to provide - if current_value is None or value is not _NO_OUTPUT_PRODUCED: + if current_value is None or not isinstance(value, _NoOutputProduced): inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}] diff --git a/haystack/core/pipeline/breakpoint.py b/haystack/core/pipeline/breakpoint.py index 932c6190ca6..16095981aa8 100644 --- a/haystack/core/pipeline/breakpoint.py +++ b/haystack/core/pipeline/breakpoint.py @@ -13,7 +13,8 @@ from haystack import logging from haystack.core.errors import PipelineInvalidPipelineSnapshotError -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState +from haystack.utils import _deserialize_value_with_schema from haystack.utils.base_serialization import _serialize_with_field_fallback logger = logging.getLogger(__name__) @@ -211,6 +212,48 @@ def _save_pipeline_snapshot( return str(full_path) +def _serialize_internal_inputs(inputs: dict[str, Any]) -> dict[str, Any]: + """ + Serialize the pipeline's internal inputs state, keeping the `sender` of every input intact. + + The `sender` tells a resumed run which inputs came from a predecessor and which from outside the pipeline, and + therefore whether the paused component can be triggered again. + + The inputs a socket received are stored keyed by position rather than as a list, because a list gets a single + schema derived from its first item. A socket with several senders can hold values of different types, for + example a value from one sender and a `_NoOutputProduced` marker from another, and those need one schema + each to survive the round trip. + + :param inputs: The pipeline's internal inputs state. + :returns: A dict of the form `{"serialization_schema": ..., "serialized_data": ...}`. + """ + positional_inputs = { + component_name: { + socket_name: {str(position): entry for position, entry in enumerate(socket_inputs)} + for socket_name, socket_inputs in socket_dict.items() + } + for component_name, socket_dict in inputs.items() + } + + return _serialize_with_field_fallback(positional_inputs, description="the inputs of the current pipeline state") + + +def _deserialize_internal_inputs(serialized_inputs: dict[str, Any]) -> dict[str, Any]: + """ + Restore the pipeline's internal inputs state from a snapshot written by `_serialize_internal_inputs`. + + :param serialized_inputs: The `inputs` of a `PipelineState` whose `inputs_format` is `INTERNAL_INPUTS_FORMAT`. + :returns: The pipeline's internal inputs state. + """ + return { + component_name: { + socket_name: [socket_inputs[position] for position in sorted(socket_inputs, key=int)] + for socket_name, socket_inputs in socket_dict.items() + } + for component_name, socket_dict in _deserialize_value_with_schema(serialized_inputs).items() + } + + def _create_pipeline_snapshot( *, inputs: dict[str, Any], @@ -226,7 +269,8 @@ def _create_pipeline_snapshot( Create a snapshot of the pipeline at the point where the breakpoint was triggered. :param inputs: The current pipeline snapshot inputs. - :param component_inputs: The inputs to the component that triggered the breakpoint. + :param component_inputs: The inputs of the component that triggered the breakpoint, as they were before the + component consumed them. :param break_point: The breakpoint that triggered the snapshot. :param component_visits: The visit count of the component that triggered the breakpoint. :param original_input_data: The original input data. @@ -239,11 +283,8 @@ def _create_pipeline_snapshot( component_name = break_point.component_name transformed_original_input_data = _transform_json_structure(original_input_data) - transformed_inputs = _transform_json_structure({**inputs, component_name: component_inputs}) - serialized_inputs = _serialize_with_field_fallback( - transformed_inputs, description="the inputs of the current pipeline state" - ) + serialized_inputs = _serialize_internal_inputs({**inputs, component_name: component_inputs}) serialized_original_input_data = _serialize_with_field_fallback( transformed_original_input_data, description="original input data for `pipeline.run`" ) @@ -253,7 +294,10 @@ def _create_pipeline_snapshot( return PipelineSnapshot( pipeline_state=PipelineState( - inputs=serialized_inputs, component_visits=component_visits, pipeline_outputs=serialized_pipeline_outputs + inputs=serialized_inputs, + component_visits=component_visits, + pipeline_outputs=serialized_pipeline_outputs, + inputs_format=INTERNAL_INPUTS_FORMAT, ), timestamp=datetime.now(), break_point=break_point, diff --git a/haystack/core/pipeline/component_checks.py b/haystack/core/pipeline/component_checks.py index d05d373338a..3825aafc360 100644 --- a/haystack/core/pipeline/component_checks.py +++ b/haystack/core/pipeline/component_checks.py @@ -4,9 +4,38 @@ from typing import Any -from haystack.core.component.types import InputSocket, _empty +from haystack.core.component.types import InputSocket -_NO_OUTPUT_PRODUCED = _empty + +class _NoOutputProduced: + """ + Marker a socket input holds when its sender ran without producing a value for that socket. + + Test for it with `isinstance(value, _NoOutputProduced)`, never with `==`: comparing runs `__eq__` on the value + being checked, which for a value such as a DataFrame does not return a bool. It carries no state, so instances + are interchangeable and every one of them means the same thing. + + It is serializable because it reaches a pipeline snapshot as part of the pipeline's inputs. + """ + + def __eq__(self, other: object) -> bool: + return isinstance(other, _NoOutputProduced) + + def __hash__(self) -> int: + return hash(_NoOutputProduced) + + def to_dict(self) -> dict[str, Any]: + """ + Serialize the marker. It carries no state, so the payload is empty. + """ + return {} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "_NoOutputProduced": # noqa: ARG003 + """ + Deserialize the marker. + """ + return cls() def can_component_run(component: dict, inputs: dict) -> bool: @@ -103,7 +132,7 @@ def any_socket_value_from_predecessor_received(socket_inputs: list[dict[str, Any :param socket_inputs: Inputs for the component's socket. """ # When sender is None, the input was provided from outside the pipeline. - return any(inp["value"] is not _NO_OUTPUT_PRODUCED and inp["sender"] is not None for inp in socket_inputs) + return any(not isinstance(inp["value"], _NoOutputProduced) and inp["sender"] is not None for inp in socket_inputs) def has_user_input(inputs: dict) -> bool: @@ -142,7 +171,7 @@ def any_socket_input_received(socket_inputs: list[dict]) -> bool: :param socket_inputs: Inputs for the socket. """ - return any(inp["value"] is not _NO_OUTPUT_PRODUCED for inp in socket_inputs) + return any(not isinstance(inp["value"], _NoOutputProduced) for inp in socket_inputs) def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool: @@ -156,7 +185,7 @@ def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inp actual_senders = { sock["sender"] for sock in socket_inputs - if sock["value"] is not _NO_OUTPUT_PRODUCED and sock["sender"] is not None + if not isinstance(sock["value"], _NoOutputProduced) and sock["sender"] is not None } return expected_senders == actual_senders @@ -176,7 +205,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict if ( socket.is_variadic and socket.is_greedy - and any(sock["value"] is not _NO_OUTPUT_PRODUCED for sock in socket_inputs) + and any(not isinstance(sock["value"], _NoOutputProduced) for sock in socket_inputs) ): return True @@ -185,7 +214,7 @@ def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict return True # The socket is not variadic and the only expected input is complete. - return not socket.is_variadic and socket_inputs[0]["value"] is not _NO_OUTPUT_PRODUCED + return not socket.is_variadic and not isinstance(socket_inputs[0]["value"], _NoOutputProduced) def all_predecessors_executed(component: dict, inputs: dict) -> bool: diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py index 227f4a66594..83f80ba2a93 100644 --- a/haystack/core/pipeline/pipeline.py +++ b/haystack/core/pipeline/pipeline.py @@ -21,13 +21,14 @@ from haystack.core.pipeline.breakpoint import ( SnapshotCallback, _create_pipeline_snapshot, + _deserialize_internal_inputs, _save_pipeline_snapshot, _validate_break_point_against_pipeline, _validate_pipeline_snapshot_against_pipeline, ) from haystack.core.pipeline.utils import _deepcopy_with_exceptions from haystack.dataclasses import AsyncStreamingCallbackT, StreamingCallbackT, StreamingChunk, select_streaming_callback -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot from haystack.dataclasses.streaming_chunk import _invoke_streaming_callback from haystack.telemetry import pipeline_running from haystack.utils import _deserialize_value_with_schema @@ -347,6 +348,9 @@ def run( # noqa: PLR0915, PLR0912, C901 include_outputs_from = set() pipeline_outputs: dict[str, Any] = {} + # Set when resuming from a snapshot that predates `INTERNAL_INPUTS_FORMAT` and therefore lost the sender of + # each input. Cleared as soon as the paused component has run. + legacy_resume_component: str | None = None if not pipeline_snapshot: # normalize `data` @@ -362,6 +366,8 @@ def run( # noqa: PLR0915, PLR0912, C901 # We track component visits to decide if a component can run. component_visits = dict.fromkeys(ordered_component_names, 0) + inputs = self._convert_to_internal_format(pipeline_inputs=data) + else: # Validate the pipeline snapshot against the current pipeline graph _validate_pipeline_snapshot_against_pipeline(pipeline_snapshot, self.graph) @@ -369,7 +375,17 @@ def run( # noqa: PLR0915, PLR0912, C901 # Handle resuming the pipeline from a snapshot component_visits = pipeline_snapshot.pipeline_state.component_visits ordered_component_names = pipeline_snapshot.ordered_component_names - data = _deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) + data = _deserialize_value_with_schema(pipeline_snapshot.original_input_data) + + if pipeline_snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT: + inputs = _deserialize_internal_inputs(pipeline_snapshot.pipeline_state.inputs) + else: + # A legacy snapshot lost the sender of each input, so the only thing we can do is treat them as if + # they came from outside the pipeline. The paused component then has to be let through once. + inputs = self._convert_to_internal_format( + pipeline_inputs=_deserialize_value_with_schema(pipeline_snapshot.pipeline_state.inputs) + ) + legacy_resume_component = pipeline_snapshot.break_point.component_name # include_outputs_from from the snapshot when resuming include_outputs_from = pipeline_snapshot.include_outputs_from @@ -391,7 +407,6 @@ def run( # noqa: PLR0915, PLR0912, C901 "haystack.pipeline.execution_mode": "sync", }, ) as span: - inputs = self._convert_to_internal_format(pipeline_inputs=data) priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits) # check if pipeline is blocked before execution @@ -431,10 +446,15 @@ def run( # noqa: PLR0915, PLR0912, C901 component_name, component_visits[component_name] ) - if pipeline_snapshot: - is_resume = pipeline_snapshot.break_point.component_name == component_name - else: - is_resume = False + is_resume = legacy_resume_component == component_name + if is_resume: + # Only the first execution of the paused component needs the legacy handling. Later visits of the + # same component inside a loop receive regular inputs and must consume them normally. + legacy_resume_component = None + + # A snapshot has to store the component's inputs as they were before the component consumed them, so + # that resuming re-triggers the component the same way this run did. + component_inputs_before_consume = inputs.get(component_name, {}) component_inputs = self._consume_component_inputs( component_name=component_name, component=component, inputs=inputs, is_resume=is_resume ) @@ -444,13 +464,6 @@ def run( # noqa: PLR0915, PLR0912, C901 # initialization component_inputs = self._add_missing_input_defaults(component_inputs, component["input_sockets"]) - # Scenario 1: Pipeline snapshot is provided to resume the pipeline at a specific component - # Deserialize the component_inputs if they are passed in the pipeline_snapshot. - # this check will prevent other component_inputs generated at runtime from being deserialized - if pipeline_snapshot and component_name in pipeline_snapshot.pipeline_state.inputs.keys(): - for key, value in component_inputs.items(): - component_inputs[key] = _deserialize_value_with_schema(value) - try: component_outputs = self._run_component( component_name=component_name, @@ -474,7 +487,7 @@ def run( # noqa: PLR0915, PLR0912, C901 # Create a snapshot of the state of the pipeline before the error occurred. pipeline_snapshot = _create_pipeline_snapshot( inputs=_deepcopy_with_exceptions(inputs), - component_inputs=_deepcopy_with_exceptions(component_inputs), + component_inputs=_deepcopy_with_exceptions(component_inputs_before_consume), break_point=saved_break_point, component_visits=component_visits, original_input_data=data, diff --git a/haystack/dataclasses/breakpoints.py b/haystack/dataclasses/breakpoints.py index 70748146a5a..f41f3f335b4 100644 --- a/haystack/dataclasses/breakpoints.py +++ b/haystack/dataclasses/breakpoints.py @@ -8,6 +8,10 @@ from haystack.utils.dataclasses import _warn_on_inplace_mutation +# Value of `PipelineState.inputs_format` used by every snapshot Haystack creates now. See that field for the shape +# each format implies. +INTERNAL_INPUTS_FORMAT = "internal" + @dataclass(frozen=True) class Breakpoint: @@ -53,11 +57,17 @@ class PipelineState: :param component_visits: A dictionary mapping component names to their visit counts. :param inputs: The inputs processed by the pipeline at the time of the snapshot. :param pipeline_outputs: Dictionary containing the final outputs of the pipeline up to the breakpoint. + :param inputs_format: Which format `inputs` are stored in. `"internal"` means each input records the component + that sent it, keyed by the position it arrived in, as in + `{component: {socket: {"0": {"sender": ..., "value": ...}}}}`. `None` marks snapshots taken before Haystack + recorded the sender, which hold one flattened value per socket, `{component: {socket: value}}`, and can only + be resumed on a component's first visit. """ inputs: dict[str, Any] component_visits: dict[str, int] pipeline_outputs: dict[str, Any] + inputs_format: str | None = None def to_dict(self) -> dict[str, Any]: """ diff --git a/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml new file mode 100644 index 00000000000..956dba579a4 --- /dev/null +++ b/releasenotes/notes/fix-pipeline-snapshot-resume-in-loops-0ab593c343026abb.yaml @@ -0,0 +1,47 @@ +--- +upgrade: + - | + ``PipelineSnapshot.pipeline_state.inputs`` changed shape. It used to store one flattened value per socket, + ``{component: {socket: value}}``. It now stores the pipeline's internal inputs, keeping the component that sent + each one and keying the inputs a socket received by their position: + ``{component: {socket: {"0": {"sender": ..., "value": ...}}}}``. Positions are used instead of a list so that + each input carries its own type schema, which a list cannot do. ``PipelineState`` gained an ``inputs_format`` + field recording which of the two shapes a snapshot uses. The same applies to ``BreakpointException.inputs``, + which returns that field. + + You are affected if you read ``pipeline_state.inputs`` (or ``BreakpointException.inputs``) directly, for example + to display or post-process a snapshot. Resuming a pipeline with ``Pipeline.run(pipeline_snapshot=...)`` is not + affected, and neither is code that only passes snapshots around or persists them. + + To adapt, read the input through its position and take its ``value``: + + .. code-block:: python + + inputs = snapshot.pipeline_state.inputs["serialized_data"] + + # before + value = inputs["my_component"]["my_socket"] + + # now + value = inputs["my_component"]["my_socket"]["0"]["value"] + + A socket that received inputs from several senders now has one entry per position, ``"0"``, ``"1"`` and so on, + each recording the ``sender`` that produced it. + + Snapshots written by earlier versions of Haystack have ``inputs_format`` set to ``None`` and keep the flattened + shape, so branch on that field if you need to handle both. +fixes: + - | + Fixed resuming a ``Pipeline`` from a ``pipeline_snapshot`` that was taken on a component's second or later visit, + which failed with ``PipelineComponentsBlockedError: Cannot run pipeline - all components are blocked``. A snapshot + stored only the values of the pipeline's inputs and dropped the information about which component had sent each + one, so on resume every restored input looked like it came from outside the pipeline, and such an input can only + trigger a component on its first visit. Snapshots now record the sender of each input. Snapshots created by + earlier versions of Haystack behave as before, so re-create them to resume anywhere in a looping pipeline. + - | + Fixed a resumed ``Pipeline`` passing malformed inputs to the component the snapshot was taken on, whenever that + component ran more than once after the resume, for example inside a loop. Every visit after the first reused the + handling meant only for the paused visit and skipped the regular input consumption, so a variadic component could + receive a bare value where it expected a list, raising errors such as + ``TypeError: object of type 'int' has no len()`` from a ``BranchJoiner``. This affected snapshots taken at any + visit count, including the first. diff --git a/test/components/agents/test_agent.py b/test/components/agents/test_agent.py index afca9ca070a..aaacacc29e3 100644 --- a/test/components/agents/test_agent.py +++ b/test/components/agents/test_agent.py @@ -2259,7 +2259,7 @@ class FilesProcessor: @component.output_types(processed_files=list[str]) def run(self, files: list[str]) -> dict: if not files: - return {} # _NO_OUTPUT_PRODUCED → blocks AttachmentsBuilder + return {} # _NoOutputProduced → blocks AttachmentsBuilder return {"processed_files": files} @component diff --git a/test/core/component/test_types.py b/test/core/component/test_types.py new file mode 100644 index 00000000000..24f30fc23bd --- /dev/null +++ b/test/core/component/test_types.py @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: 2022-present deepset GmbH +# +# SPDX-License-Identifier: Apache-2.0 + +from pandas import DataFrame + +from haystack.core.component.types import InputSocket + + +def test_input_socket_with_a_default_that_cannot_be_compared_is_not_mandatory(): + """`is_mandatory` uses `is`, because comparing a DataFrame with `==` returns a DataFrame, not True or False.""" + socket = InputSocket(name="value", type=DataFrame, default_value=DataFrame.from_dict([{"value": 42}])) + + assert socket.is_mandatory is False diff --git a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py b/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py index 99b9f5d3091..844c14902bf 100644 --- a/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py +++ b/test/core/pipeline/breakpoints/test_pipeline_breakpoints_loops.py @@ -45,6 +45,53 @@ class CitiesData(BaseModel): cities: list[City] +VALID_CITIES_JSON = json.dumps( + { + "cities": [ + {"name": "Berlin", "country": "Germany", "population": 3850809}, + {"name": "Paris", "country": "France", "population": 2161000}, + {"name": "Lisbon", "country": "Portugal", "population": 504718}, + ] + } +) + + +def build_validation_loop_pipeline(llm: MockChatGenerator) -> Pipeline: + """Build a pipeline that sends invalid replies back to the prompt builder for another attempt.""" + prompt_template = [ + ChatMessage.from_user( + """ + Create a JSON object from the information present in this passage: {{passage}}. + Only use information that is present in the passage. Follow this JSON schema, but only return the + actual instances without any additional schema definition: + {{schema}} + Make sure your response is a dict and not a list. + {% if invalid_replies and error_message %} + You already created the following output in a previous attempt: {{invalid_replies}} + However, this doesn't comply with the format requirements from above and triggered this + Python exception: {{error_message}} + Correct the output and try again. Just return the corrected output without any extra explanations. + {% endif %} + """ + ) + ] + + pipeline = Pipeline(max_runs_per_component=5) + pipeline.add_component( + instance=ChatPromptBuilder(template=prompt_template, required_variables=["passage", "schema"]), + name="prompt_builder", + ) + pipeline.add_component(instance=llm, name="llm") + pipeline.add_component(instance=OutputValidator(pydantic_model=CitiesData), name="output_validator") + + pipeline.connect("prompt_builder.prompt", "llm.messages") + pipeline.connect("llm.replies", "output_validator") + pipeline.connect("output_validator.invalid_replies", "prompt_builder.invalid_replies") + pipeline.connect("output_validator.error_message", "prompt_builder.error_message") + + return pipeline + + class TestPipelineBreakpointsLoops: """ This class contains tests for pipelines with validation loops and breakpoints. @@ -52,49 +99,13 @@ class TestPipelineBreakpointsLoops: @pytest.fixture def validation_loop_pipeline(self): - """Create a pipeline with validation loops for testing.""" - prompt_template = [ - ChatMessage.from_user( - """ - Create a JSON object from the information present in this passage: {{passage}}. - Only use information that is present in the passage. Follow this JSON schema, but only return the - actual instances without any additional schema definition: - {{schema}} - Make sure your response is a dict and not a list. - {% if invalid_replies and error_message %} - You already created the following output in a previous attempt: {{invalid_replies}} - However, this doesn't comply with the format requirements from above and triggered this - Python exception: {{error_message}} - Correct the output and try again. Just return the corrected output without any extra explanations. - {% endif %} - """ - ) - ] + """The first reply already validates, so every component runs exactly once.""" + return build_validation_loop_pipeline(MockChatGenerator(VALID_CITIES_JSON)) - response_json = json.dumps( - { - "cities": [ - {"name": "Berlin", "country": "Germany", "population": 3850809}, - {"name": "Paris", "country": "France", "population": 2161000}, - {"name": "Lisbon", "country": "Portugal", "population": 504718}, - ] - } - ) - - pipeline = Pipeline(max_runs_per_component=5) - pipeline.add_component( - instance=ChatPromptBuilder(template=prompt_template, required_variables=["passage", "schema"]), - name="prompt_builder", - ) - pipeline.add_component(instance=MockChatGenerator(response_json), name="llm") - pipeline.add_component(instance=OutputValidator(pydantic_model=CitiesData), name="output_validator") - - pipeline.connect("prompt_builder.prompt", "llm.messages") - pipeline.connect("llm.replies", "output_validator") - pipeline.connect("output_validator.invalid_replies", "prompt_builder.invalid_replies") - pipeline.connect("output_validator.error_message", "prompt_builder.error_message") - - return pipeline + @pytest.fixture + def retrying_validation_loop_pipeline(self): + """The first reply fails validation, so the loop closes and every component runs twice.""" + return build_validation_loop_pipeline(MockChatGenerator(["this is not JSON", VALID_CITIES_JSON])) BREAKPOINT_COMPONENTS = ["prompt_builder", "llm", "output_validator"] @@ -134,3 +145,31 @@ def test_pipeline_breakpoints_validation_loop( assert cities_data.cities[0].name == "Berlin" assert cities_data.cities[1].name == "Paris" assert cities_data.cities[2].name == "Lisbon" + + @pytest.mark.parametrize("component", BREAKPOINT_COMPONENTS, ids=BREAKPOINT_COMPONENTS) + @pytest.mark.integration + def test_pipeline_breakpoints_second_visit_of_validation_loop( + self, retrying_validation_loop_pipeline, output_directory, component, load_and_resume_pipeline_snapshot + ): + """ + Test that a pipeline paused on a component's second visit of a loop resumes and runs to completion. + + `pytest.raises` also asserts that the loop really closes: if it did not, the component would never reach a + second visit and the breakpoint would not trigger. + """ + data = {"prompt_builder": {"passage": "Berlin, Paris, Lisbon...", "schema": "CitiesData schema"}} + + break_point = Breakpoint(component_name=component, visit_count=1, snapshot_file_path=str(output_directory)) + + with pytest.raises(BreakpointException): + retrying_validation_loop_pipeline.run(data, break_point=break_point) + + result = load_and_resume_pipeline_snapshot( + pipeline=retrying_validation_loop_pipeline, + output_directory=output_directory, + component_name=break_point.component_name, + data=data, + ) + + cities_data = CitiesData.model_validate(json.loads(result["output_validator"]["valid_replies"][0].text)) + assert [city.name for city in cities_data.cities] == ["Berlin", "Paris", "Lisbon"] diff --git a/test/core/pipeline/test_breakpoint.py b/test/core/pipeline/test_breakpoint.py index 956595da3d9..09fdd6bd3fe 100644 --- a/test/core/pipeline/test_breakpoint.py +++ b/test/core/pipeline/test_breakpoint.py @@ -4,23 +4,31 @@ import json import logging +from dataclasses import replace +from typing import Any import pytest from haystack import component +from haystack.components.joiners import BranchJoiner +from haystack.components.routers import ConditionalRouter +from haystack.components.routers.conditional_router import Route from haystack.core.errors import BreakpointException, PipelineInvalidPipelineSnapshotError from haystack.core.pipeline import Pipeline from haystack.core.pipeline.breakpoint import ( HAYSTACK_PIPELINE_SNAPSHOT_SAVE_ENABLED, _create_pipeline_snapshot, + _deserialize_internal_inputs, _is_snapshot_save_enabled, _save_pipeline_snapshot, _transform_json_structure, load_pipeline_snapshot, ) +from haystack.core.pipeline.component_checks import _NoOutputProduced from haystack.dataclasses import ChatMessage -from haystack.dataclasses.breakpoints import Breakpoint, PipelineSnapshot, PipelineState +from haystack.dataclasses.breakpoints import INTERNAL_INPUTS_FORMAT, Breakpoint, PipelineSnapshot, PipelineState from haystack.utils import _deserialize_value_with_schema +from haystack.utils.base_serialization import _serialize_value_with_schema _EMPTY_OBJECT_PAYLOAD = {"serialization_schema": {"type": "object", "properties": {}}, "serialized_data": {}} @@ -101,38 +109,48 @@ def run(self, input_value: str) -> dict[str, str]: # breakpoint on comp2 break_point = Breakpoint(component_name="comp2", visit_count=0, snapshot_file_path=str(tmp_path)) - try: - # run with include_outputs_from to capture intermediate outputs + # run with include_outputs_from to capture intermediate outputs + with pytest.raises(BreakpointException) as exc_info: pipeline.run(data={"comp1": {"input_value": "test"}}, include_outputs_from={"comp1"}, break_point=break_point) - except BreakpointException as e: - # breakpoint should be triggered - assert e.component == "comp2" - # verify snapshot file contains the intermediate outputs - snapshot_files = list(tmp_path.glob("comp2_*.json")) - assert len(snapshot_files) == 1, f"Expected exactly one snapshot file, found {len(snapshot_files)}" + # breakpoint should be triggered + assert exc_info.value.component == "comp2" - snapshot_file = snapshot_files[0] - loaded_snapshot = load_pipeline_snapshot(snapshot_file) + # verify snapshot file contains the intermediate outputs + snapshot_files = list(tmp_path.glob("comp2_*.json")) + assert len(snapshot_files) == 1, f"Expected exactly one snapshot file, found {len(snapshot_files)}" - # verify the snapshot contains the intermediate outputs from comp1 - assert loaded_snapshot.pipeline_state.pipeline_outputs == ( - { - "serialization_schema": { - "type": "object", - "properties": {"comp1": {"type": "object", "properties": {"result": {"type": "string"}}}}, - }, - "serialized_data": {"comp1": {"result": "processed_test"}}, - } - ) + snapshot_file = snapshot_files[0] + loaded_snapshot = load_pipeline_snapshot(snapshot_file) + + # verify the snapshot contains the intermediate outputs from comp1 + assert loaded_snapshot.pipeline_state.pipeline_outputs == ( + { + "serialization_schema": { + "type": "object", + "properties": {"comp1": {"type": "object", "properties": {"result": {"type": "string"}}}}, + }, + "serialized_data": {"comp1": {"result": "processed_test"}}, + } + ) + + # verify the saved inputs record which component sent each one, keyed by the position it arrived in. + # The accompanying schema is asserted in TestCreatePipelineSnapshot. + assert loaded_snapshot.pipeline_state.inputs_format == INTERNAL_INPUTS_FORMAT + assert loaded_snapshot.pipeline_state.inputs["serialized_data"] == { + # comp1 was given its input from outside the pipeline + "comp1": {"input_value": {"0": {"sender": None, "value": "test"}}}, + # comp2 was given its input by comp1, and had not consumed it yet when the breakpoint hit + "comp2": {"input_value": {"0": {"sender": "comp1", "value": "processed_test"}}}, + } - # verify the whole pipeline state contains the expected data - assert loaded_snapshot.pipeline_state.component_visits["comp1"] == 1 - assert loaded_snapshot.pipeline_state.component_visits["comp2"] == 0 - assert "comp1" in loaded_snapshot.include_outputs_from - assert isinstance(loaded_snapshot.break_point, Breakpoint) - assert loaded_snapshot.break_point.component_name == "comp2" - assert loaded_snapshot.break_point.visit_count == 0 + # verify the whole pipeline state contains the expected data + assert loaded_snapshot.pipeline_state.component_visits["comp1"] == 1 + assert loaded_snapshot.pipeline_state.component_visits["comp2"] == 0 + assert "comp1" in loaded_snapshot.include_outputs_from + assert isinstance(loaded_snapshot.break_point, Breakpoint) + assert loaded_snapshot.break_point.component_name == "comp2" + assert loaded_snapshot.break_point.visit_count == 0 @component @@ -142,6 +160,25 @@ def run(self, input_value: str) -> dict[str, str]: return {"result": f"{input_value}_processed"} +@component +class _CountUpTo: + def __init__(self, limit: int) -> None: + self.limit = limit + + @component.output_types(retry=int, done=str) + def run(self, value: int) -> dict[str, Any]: + if value < self.limit: + return {"retry": value + 1} + return {"done": f"finished at {value}"} + + +@component +class _CollectBranches: + @component.output_types(result=str) + def run(self, a: str | None = None, b: str | None = None) -> dict[str, str]: + return {"result": f"a={a} b={b}"} + + def _three_component_pipeline() -> Pipeline: pipeline = Pipeline() pipeline.add_component("comp1", _AppendingComponent()) @@ -152,50 +189,147 @@ def _three_component_pipeline() -> Pipeline: return pipeline -def test_break_point_with_pipeline_snapshot_steps_through_pipeline(): - pipeline = _three_component_pipeline() +def _looping_pipeline() -> Pipeline: + pipeline = Pipeline(max_runs_per_component=20) + pipeline.add_component("joiner", BranchJoiner(int)) + pipeline.add_component("counter", _CountUpTo(limit=5)) + pipeline.connect("joiner.value", "counter.value") + pipeline.connect("counter.retry", "joiner.value") + return pipeline - # run until the breakpoint on comp2 - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - first_snapshot = exc_info.value.pipeline_snapshot - assert first_snapshot is not None - assert first_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 0, "comp3": 0} - # step: resume from the snapshot and pause again at comp3 - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={}, pipeline_snapshot=first_snapshot, break_point=Breakpoint(component_name="comp3")) - second_snapshot = exc_info.value.pipeline_snapshot - assert second_snapshot is not None - assert second_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 1, "comp3": 0} +class TestResumeFromPipelineSnapshot: + def test_break_point_with_pipeline_snapshot_steps_through_pipeline(self): + pipeline = _three_component_pipeline() - # resume from the second snapshot and run to completion - result = pipeline.run(data={}, pipeline_snapshot=second_snapshot) - assert result["comp3"]["result"] == "test_processed_processed_processed" + # run until the breakpoint on comp2 + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + first_snapshot = exc_info.value.pipeline_snapshot + assert first_snapshot is not None + assert first_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 0, "comp3": 0} + # step: resume from the snapshot and pause again at comp3 + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={}, pipeline_snapshot=first_snapshot, break_point=Breakpoint(component_name="comp3")) + second_snapshot = exc_info.value.pipeline_snapshot + assert second_snapshot is not None + assert second_snapshot.pipeline_state.component_visits == {"comp1": 1, "comp2": 1, "comp3": 0} -def test_break_point_on_earlier_component_than_pipeline_snapshot_never_triggers(): - pipeline = _three_component_pipeline() + # resume from the second snapshot and run to completion + result = pipeline.run(data={}, pipeline_snapshot=second_snapshot) + assert result["comp3"]["result"] == "test_processed_processed_processed" - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - snapshot = exc_info.value.pipeline_snapshot + def test_break_point_on_earlier_component_than_pipeline_snapshot_never_triggers(self): + pipeline = _three_component_pipeline() - # comp1 already ran before the snapshot was taken, so a breakpoint on it never triggers - # and the resumed run completes normally - result = pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp1")) - assert result["comp3"]["result"] == "test_processed_processed_processed" + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot + # comp1 already ran before the snapshot was taken, so a breakpoint on it never triggers + # and the resumed run completes normally + result = pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp1")) + assert result["comp3"]["result"] == "test_processed_processed_processed" -def test_break_point_matching_pipeline_snapshot_break_point_raises(): - pipeline = _three_component_pipeline() + def test_break_point_matching_pipeline_snapshot_break_point_raises(self): + pipeline = _three_component_pipeline() - with pytest.raises(BreakpointException) as exc_info: - pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) - snapshot = exc_info.value.pipeline_snapshot + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot - with pytest.raises(PipelineInvalidPipelineSnapshotError, match="different component or visit count"): - pipeline.run(data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp2", visit_count=0)) + with pytest.raises(PipelineInvalidPipelineSnapshotError, match="different component or visit count"): + pipeline.run( + data={}, pipeline_snapshot=snapshot, break_point=Breakpoint(component_name="comp2", visit_count=0) + ) + + @pytest.mark.parametrize("visit_count", [0, 1, 2, 3]) + def test_break_point_in_loop_resumes_on_any_visit(self, visit_count): + """A component paused on a later visit of a loop must still resume and finish the loop.""" + with pytest.raises(BreakpointException) as exc_info: + _looping_pipeline().run( + {"joiner": {"value": 0}}, break_point=Breakpoint(component_name="joiner", visit_count=visit_count) + ) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + assert snapshot.pipeline_state.component_visits["joiner"] == visit_count + + # The loop runs to `_CountUpTo(limit=5)` regardless of where it was paused. + assert _looping_pipeline().run(data={}, pipeline_snapshot=snapshot) == {"counter": {"done": "finished at 5"}} + + def test_snapshot_preserves_sockets_whose_sender_produced_no_output(self): + """A router that leaves a branch inactive puts the `_NoOutputProduced()` sentinel in the pipeline inputs.""" + routes: list[Route] = [ + {"condition": "{{ n > 5 }}", "output": "{{ 'big' }}", "output_name": "big", "output_type": str}, + {"condition": "{{ n <= 5 }}", "output": "{{ 'small' }}", "output_name": "small", "output_type": str}, + ] + pipeline = Pipeline() + pipeline.add_component("router", ConditionalRouter(routes=routes)) + pipeline.add_component("collect", _CollectBranches()) + pipeline.connect("router.big", "collect.a") + pipeline.connect("router.small", "collect.b") + + expected = pipeline.run({"router": {"n": 9}}) + + with pytest.raises(BreakpointException) as exc_info: + pipeline.run({"router": {"n": 9}}, break_point=Breakpoint(component_name="collect")) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + + restored = _deserialize_internal_inputs(snapshot.pipeline_state.inputs) + assert restored["collect"]["b"] == [{"sender": "router", "value": _NoOutputProduced()}] + + assert pipeline.run(data={}, pipeline_snapshot=snapshot) == expected + + +class TestResumeFromLegacyPipelineSnapshot: + def test_resume_from_legacy_snapshot_without_sender_information(self): + """Snapshots taken before the sender was recorded store flattened values and must still resume.""" + pipeline = _three_component_pipeline() + + with pytest.raises(BreakpointException) as exc_info: + pipeline.run(data={"comp1": {"input_value": "test"}}, break_point=Breakpoint(component_name="comp2")) + snapshot = exc_info.value.pipeline_snapshot + assert snapshot is not None + + legacy_inputs = _serialize_value_with_schema( + _transform_json_structure(_deserialize_internal_inputs(snapshot.pipeline_state.inputs)) + ) + assert legacy_inputs["serialized_data"]["comp2"] == {"input_value": "test_processed"} + legacy_snapshot = replace( + snapshot, pipeline_state=replace(snapshot.pipeline_state, inputs=legacy_inputs, inputs_format=None) + ) + + result = pipeline.run(data={}, pipeline_snapshot=legacy_snapshot) + assert result["comp3"]["result"] == "test_processed_processed_processed" + + def test_resume_from_legacy_snapshot_into_a_loop(self): + """ + A legacy snapshot needs its special input handling for the visit it was paused on, and only that visit. + + The loop brings the paused component ordinary inputs again afterwards, which it has to consume the ordinary + way. Keeping the special handling re-reads the restored input on every visit, so the loop never advances. + + The snapshot is written out literally rather than derived from a current one, because that is what a snapshot + left over from an older Haystack looks like: a greedy socket stored the value it had already consumed. + """ + legacy_snapshot = PipelineSnapshot( + pipeline_state=PipelineState( + inputs=_serialize_value_with_schema({"joiner": {"value": [0]}, "counter": {}}), + component_visits={"joiner": 0, "counter": 0}, + pipeline_outputs=_serialize_value_with_schema({}), + inputs_format=None, + ), + break_point=Breakpoint(component_name="joiner", visit_count=0), + original_input_data=_serialize_value_with_schema({"joiner": {"value": 0}}), + ordered_component_names=["counter", "joiner"], + include_outputs_from=set(), + ) + + # The joiner is visited five more times after the resume. + result = _looping_pipeline().run(data={}, pipeline_snapshot=legacy_snapshot) + assert result == {"counter": {"done": "finished at 5"}} class TestCreatePipelineSnapshot: @@ -206,7 +340,7 @@ def test_create_pipeline_snapshot_all_fields(self): snapshot = _create_pipeline_snapshot( inputs={"comp1": {"input_value": [{"sender": None, "value": "test"}]}, "comp2": {}}, - component_inputs={"input_value": "processed_test"}, + component_inputs={"input_value": [{"sender": "comp1", "value": "processed_test"}]}, break_point=break_point, component_visits={"comp1": 1, "comp2": 0}, original_input_data={"comp1": {"input_value": "test"}}, @@ -225,16 +359,32 @@ def test_create_pipeline_snapshot_all_fields(self): assert snapshot.ordered_component_names == ordered_component_names assert snapshot.break_point == break_point assert snapshot.include_outputs_from == include_outputs_from + + # Each input a socket received is stored under its position, so that it carries its own schema. + def socket_schema(sender_type: str) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "0": { + "type": "object", + "properties": {"sender": {"type": sender_type}, "value": {"type": "string"}}, + } + }, + } + assert snapshot.pipeline_state == PipelineState( inputs={ "serialization_schema": { "type": "object", "properties": { - "comp1": {"type": "object", "properties": {"input_value": {"type": "string"}}}, - "comp2": {"type": "object", "properties": {"input_value": {"type": "string"}}}, + "comp1": {"type": "object", "properties": {"input_value": socket_schema("null")}}, + "comp2": {"type": "object", "properties": {"input_value": socket_schema("string")}}, }, }, - "serialized_data": {"comp1": {"input_value": "test"}, "comp2": {"input_value": "processed_test"}}, + "serialized_data": { + "comp1": {"input_value": {"0": {"sender": None, "value": "test"}}}, + "comp2": {"input_value": {"0": {"sender": "comp1", "value": "processed_test"}}}, + }, }, component_visits={"comp1": 1, "comp2": 0}, pipeline_outputs={ @@ -244,6 +394,7 @@ def test_create_pipeline_snapshot_all_fields(self): }, "serialized_data": {"comp1": {"result": "processed_test"}}, }, + inputs_format=INTERNAL_INPUTS_FORMAT, ) def test_create_pipeline_snapshot_with_dataclasses_in_pipeline_outputs(self): @@ -281,6 +432,7 @@ def test_create_pipeline_snapshot_with_dataclasses_in_pipeline_outputs(self): "comp1": {"result": {"role": "user", "meta": {}, "name": None, "content": [{"text": "hello"}]}} }, }, + inputs_format=INTERNAL_INPUTS_FORMAT, ) def test_create_pipeline_snapshot_non_serializable_inputs(self, caplog): @@ -337,7 +489,7 @@ def to_dict(self): # The non-serializable comp1 field is omitted while the serializable siblings are preserved. assert "comp1" not in deserialized_inputs - assert deserialized_inputs["comp2"] == {"input_value": "keep me"} + assert deserialized_inputs["comp2"] == {"input_value": {"0": {"sender": None, "value": "keep me"}}} assert deserialized_inputs["comp3"] == {} # original_input_data and pipeline_outputs degrade to empty-but-valid payloads. assert deserialized_original_input_data == {} diff --git a/test/core/pipeline/test_component_checks.py b/test/core/pipeline/test_component_checks.py index e60e7c2bd10..70ac5564e3b 100644 --- a/test/core/pipeline/test_component_checks.py +++ b/test/core/pipeline/test_component_checks.py @@ -9,7 +9,7 @@ from haystack.core.component.types import GreedyVariadic, InputSocket, OutputSocket, Variadic from haystack.core.pipeline.component_checks import ( - _NO_OUTPUT_PRODUCED, + _NoOutputProduced, all_predecessors_executed, all_socket_predecessors_executed, any_predecessors_provided_input, @@ -24,6 +24,7 @@ has_user_input, is_any_greedy_socket_ready, ) +from haystack.utils.base_serialization import _deserialize_value_with_schema, _serialize_value_with_schema @pytest.fixture @@ -148,8 +149,8 @@ def test_component_missing_mandatory_input(self, basic_component): assert can_component_run(basic_component, inputs) is False # We added these tests because a component that returned a pandas dataframe caused the pipeline to fail. - # Previously, we compared the value of the socket using '!=' which leads to an error with dataframes. - # Instead, we use 'is not' to compare with the sentinel value. + # Comparing the socket value with '!=' errors out on dataframes, so the checks test the sentinel by type + # instead, which never compares the value at all. def test_sockets_with_ambiguous_truth_value(self, basic_component, greedy_variadic_socket, regular_socket): inputs = {"mandatory_input": [{"sender": "previous_component", "value": DataFrame.from_dict([{"value": 42}])}]} @@ -322,10 +323,10 @@ def test_any_predecessors_provided_input_no_predecessor(self, component_with_mul def test_any_predecessors_provided_input_with_no_output(self, component_with_multiple_sockets): """ - Ensures that _NO_OUTPUT_PRODUCED from a predecessor is ignored in the predecessor detection. + Ensures that _NoOutputProduced() from a predecessor is ignored in the predecessor detection. """ inputs = { - "socket1": [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}], + "socket1": [{"sender": "component1", "value": _NoOutputProduced()}], "socket2": [{"sender": None, "value": "test"}], } assert any_predecessors_provided_input(component_with_multiple_sockets, inputs) is False @@ -345,12 +346,12 @@ class TestSocketValueFromPredecessor: "socket_inputs, expected_result", [ pytest.param([{"sender": "component1", "value": 42}], True, id="valid_input"), - pytest.param([{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}], False, id="no_output"), + pytest.param([{"sender": "component1", "value": _NoOutputProduced()}], False, id="no_output"), pytest.param([{"sender": None, "value": 42}], False, id="user_input"), pytest.param( [ {"sender": None, "value": 42}, - {"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 100}, ], True, @@ -387,10 +388,10 @@ def test_has_user_input_empty_inputs(self): def test_has_user_input_with_no_output(self): """ - Even if the input value is _NO_OUTPUT_PRODUCED, if sender=None + Even if the input value is _NoOutputProduced(), if sender=None it still counts as user input being provided. """ - inputs = {"socket1": [{"sender": None, "value": _NO_OUTPUT_PRODUCED}]} + inputs = {"socket1": [{"sender": None, "value": _NoOutputProduced()}]} assert has_user_input(inputs) is True @@ -459,18 +460,18 @@ def test_variadic_socket_no_execution(self, variadic_socket_with_senders): class TestSocketInputReceived: def test_any_socket_input_received_with_value(self): - """Checks that if there's a non-_NO_OUTPUT_PRODUCED value, the socket is marked as having input.""" + """Checks that if there's a non-_NoOutputProduced() value, the socket is marked as having input.""" socket_inputs = [{"sender": "component1", "value": 42}] assert any_socket_input_received(socket_inputs) is True def test_any_socket_input_received_with_no_output(self): - """If all inputs are _NO_OUTPUT_PRODUCED, the socket has no effective input.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """If all inputs are _NoOutputProduced(), the socket has no effective input.""" + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert any_socket_input_received(socket_inputs) is False def test_any_socket_input_received_mixed_inputs(self): """A single valid input among many is enough to consider the socket as having input.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, {"sender": "component2", "value": 42}] + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 42}] assert any_socket_input_received(socket_inputs) is True def test_any_socket_input_received_empty_list(self): @@ -478,6 +479,21 @@ def test_any_socket_input_received_empty_list(self): assert any_socket_input_received([]) is False +class TestNoOutputProduced: + """The marker reaches a pipeline snapshot, so it has to survive serialization.""" + + def test_markers_are_interchangeable(self): + """The marker carries no state, so any two of them are the same value.""" + assert _NoOutputProduced() == _NoOutputProduced() + assert len({_NoOutputProduced(), _NoOutputProduced()}) == 1 + + def test_the_marker_survives_serialization(self): + """A pipeline snapshot serializes the inputs, marker included.""" + restored = _deserialize_value_with_schema(_serialize_value_with_schema({"value": _NoOutputProduced()})) + + assert isinstance(restored["value"], _NoOutputProduced) + + class TestLazyVariadicSocket: def test_lazy_variadic_all_inputs_received(self, variadic_socket_with_senders): """Lazy variadic socket is ready only if all named senders provided outputs.""" @@ -490,8 +506,8 @@ def test_lazy_variadic_partial_inputs(self, variadic_socket_with_senders): assert has_lazy_variadic_socket_received_all_inputs(variadic_socket_with_senders, socket_inputs) is False def test_lazy_variadic_with_no_output(self, variadic_socket_with_senders): - """_NO_OUTPUT_PRODUCED from a sender doesn't count as valid input, so it's not fully received.""" - socket_inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}, {"sender": "component2", "value": 42}] + """_NoOutputProduced() from a sender doesn't count as valid input, so it's not fully received.""" + socket_inputs = [{"sender": "component1", "value": _NoOutputProduced()}, {"sender": "component2", "value": 42}] assert has_lazy_variadic_socket_received_all_inputs(variadic_socket_with_senders, socket_inputs) is False def test_lazy_variadic_with_user_input(self, variadic_socket_with_senders): @@ -535,8 +551,8 @@ def test_regular_socket_complete(self, regular_socket): assert has_socket_received_all_inputs(regular_socket, inputs) is True def test_regular_socket_incomplete(self, regular_socket): - """_NO_OUTPUT_PRODUCED means the socket is not complete.""" - inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """_NoOutputProduced() means the socket is not complete.""" + inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(regular_socket, inputs) is False def test_regular_socket_no_inputs(self, regular_socket): @@ -554,8 +570,8 @@ def test_lazy_variadic_socket_partial_inputs(self, lazy_variadic_socket): assert has_socket_received_all_inputs(lazy_variadic_socket, inputs) is False def test_lazy_variadic_socket_with_no_output(self, lazy_variadic_socket): - """A sender that produces _NO_OUTPUT_PRODUCED does not fulfill the lazy socket requirement.""" - inputs = [{"sender": "component1", "value": 42}, {"sender": "component2", "value": _NO_OUTPUT_PRODUCED}] + """A sender that produces _NoOutputProduced() does not fulfill the lazy socket requirement.""" + inputs = [{"sender": "component1", "value": 42}, {"sender": "component2", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(lazy_variadic_socket, inputs) is False def test_greedy_variadic_socket_one_input(self, greedy_variadic_socket): @@ -569,8 +585,8 @@ def test_greedy_variadic_socket_multiple_inputs(self, greedy_variadic_socket): assert has_socket_received_all_inputs(greedy_variadic_socket, inputs) is True def test_greedy_variadic_socket_no_valid_inputs(self, greedy_variadic_socket): - """All _NO_OUTPUT_PRODUCED means the greedy socket is not complete.""" - inputs = [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}] + """All _NoOutputProduced() means the greedy socket is not complete.""" + inputs = [{"sender": "component1", "value": _NoOutputProduced()}] assert has_socket_received_all_inputs(greedy_variadic_socket, inputs) is False @@ -629,8 +645,8 @@ def test_greedy_socket_multiple_inputs_ready(self, complex_component): assert is_any_greedy_socket_ready(complex_component, inputs) is True def test_greedy_socket_not_ready(self, complex_component): - """If the only input is _NO_OUTPUT_PRODUCED, the greedy socket isn't ready.""" - inputs = {"greedy_var": [{"sender": "component1", "value": _NO_OUTPUT_PRODUCED}]} + """If the only input is _NoOutputProduced(), the greedy socket isn't ready.""" + inputs = {"greedy_var": [{"sender": "component1", "value": _NoOutputProduced()}]} assert is_any_greedy_socket_ready(complex_component, inputs) is False def test_greedy_socket_no_inputs(self, complex_component): diff --git a/test/core/pipeline/test_pipeline_base.py b/test/core/pipeline/test_pipeline_base.py index 6387df0b534..df63e1ebc67 100644 --- a/test/core/pipeline/test_pipeline_base.py +++ b/test/core/pipeline/test_pipeline_base.py @@ -21,7 +21,8 @@ PipelineRuntimeError, ) from haystack.core.pipeline import Pipeline -from haystack.core.pipeline.base import _NO_OUTPUT_PRODUCED, ComponentPriority, PipelineBase +from haystack.core.pipeline.base import ComponentPriority, PipelineBase +from haystack.core.pipeline.component_checks import _NoOutputProduced from haystack.core.pipeline.utils import FIFOPriorityQueue from haystack.core.serialization import DeserializationCallbacks from haystack.core.type_utils import ConversionStrategy @@ -887,7 +888,7 @@ def test__find_receivers_from(self): { "variadic_input": [ {"sender": "component1", "value": "test"}, - {"sender": "component2", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "component2", "value": _NoOutputProduced()}, ], "normal_input": [{"sender": "component3", "value": "test"}], }, @@ -1050,7 +1051,7 @@ def test__write_component_outputs_output_pruning( @pytest.mark.parametrize( "output_value", - [42, None, _NO_OUTPUT_PRODUCED, "string_value", 3.14], + [42, None, _NoOutputProduced(), "string_value", 3.14], ids=["int", "none", "no-output", "string", "float"], ) def test__write_component_outputs_different_output_values( @@ -1071,9 +1072,9 @@ def test__write_component_outputs_different_output_values( assert inputs["receiver1"]["input1"] == [{"sender": "sender1", "value": output_value}] def test__write_component_outputs_dont_overwrite_with_no_output(self, regular_output_socket, regular_input_socket): - """Test that existing inputs are not overwritten with _NO_OUTPUT_PRODUCED""" + """Test that existing inputs are not overwritten with _NoOutputProduced()""" receivers = [("receiver1", regular_output_socket, regular_input_socket, None)] - component_outputs = {"output1": _NO_OUTPUT_PRODUCED} + component_outputs = {"output1": _NoOutputProduced()} inputs = {"receiver1": {"input1": [{"sender": "sender1", "value": "keep"}]}} PipelineBase()._write_component_outputs( component_name="sender1", @@ -1328,17 +1329,17 @@ def test_fill_queue(self, mock_get_metadata, mock_calc_priority): {"regular": 42, "greedy": [33], "lazy": [55, 66]}, {"regular": [{"sender": None, "value": 24}]}, # Only non-greedy user input remains ), - # Filtering _NO_OUTPUT_PRODUCED + # Filtering _NoOutputProduced() ( {"input1": InputSocket("input1", int)}, { "input1": [ - {"sender": "comp1", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "comp1", "value": _NoOutputProduced()}, {"sender": "comp2", "value": 42}, - {"sender": "comp2", "value": _NO_OUTPUT_PRODUCED}, + {"sender": "comp2", "value": _NoOutputProduced()}, ] }, - {"input1": 42}, # Should skip _NO_OUTPUT_PRODUCED values + {"input1": 42}, # Should skip _NoOutputProduced() values {}, # All inputs consumed ), ],