yujunjun/haystack
1
0
Fork 0
mirror of https://github.com/deepset-ai/haystack.git synced 2026-01-06 03:57:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Refactor Pipeline.run() (#6729)

Browse Source 
* First rough implementation of refactored run

* Further improve run logic

* Properly handle variadic input in run

* Further work

* Enhance names and add more documentation

* Fix issue with output distribution

* This works

* Enhance run comments

* Mark Multiplexer as greedy

* Remove MergeLoop in favour of Multiplexer in tests

* Remove FirstIntSelector in favour of Multiplexer

* Handle corner when waiting for input is stuck

* Remove unused import

* Handle mutable input data in run and misbehaving components

* Handle run input validation

* Test validation

* Fix pylint

* Fix mypy

* Call warm_up in run to fix tests
This commit is contained in:
Silvano Cerza 2024-01-18 17:53:47 +01:00 committed by GitHub
parent 40a8b2b4a9
commit d4f6531c52
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 381 additions and 440 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
haystack/components/others/multiplexer.py
View File

@ -1,10 +1,10 @@
import sys
import logging
import sys
from typing import Any, Dict
from haystack import component, default_from_dict, default_to_dict
from haystack.components.routers.conditional_router import deserialize_type, serialize_type
from haystack.core.component.types import Variadic
from haystack import component, default_to_dict, default_from_dict
from haystack.components.routers.conditional_router import serialize_type, deserialize_type
if sys.version_info < (3, 10):
from typing_extensions import TypeAlias
@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
@component
class Multiplexer:
is_greedy = True
"""
This component is used to distribute a single value to many components that may need it.
It can take such value from different sources (the user's input, or another component), so

345
haystack/core/pipeline/pipeline.py
View File

@ -1,33 +1,30 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from typing import Optional, Any, Dict, List, Union, TypeVar, Type, Set
import os
import json
import datetime
import logging
import importlib
from pathlib import Path
from copy import deepcopy
import logging
from collections import defaultdict
from copy import copy
from pathlib import Path
from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Type, TypeVar, Union
import networkx # type:ignore
from haystack.core.component import component, Component, InputSocket, OutputSocket
from haystack.core.component import Component, InputSocket, OutputSocket, component
from haystack.core.component.connection import Connection, parse_connect_string
from haystack.core.errors import (
PipelineError,
PipelineConnectError,
PipelineError,
PipelineMaxLoops,
PipelineRuntimeError,
PipelineValidationError,
)
from haystack.core.pipeline.descriptions import find_pipeline_outputs
from haystack.core.pipeline.draw.draw import _draw, RenderingEngines
from haystack.core.pipeline.validation import validate_pipeline_input, find_pipeline_inputs
from haystack.core.component.connection import Connection, parse_connect_string
from haystack.core.pipeline.draw.draw import RenderingEngines, _draw
from haystack.core.pipeline.validation import find_pipeline_inputs
from haystack.core.serialization import component_from_dict, component_to_dict
from haystack.core.type_utils import _type_name
from haystack.core.serialization import component_to_dict, component_from_dict
logger = logging.getLogger(__name__)
@ -422,102 +419,266 @@ class Pipeline:
logger.info("Warming up component %s...", node)
self.graph.nodes[node]["instance"].warm_up()
def run(self, data: Dict[str, Any], debug: bool = False) -> Dict[str, Any]: # pylint: disable=too-many-locals
def _validate_input(self, data: Dict[str, Any]):
"""
Runs the pipeline.
Validates that data:
* Each Component name actually exists in the Pipeline
* Each Component is not missing any input
* Each Component has only one input per input socket, if not variadic
* Each Component doesn't receive inputs that are already sent by another Component
Args:
data: the inputs to give to the input components of the Pipeline.
debug: whether to collect and return debug information.
Returns:
A dictionary with the outputs of the output components of the Pipeline.
Raises:
PipelineRuntimeError: if the any of the components fail or return unexpected output.
Raises ValueError if any of the above is not true.
"""
self._clear_visits_count()
data = validate_pipeline_input(self.graph, input_values=data)
logger.info("Pipeline execution started.")
for component_name, component_inputs in data.items():
if component_name not in self.graph.nodes:
raise ValueError(f"Component named {component_name} not found in the pipeline.")
instance = self.graph.nodes[component_name]["instance"]
for socket_name, socket in instance.__canals_input__.items():
if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
raise ValueError("Missing input for component {component_name}: {socket_name}")
for input_name in component_inputs.keys():
if input_name not in instance.__canals_input__:
raise ValueError(f"Input {input_name} not found in component {component_name}.")
self._debug = {}
if debug:
logger.info("Debug mode ON.")
os.makedirs("debug", exist_ok=True)
for component_name in self.graph.nodes:
instance = self.graph.nodes[component_name]["instance"]
for socket_name, socket in instance.__canals_input__.items():
component_inputs = data.get(component_name, {})
if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
raise ValueError("Missing input for component {component_name}: {socket_name}")
if socket.senders and socket_name in component_inputs and not socket.is_variadic:
raise ValueError(
f"Input {socket_name} for component {component_name} is already sent by {socket.senders}."
)
logger.debug(
"Mandatory connections:\n%s",
"\n".join(
f" - {component}: {', '.join([str(s) for s in sockets])}"
for component, sockets in self._mandatory_connections.items()
),
)
# TODO: We're ignoring this linting rules for the time being, after we properly optimize this function we'll remove the noqa
def run( # noqa: C901, PLR0912 pylint: disable=too-many-branches
self, data: Dict[str, Any], debug: bool = False
) -> Dict[str, Any]:
# NOTE: We're assuming data is formatted like so as of now
# data = {
# "comp1": {"input1": 1, "input2": 2},
# }
#
# TODO: Support also this format:
# data = {
# "input1": 1, "input2": 2,
# }
# TODO: Remove this warmup once we can check reliably whether a component has been warmed up or not
# As of now it's here to make sure we don't have failing tests that assume warm_up() is called in run()
self.warm_up()
# Prepare the inputs buffers and components queue
components_queue: List[str] = []
mandatory_values_buffer: Dict[Connection, Any] = {}
optional_values_buffer: Dict[Connection, Any] = {}
pipeline_output: Dict[str, Dict[str, Any]] = defaultdict(dict)
# Raise if input is malformed in some way
self._validate_input(data)
for node_name, input_data in data.items():
for socket_name, value in input_data.items():
# Make a copy of the input value so components don't need to
# take care of mutability.
value = deepcopy(value)
connection = Connection(
None, None, node_name, self.graph.nodes[node_name]["input_sockets"][socket_name]
)
self._add_value_to_buffers(
value, connection, components_queue, mandatory_values_buffer, optional_values_buffer
)
# NOTE: The above NOTE and TODO are technically not true.
# This implementation of run supports only the first format, but the second format is actually
# never received by this method. It's handled by the `run()` method of the `Pipeline` class
# defined in `haystack/pipeline.py`.
# As of now we're ok with this, but we'll need to merge those two classes at some point.
for component_name, component_inputs in data.items():
if component_name not in self.graph.nodes:
# This is not a component name, it must be the name of one or more input sockets.
# Those are handled in a different way, so we skip them here.
continue
instance = self.graph.nodes[component_name]["instance"]
for component_input, input_value in component_inputs.items():
# Handle mutable input data
data[component_name][component_input] = copy(input_value)
if instance.__canals_input__[component_input].is_variadic:
# Components that have variadic inputs need to receive lists as input.
# We don't want to force the user to always pass lists, so we convert single values to lists here.
# If it's already a list we assume the component takes a variadic input of lists, so we
# convert it in any case.
data[component_name][component_input] = [input_value]
# *** PIPELINE EXECUTION LOOP ***
step = 0
while components_queue: # pylint: disable=too-many-nested-blocks
step += 1
if debug:
self._record_pipeline_step(
step, components_queue, mandatory_values_buffer, optional_values_buffer, pipeline_output
)
last_inputs: Dict[str, Dict[str, Any]] = {**data}
component_name = components_queue.pop(0)
logger.debug("> Queue at step %s: %s %s", step, component_name, components_queue)
self._check_max_loops(component_name)
# Take all components that have at least 1 input not connected or is variadic,
# and all components that have no inputs at all
to_run: List[Tuple[str, Component]] = []
for node_name in self.graph.nodes:
component = self.graph.nodes[node_name]["instance"]
# **** RUN THE NODE ****
if not self._ready_to_run(component_name, mandatory_values_buffer, components_queue):
if len(component.__canals_input__) == 0:
# Component has no input, can run right away
to_run.append((node_name, component))
continue
inputs = {
**self._extract_inputs_from_buffer(component_name, mandatory_values_buffer),
**self._extract_inputs_from_buffer(component_name, optional_values_buffer),
}
outputs = self._run_component(name=component_name, inputs=dict(inputs))
for socket in component.__canals_input__.values():
if not socket.senders or socket.is_variadic:
# Component has at least one input not connected or is variadic, can run right away.
to_run.append((node_name, component))
break
# **** PROCESS THE OUTPUT ****
for socket_name, value in outputs.items():
targets = self._collect_targets(component_name, socket_name)
if not targets:
pipeline_output[component_name][socket_name] = value
else:
for target in targets:
self._add_value_to_buffers(
value, target, components_queue, mandatory_values_buffer, optional_values_buffer
)
# These variables are used to detect when we're stuck in a loop.
# Stuck loops can happen when one or more components are waiting for input but
# no other component is going to run.
# This can happen when a whole branch of the graph is skipped for example.
# When we find that two consecutive iterations of the loop where the waiting_for_input list is the same,
# we know we're stuck in a loop and we can't make any progress.
before_last_waiting_for_input: Optional[Set[str]] = None
last_waiting_for_input: Optional[Set[str]] = None
if debug:
self._record_pipeline_step(
step + 1, components_queue, mandatory_values_buffer, optional_values_buffer, pipeline_output
)
os.makedirs(self._debug_path, exist_ok=True)
with open(self._debug_path / "data.json", "w", encoding="utf-8") as datafile:
json.dump(self._debug, datafile, indent=4, default=str)
pipeline_output["_debug"] = self._debug # type: ignore
# The waiting_for_input list is used to keep track of components that are waiting for input.
waiting_for_input: List[Tuple[str, Component]] = []
logger.info("Pipeline executed successfully.")
return dict(pipeline_output)
# This is what we'll return at the end
final_outputs = {}
while len(to_run) > 0:
name, comp = to_run.pop(0)
if any(socket.is_variadic for socket in comp.__canals_input__.values()) and not getattr( # type: ignore
comp, "is_greedy", False
):
there_are_non_variadics = False
for _, other_comp in to_run:
if not any(socket.is_variadic for socket in other_comp.__canals_input__.values()): # type: ignore
there_are_non_variadics = True
break
if there_are_non_variadics:
if (name, comp) not in waiting_for_input:
waiting_for_input.append((name, comp))
continue
if name in last_inputs and len(comp.__canals_input__) == len(last_inputs[name]): # type: ignore
# This component has all the inputs it needs to run
res = comp.run(**last_inputs[name])
if not isinstance(res, Mapping):
raise PipelineRuntimeError(
f"Component '{name}' didn't return a dictionary. "
"Components must always return dictionaries: check the the documentation."
)
# Reset the waiting for input previous states, we managed to run a component
before_last_waiting_for_input = None
last_waiting_for_input = None
if (name, comp) in waiting_for_input:
# We manage to run this component that was in the waiting list, we can remove it.
# This happens when a component was put in the waiting list but we reached it from another edge.
waiting_for_input.remove((name, comp))
# We keep track of which keys to remove from res at the end of the loop.
# This is done after the output has been distributed to the next components, so that
# we're sure all components that need this output have received it.
to_remove_from_res = set()
for sender_component_name, receiver_component_name, edge_data in self.graph.edges(data=True):
if receiver_component_name == name and edge_data["to_socket"].is_variadic:
# Delete variadic inputs that were already consumed
last_inputs[name][edge_data["to_socket"].name] = []
if name != sender_component_name:
continue
if edge_data["from_socket"].name not in res:
# This output has not been produced by the component, skip it
continue
if receiver_component_name not in last_inputs:
last_inputs[receiver_component_name] = {}
to_remove_from_res.add(edge_data["from_socket"].name)
value = res[edge_data["from_socket"].name]
if edge_data["to_socket"].is_variadic:
if edge_data["to_socket"].name not in last_inputs[receiver_component_name]:
last_inputs[receiver_component_name][edge_data["to_socket"].name] = []
# Add to the list of variadic inputs
last_inputs[receiver_component_name][edge_data["to_socket"].name].append(value)
else:
last_inputs[receiver_component_name][edge_data["to_socket"].name] = value
pair = (receiver_component_name, self.graph.nodes[receiver_component_name]["instance"])
if pair not in waiting_for_input and pair not in to_run:
to_run.append(pair)
res = {k: v for k, v in res.items() if k not in to_remove_from_res}
if len(res) > 0:
final_outputs[name] = res
else:
# This component doesn't have enough inputs so we can't run it yet
if (name, comp) not in waiting_for_input:
waiting_for_input.append((name, comp))
if len(to_run) == 0 and len(waiting_for_input) > 0:
# Check if we're stuck in a loop.
# It's important to check whether previous waitings are None as it could be that no
# Component has actually been run yet.
if (
before_last_waiting_for_input is not None
and last_waiting_for_input is not None
and before_last_waiting_for_input == last_waiting_for_input
):
# Are we actually stuck or there's a lazy variadic waiting for input?
# This is our last resort, if there's no lazy variadic waiting for input
# we're stuck for real and we can't make any progress.
for name, comp in waiting_for_input:
is_variadic = any(socket.is_variadic for socket in comp.__canals_input__.values()) # type: ignore
if is_variadic and not getattr(comp, "is_greedy", False):
break
else:
# We're stuck in a loop for real, we can't make any progress.
# BAIL!
break
if len(waiting_for_input) == 1:
# We have a single component with variadic input waiting for input.
# If we're at this point it means it has been waiting for input for at least 2 iterations.
# This will never run.
# BAIL!
break
# There was a lazy variadic waiting for input, we can run it
waiting_for_input.remove((name, comp))
to_run.append((name, comp))
continue
before_last_waiting_for_input = (
last_waiting_for_input.copy() if last_waiting_for_input is not None else None
)
last_waiting_for_input = {item[0] for item in waiting_for_input}
# Remove from waiting only if there is actually enough input to run
for name, comp in waiting_for_input:
if name not in last_inputs:
last_inputs[name] = {}
# Lazy variadics must be removed only if there's nothing else to run at this stage
is_variadic = any(socket.is_variadic for socket in comp.__canals_input__.values()) # type: ignore
if is_variadic and not getattr(comp, "is_greedy", False):
there_are_only_lazy_variadics = True
for other_name, other_comp in waiting_for_input:
if name == other_name:
continue
there_are_only_lazy_variadics &= any(
socket.is_variadic for socket in other_comp.__canals_input__.values() # type: ignore
) and not getattr(other_comp, "is_greedy", False)
if not there_are_only_lazy_variadics:
continue
# Find the first component that has all the inputs it needs to run
has_enough_inputs = True
for input_socket in comp.__canals_input__.values(): # type: ignore
if input_socket.is_mandatory and input_socket.name not in last_inputs[name]:
has_enough_inputs = False
break
if input_socket.is_mandatory:
continue
if input_socket.name not in last_inputs[name]:
last_inputs[name][input_socket.name] = input_socket.default_value
if has_enough_inputs:
break
waiting_for_input.remove((name, comp))
to_run.append((name, comp))
return final_outputs
def _record_pipeline_step(
self, step, components_queue, mandatory_values_buffer, optional_values_buffer, pipeline_output

25
haystack/testing/sample_components/__init__.py
View File

@ -1,23 +1,22 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.testing.sample_components.accumulate import Accumulate
from haystack.testing.sample_components.add_value import AddFixedValue
from haystack.testing.sample_components.concatenate import Concatenate
from haystack.testing.sample_components.subtract import Subtract
from haystack.testing.sample_components.double import Double
from haystack.testing.sample_components.fstring import FString
from haystack.testing.sample_components.greet import Greet
from haystack.testing.sample_components.hello import Hello
from haystack.testing.sample_components.joiner import StringJoiner, StringListJoiner
from haystack.testing.sample_components.parity import Parity
from haystack.testing.sample_components.remainder import Remainder
from haystack.testing.sample_components.accumulate import Accumulate
from haystack.testing.sample_components.threshold import Threshold
from haystack.testing.sample_components.add_value import AddFixedValue
from haystack.testing.sample_components.repeat import Repeat
from haystack.testing.sample_components.sum import Sum
from haystack.testing.sample_components.greet import Greet
from haystack.testing.sample_components.double import Double
from haystack.testing.sample_components.joiner import StringJoiner, StringListJoiner, FirstIntSelector
from haystack.testing.sample_components.hello import Hello
from haystack.testing.sample_components.text_splitter import TextSplitter
from haystack.testing.sample_components.merge_loop import MergeLoop
from haystack.testing.sample_components.self_loop import SelfLoop
from haystack.testing.sample_components.fstring import FString
from haystack.testing.sample_components.subtract import Subtract
from haystack.testing.sample_components.sum import Sum
from haystack.testing.sample_components.text_splitter import TextSplitter
from haystack.testing.sample_components.threshold import Threshold
__all__ = [
"Concatenate",
@ -27,7 +26,6 @@ __all__ = [
"Accumulate",
"Threshold",
"AddFixedValue",
"MergeLoop",
"Repeat",
"Sum",
"Greet",
@ -36,7 +34,6 @@ __all__ = [
"Hello",
"TextSplitter",
"StringListJoiner",
"FirstIntSelector",
"SelfLoop",
"FString",
]

15
haystack/testing/sample_components/joiner.py
View File

@ -33,18 +33,3 @@ class StringListJoiner:
retval += list_of_strings
return {"output": retval}
@component
class FirstIntSelector:
@component.output_types(output=int)
def run(self, inputs: Variadic[int]):
"""
Take intd from multiple input nodes and return the first one
that is not None. Since `input` is Variadic, we know we'll
receive a List[int].
"""
for inp in inputs: # type: ignore
if inp is not None:
return {"output": inp}
return {}

68
haystack/testing/sample_components/merge_loop.py
View File

@ -1,68 +0,0 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from typing import List, Any, Optional, Dict
import sys
from haystack.core.component import component
from haystack.core.errors import DeserializationError
from haystack.core.serialization import default_to_dict
@component
class MergeLoop:
def __init__(self, expected_type: Any, inputs: List[str]):
component.set_input_types(self, **{input_name: Optional[expected_type] for input_name in inputs})
component.set_output_types(self, value=expected_type)
if expected_type.__module__ == "builtins":
self.expected_type = f"builtins.{expected_type.__name__}"
elif expected_type.__module__ == "typing":
self.expected_type = str(expected_type)
else:
self.expected_type = f"{expected_type.__module__}.{expected_type.__name__}"
self.inputs = inputs
def to_dict(self) -> Dict[str, Any]: # pylint: disable=missing-function-docstring
return default_to_dict(self, expected_type=self.expected_type, inputs=self.inputs)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "MergeLoop": # pylint: disable=missing-function-docstring
if "type" not in data:
raise DeserializationError("Missing 'type' in component serialization data")
if data["type"] != f"{cls.__module__}.{cls.__name__}":
raise DeserializationError(f"Class '{data['type']}' can't be deserialized as '{cls.__name__}'")
init_params = data.get("init_parameters", {})
if "expected_type" not in init_params:
raise DeserializationError("Missing 'expected_type' field in 'init_parameters'")
if "inputs" not in init_params:
raise DeserializationError("Missing 'inputs' field in 'init_parameters'")
module = sys.modules[__name__]
fully_qualified_type_name = init_params["expected_type"]
if fully_qualified_type_name.startswith("builtins."):
module = sys.modules["builtins"]
type_name = fully_qualified_type_name.split(".")[-1]
try:
expected_type = getattr(module, type_name)
except AttributeError as exc:
raise DeserializationError(
f"Can't find type '{type_name}', import '{fully_qualified_type_name}' to fix the issue"
) from exc
inputs = init_params["inputs"]
return cls(expected_type=expected_type, inputs=inputs)
def run(self, **kwargs):
"""
:param kwargs: find the first non-None value and return it.
"""
for value in kwargs.values():
if value is not None:
return {"value": value}
return {}

21
test/core/pipeline/test_complex_pipeline.py
View File

@ -3,27 +3,24 @@
# SPDX-License-Identifier: Apache-2.0
import logging
from haystack.components.others import Multiplexer
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import (
Accumulate,
AddFixedValue,
Double,
Greet,
Parity,
Threshold,
Double,
Sum,
Repeat,
Subtract,
MergeLoop,
Sum,
Threshold,
)
logging.basicConfig(level=logging.DEBUG)
def test_complex_pipeline():
loop_merger = MergeLoop(expected_type=int, inputs=["in_1", "in_2"])
summer = Sum()
pipeline = Pipeline(max_loops_allowed=2)
pipeline.add_component("greet_first", Greet(message="Hello, the value is {value}."))
pipeline.add_component("accumulate_1", Accumulate())
@ -32,12 +29,12 @@ def test_complex_pipeline():
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("accumulate_2", Accumulate())
pipeline.add_component("loop_merger", loop_merger)
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("double", Double())
pipeline.add_component("greet_again", Greet(message="Hello again, now the value is {value}."))
pipeline.add_component("sum", summer)
pipeline.add_component("sum", Sum())
pipeline.add_component("greet_enumerator", Greet(message="Hello from enumerator, here the value became {value}."))
pipeline.add_component("enumerate", Repeat(outputs=["first", "second"]))
@ -64,11 +61,11 @@ def test_complex_pipeline():
pipeline.connect("add_four", "accumulate_3")
pipeline.connect("parity.odd", "add_one.value")
pipeline.connect("add_one", "loop_merger.in_1")
pipeline.connect("loop_merger", "below_10")
pipeline.connect("add_one", "multiplexer.value")
pipeline.connect("multiplexer", "below_10")
pipeline.connect("below_10.below", "double")
pipeline.connect("double", "loop_merger.in_2")
pipeline.connect("double", "multiplexer.value")
pipeline.connect("below_10.above", "accumulate_2")
pipeline.connect("accumulate_2", "diff.second_value")

64
test/core/pipeline/test_distinct_loops_pipeline.py
View File

@ -1,100 +1,104 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, MergeLoop, Remainder, FirstIntSelector
import logging
from pathlib import Path
from haystack.components.others import Multiplexer
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, Remainder
logging.basicConfig(level=logging.DEBUG)
def test_pipeline_equally_long_branches():
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in", "in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("remainder", Remainder(divisor=3))
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("merge.value", "remainder.value")
pipeline.connect("multiplexer.value", "remainder.value")
pipeline.connect("remainder.remainder_is_1", "add_two.value")
pipeline.connect("remainder.remainder_is_2", "add_one.value")
pipeline.connect("add_two", "merge.in_2")
pipeline.connect("add_one", "merge.in_1")
pipeline.connect("add_two", "multiplexer.value")
pipeline.connect("add_one", "multiplexer.value")
results = pipeline.run({"merge": {"in": 0}})
pipeline.draw(Path(__file__).parent / Path(__file__).name.replace(".py", ".png"))
results = pipeline.run({"multiplexer": {"value": 0}})
assert results == {"remainder": {"remainder_is_0": 0}}
results = pipeline.run({"merge": {"in": 3}})
results = pipeline.run({"multiplexer": {"value": 3}})
assert results == {"remainder": {"remainder_is_0": 3}}
results = pipeline.run({"merge": {"in": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"in": 5}})
results = pipeline.run({"multiplexer": {"value": 5}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"in": 6}})
results = pipeline.run({"multiplexer": {"value": 6}})
assert results == {"remainder": {"remainder_is_0": 6}}
def test_pipeline_differing_branches():
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in", "in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("remainder", Remainder(divisor=3))
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("add_two_1", AddFixedValue(add=1))
pipeline.add_component("add_two_2", AddFixedValue(add=1))
pipeline.connect("merge.value", "remainder.value")
pipeline.connect("multiplexer.value", "remainder.value")
pipeline.connect("remainder.remainder_is_1", "add_two_1.value")
pipeline.connect("add_two_1", "add_two_2.value")
pipeline.connect("add_two_2", "merge.in_2")
pipeline.connect("add_two_2", "multiplexer")
pipeline.connect("remainder.remainder_is_2", "add_one.value")
pipeline.connect("add_one", "merge.in_1")
pipeline.connect("add_one", "multiplexer")
results = pipeline.run({"merge": {"in": 0}})
results = pipeline.run({"multiplexer": {"value": 0}})
assert results == {"remainder": {"remainder_is_0": 0}}
results = pipeline.run({"merge": {"in": 3}})
results = pipeline.run({"multiplexer": {"value": 3}})
assert results == {"remainder": {"remainder_is_0": 3}}
results = pipeline.run({"merge": {"in": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"in": 5}})
results = pipeline.run({"multiplexer": {"value": 5}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"in": 6}})
results = pipeline.run({"multiplexer": {"value": 6}})
assert results == {"remainder": {"remainder_is_0": 6}}
def test_pipeline_differing_branches_variadic():
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", FirstIntSelector())
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("remainder", Remainder(divisor=3))
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("add_two_1", AddFixedValue(add=1))
pipeline.add_component("add_two_2", AddFixedValue(add=1))
pipeline.connect("merge", "remainder.value")
pipeline.connect("multiplexer", "remainder.value")
pipeline.connect("remainder.remainder_is_1", "add_two_1.value")
pipeline.connect("add_two_1", "add_two_2.value")
pipeline.connect("add_two_2", "merge.inputs")
pipeline.connect("add_two_2", "multiplexer.value")
pipeline.connect("remainder.remainder_is_2", "add_one.value")
pipeline.connect("add_one", "merge.inputs")
pipeline.connect("add_one", "multiplexer.value")
results = pipeline.run({"merge": {"inputs": 0}})
results = pipeline.run({"multiplexer": {"value": 0}})
assert results == {"remainder": {"remainder_is_0": 0}}
results = pipeline.run({"merge": {"inputs": 3}})
results = pipeline.run({"multiplexer": {"value": 3}})
assert results == {"remainder": {"remainder_is_0": 3}}
results = pipeline.run({"merge": {"inputs": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"inputs": 5}})
results = pipeline.run({"multiplexer": {"value": 5}})
assert results == {"remainder": {"remainder_is_0": 6}}
results = pipeline.run({"merge": {"inputs": 6}})
results = pipeline.run({"multiplexer": {"value": 6}})
assert results == {"remainder": {"remainder_is_0": 6}}

17
test/core/pipeline/test_double_loop_pipeline.py
View File

@ -1,11 +1,12 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Threshold, MergeLoop
import logging
from haystack.components.others import Multiplexer
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Threshold
logging.basicConfig(level=logging.DEBUG)
@ -14,20 +15,20 @@ def test_pipeline(tmp_path):
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in_1", "in_2", "in_3"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("below_5", Threshold(threshold=5))
pipeline.add_component("add_three", AddFixedValue(add=3))
pipeline.add_component("accumulator", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("add_one.result", "merge.in_1")
pipeline.connect("merge.value", "below_10.value")
pipeline.connect("add_one.result", "multiplexer")
pipeline.connect("multiplexer.value", "below_10.value")
pipeline.connect("below_10.below", "accumulator.value")
pipeline.connect("accumulator.value", "below_5.value")
pipeline.connect("below_5.above", "add_three.value")
pipeline.connect("below_5.below", "merge.in_2")
pipeline.connect("add_three.result", "merge.in_3")
pipeline.connect("below_5.below", "multiplexer")
pipeline.connect("add_three.result", "multiplexer")
pipeline.connect("below_10.above", "add_two.value")
pipeline.draw(tmp_path / "double_loop_pipeline.png")

6
test/core/pipeline/test_joiners.py
View File

@ -1,11 +1,11 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import StringJoiner, StringListJoiner, Hello, TextSplitter
import logging
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Hello, StringJoiner, StringListJoiner, TextSplitter
logging.basicConfig(level=logging.DEBUG)

8
test/core/pipeline/test_linear_pipeline.py
View File

@ -1,15 +1,15 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
import logging
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, Double
import logging
logging.basicConfig(level=logging.DEBUG)
def test_pipeline(tmp_path):
def test_pipeline():
pipeline = Pipeline()
pipeline.add_component("first_addition", AddFixedValue(add=2))
pipeline.add_component("second_addition", AddFixedValue())
@ -17,7 +17,5 @@ def test_pipeline(tmp_path):
pipeline.connect("first_addition", "double")
pipeline.connect("double", "second_addition")
pipeline.draw(tmp_path / "linear_pipeline.png")
results = pipeline.run({"first_addition": {"value": 1}})
assert results == {"second_addition": {"result": 7}}

23
test/core/pipeline/test_looping_and_merge_pipeline.py
View File

@ -1,11 +1,12 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Threshold, Sum, FirstIntSelector, MergeLoop
import logging
from haystack.components.others import Multiplexer
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Sum, Threshold
logging.basicConfig(level=logging.DEBUG)
@ -13,18 +14,18 @@ def test_pipeline_fixed():
accumulator = Accumulate()
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("add_zero", AddFixedValue(add=0))
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("sum", Sum())
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("counter", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("add_zero", "merge.in_1")
pipeline.connect("merge", "below_10.value")
pipeline.connect("add_zero", "multiplexer.value")
pipeline.connect("multiplexer", "below_10.value")
pipeline.connect("below_10.below", "add_one.value")
pipeline.connect("add_one.result", "counter.value")
pipeline.connect("counter.value", "merge.in_2")
pipeline.connect("counter.value", "multiplexer.value")
pipeline.connect("below_10.above", "add_two.value")
pipeline.connect("add_two.result", "sum.values")
@ -37,18 +38,18 @@ def test_pipeline_variadic():
accumulator = Accumulate()
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("add_zero", AddFixedValue(add=0))
pipeline.add_component("merge", FirstIntSelector())
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("sum", Sum())
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("counter", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("add_zero", "merge")
pipeline.connect("merge", "below_10.value")
pipeline.connect("add_zero", "multiplexer")
pipeline.connect("multiplexer", "below_10.value")
pipeline.connect("below_10.below", "add_one.value")
pipeline.connect("add_one.result", "counter.value")
pipeline.connect("counter.value", "merge.inputs")
pipeline.connect("counter.value", "multiplexer.value")
pipeline.connect("below_10.above", "add_two.value")
pipeline.connect("add_two.result", "sum.values")

39
test/core/pipeline/test_looping_pipeline.py
View File

@ -1,11 +1,12 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Threshold, MergeLoop, FirstIntSelector
import logging
from haystack.components.others import Multiplexer
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import Accumulate, AddFixedValue, Threshold
logging.basicConfig(level=logging.DEBUG)
@ -14,15 +15,15 @@ def test_pipeline():
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("add_one", AddFixedValue(add=1))
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("accumulator", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("add_one.result", "merge.in_1")
pipeline.connect("merge.value", "below_10.value")
pipeline.connect("add_one.result", "multiplexer")
pipeline.connect("multiplexer.value", "below_10.value")
pipeline.connect("below_10.below", "accumulator.value")
pipeline.connect("accumulator.value", "merge.in_2")
pipeline.connect("accumulator.value", "multiplexer")
pipeline.connect("below_10.above", "add_two.value")
results = pipeline.run({"add_one": {"value": 3}})
@ -34,15 +35,15 @@ def test_pipeline_direct_io_loop():
accumulator = Accumulate()
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("accumulator", accumulator)
pipeline.connect("merge.value", "below_10.value")
pipeline.connect("multiplexer.value", "below_10.value")
pipeline.connect("below_10.below", "accumulator.value")
pipeline.connect("accumulator.value", "merge.in_2")
pipeline.connect("accumulator.value", "multiplexer")
results = pipeline.run({"merge": {"in_1": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"below_10": {"above": 16}}
assert accumulator.state == 16
@ -51,17 +52,17 @@ def test_pipeline_fixed_merger_input():
accumulator = Accumulate()
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", MergeLoop(expected_type=int, inputs=["in_1", "in_2"]))
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("accumulator", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("merge.value", "below_10.value")
pipeline.connect("multiplexer.value", "below_10.value")
pipeline.connect("below_10.below", "accumulator.value")
pipeline.connect("accumulator.value", "merge.in_2")
pipeline.connect("accumulator.value", "multiplexer")
pipeline.connect("below_10.above", "add_two.value")
results = pipeline.run({"merge": {"in_1": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"add_two": {"result": 18}}
assert accumulator.state == 16
@ -70,16 +71,16 @@ def test_pipeline_variadic_merger_input():
accumulator = Accumulate()
pipeline = Pipeline(max_loops_allowed=10)
pipeline.add_component("merge", FirstIntSelector())
pipeline.add_component("multiplexer", Multiplexer(type_=int))
pipeline.add_component("below_10", Threshold(threshold=10))
pipeline.add_component("accumulator", accumulator)
pipeline.add_component("add_two", AddFixedValue(add=2))
pipeline.connect("merge", "below_10.value")
pipeline.connect("multiplexer", "below_10.value")
pipeline.connect("below_10.below", "accumulator.value")
pipeline.connect("accumulator.value", "merge.inputs")
pipeline.connect("accumulator.value", "multiplexer.value")
pipeline.connect("below_10.above", "add_two.value")
results = pipeline.run({"merge": {"inputs": 4}})
results = pipeline.run({"multiplexer": {"value": 4}})
assert results == {"add_two": {"result": 18}}
assert accumulator.state == 16

2
test/core/pipeline/test_mutable_inputs.py
View File

@ -3,8 +3,8 @@
# SPDX-License-Identifier: Apache-2.0
from typing import List
from haystack.core.pipeline import Pipeline
from haystack.core.component import component
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import StringListJoiner

24
test/core/pipeline/test_pipeline.py
View File

@ -1,32 +1,20 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from typing import Optional
import logging
from typing import Optional
import pytest
from haystack.core.pipeline import Pipeline
from haystack.core.component.sockets import InputSocket, OutputSocket
from haystack.core.errors import PipelineMaxLoops, PipelineError, PipelineRuntimeError
from haystack.testing.sample_components import AddFixedValue, Threshold, Double, Sum
from haystack.core.errors import PipelineError, PipelineMaxLoops, PipelineRuntimeError
from haystack.core.pipeline import Pipeline
from haystack.testing.factory import component_class
from haystack.testing.sample_components import AddFixedValue, Double, Sum, Threshold
logging.basicConfig(level=logging.DEBUG)
def test_max_loops():
pipe = Pipeline(max_loops_allowed=10)
pipe.add_component("add", AddFixedValue())
pipe.add_component("threshold", Threshold(threshold=100))
pipe.add_component("sum", Sum())
pipe.connect("threshold.below", "add.value")
pipe.connect("add.result", "sum.values")
pipe.connect("sum.total", "threshold.value")
with pytest.raises(PipelineMaxLoops):
pipe.run({"sum": {"values": 1}})
def test_run_with_component_that_does_not_return_dict():
BrokenComponent = component_class(
"BrokenComponent", input_types={"a": int}, output_types={"b": int}, output=1 # type:ignore
@ -34,9 +22,7 @@ def test_run_with_component_that_does_not_return_dict():
pipe = Pipeline(max_loops_allowed=10)
pipe.add_component("comp", BrokenComponent())
with pytest.raises(
PipelineRuntimeError, match="Component 'comp' returned a value of type 'int' instead of a dict."
):
with pytest.raises(PipelineRuntimeError):
pipe.run({"comp": {"a": 1}})

5
test/core/pipeline/test_self_loop.py
View File

@ -1,12 +1,11 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from haystack.core.component import component
import logging
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, SelfLoop
import logging
logging.basicConfig(level=logging.DEBUG)

24
test/core/pipeline/test_validation_pipeline_io.py
View File

@ -5,12 +5,12 @@ from typing import Optional
import pytest
from haystack.core.pipeline import Pipeline
from haystack.core.component.sockets import InputSocket, OutputSocket
from haystack.core.component.types import Variadic
from haystack.core.errors import PipelineValidationError
from haystack.core.component.sockets import InputSocket, OutputSocket
from haystack.core.pipeline import Pipeline
from haystack.core.pipeline.descriptions import find_pipeline_inputs, find_pipeline_outputs
from haystack.testing.sample_components import Double, AddFixedValue, Sum, Parity
from haystack.testing.sample_components import AddFixedValue, Double, Parity, Sum
def test_find_pipeline_input_no_input():
@ -124,8 +124,8 @@ def test_validate_pipeline_input_pipeline_with_no_inputs():
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
pipe.connect("comp2", "comp1")
with pytest.raises(PipelineValidationError, match="This pipeline has no inputs."):
pipe.run({})
res = pipe.run({})
assert res == {}
def test_validate_pipeline_input_unknown_component():
@ -133,7 +133,7 @@ def test_validate_pipeline_input_unknown_component():
pipe.add_component("comp1", Double())
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
with pytest.raises(ValueError, match=r"Pipeline received data for unknown component\(s\): test_component"):
with pytest.raises(ValueError):
pipe.run({"test_component": {"value": 1}})
@ -142,7 +142,7 @@ def test_validate_pipeline_input_all_necessary_input_is_present():
pipe.add_component("comp1", Double())
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
with pytest.raises(ValueError, match="Missing input: comp1.value"):
with pytest.raises(ValueError):
pipe.run({})
@ -153,7 +153,7 @@ def test_validate_pipeline_input_all_necessary_input_is_present_considering_defa
pipe.connect("comp1", "comp2")
pipe.run({"comp1": {"value": 1}})
pipe.run({"comp1": {"value": 1, "add": 2}})
with pytest.raises(ValueError, match="Missing input: comp1.value"):
with pytest.raises(ValueError):
pipe.run({"comp1": {"add": 3}})
@ -162,7 +162,7 @@ def test_validate_pipeline_input_only_expected_input_is_present():
pipe.add_component("comp1", Double())
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
with pytest.raises(ValueError, match=r"The input value of comp2 is already sent by: \['comp1'\]"):
with pytest.raises(ValueError):
pipe.run({"comp1": {"value": 1}, "comp2": {"value": 2}})
@ -171,7 +171,7 @@ def test_validate_pipeline_input_only_expected_input_is_present_falsy():
pipe.add_component("comp1", Double())
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
with pytest.raises(ValueError, match=r"The input value of comp2 is already sent by: \['comp1'\]"):
with pytest.raises(ValueError):
pipe.run({"comp1": {"value": 1}, "comp2": {"value": 0}})
@ -184,7 +184,7 @@ def test_validate_pipeline_falsy_input_present():
def test_validate_pipeline_falsy_input_missing():
pipe = Pipeline()
pipe.add_component("comp", Double())
with pytest.raises(ValueError, match="Missing input: comp.value"):
with pytest.raises(ValueError):
pipe.run({"comp": {}})
@ -194,7 +194,7 @@ def test_validate_pipeline_input_only_expected_input_is_present_including_unknow
pipe.add_component("comp2", Double())
pipe.connect("comp1", "comp2")
with pytest.raises(ValueError, match="Component comp1 is not expecting any input value called add"):
with pytest.raises(ValueError):
pipe.run({"comp1": {"value": 1, "add": 2}})

2
test/core/pipeline/test_variable_decision_and_merge_pipeline.py
View File

@ -4,7 +4,7 @@
import logging
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, Remainder, Double, Sum
from haystack.testing.sample_components import AddFixedValue, Double, Remainder, Sum
logging.basicConfig(level=logging.DEBUG)

4
test/core/pipeline/test_variable_merging_pipeline.py
View File

@ -1,11 +1,11 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
import logging
from haystack.core.pipeline import Pipeline
from haystack.testing.sample_components import AddFixedValue, Sum
import logging
logging.basicConfig(level=logging.DEBUG)

122
test/core/sample_components/test_merge_loop.py
View File

@ -1,122 +0,0 @@
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0
from typing import Dict
import pytest
from haystack.core.errors import DeserializationError
from haystack.testing.sample_components import MergeLoop
def test_to_dict():
component = MergeLoop(expected_type=int, inputs=["first", "second"])
res = component.to_dict()
assert res == {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "builtins.int", "inputs": ["first", "second"]},
}
def test_to_dict_with_typing_class():
component = MergeLoop(expected_type=Dict, inputs=["first", "second"])
res = component.to_dict()
assert res == {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "typing.Dict", "inputs": ["first", "second"]},
}
def test_to_dict_with_custom_class():
component = MergeLoop(expected_type=MergeLoop, inputs=["first", "second"])
res = component.to_dict()
assert res == {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {
"expected_type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"inputs": ["first", "second"],
},
}
def test_from_dict():
data = {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "builtins.int", "inputs": ["first", "second"]},
}
component = MergeLoop.from_dict(data)
assert component.expected_type == "builtins.int"
assert component.inputs == ["first", "second"]
def test_from_dict_with_typing_class():
data = {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "typing.Dict", "inputs": ["first", "second"]},
}
component = MergeLoop.from_dict(data)
assert component.expected_type == "typing.Dict"
assert component.inputs == ["first", "second"]
def test_from_dict_with_custom_class():
data = {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "sample_components.merge_loop.MergeLoop", "inputs": ["first", "second"]},
}
component = MergeLoop.from_dict(data)
assert component.expected_type == "haystack.testing.sample_components.merge_loop.MergeLoop"
assert component.inputs == ["first", "second"]
def test_from_dict_without_expected_type():
data = {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"inputs": ["first", "second"]},
}
with pytest.raises(DeserializationError) as exc:
MergeLoop.from_dict(data)
exc.match("Missing 'expected_type' field in 'init_parameters'")
def test_from_dict_without_inputs():
data = {
"type": "haystack.testing.sample_components.merge_loop.MergeLoop",
"init_parameters": {"expected_type": "sample_components.merge_loop.MergeLoop"},
}
with pytest.raises(DeserializationError) as exc:
MergeLoop.from_dict(data)
exc.match("Missing 'inputs' field in 'init_parameters'")
def test_merge_first():
component = MergeLoop(expected_type=int, inputs=["in_1", "in_2"])
results = component.run(in_1=5)
assert results == {"value": 5}
def test_merge_second():
component = MergeLoop(expected_type=int, inputs=["in_1", "in_2"])
results = component.run(in_2=5)
assert results == {"value": 5}
def test_merge_nones():
component = MergeLoop(expected_type=int, inputs=["in_1", "in_2", "in_3"])
results = component.run()
assert results == {}
def test_merge_one():
component = MergeLoop(expected_type=int, inputs=["in_1"])
results = component.run(in_1=1)
assert results == {"value": 1}
def test_merge_one_none():
component = MergeLoop(expected_type=int, inputs=[])
results = component.run()
assert results == {}