From ba804feefcc00558b4b2e79225671604c10e85e5 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Thu, 16 Jul 2026 07:19:31 +0200 Subject: [PATCH 1/6] Auto deserialization of dataclasses in pipeline.run --- haystack/tools/component_tool.py | 37 ++--- haystack/utils/__init__.py | 3 +- haystack/utils/deserialization.py | 101 ++++++++++++- ...eline-inputs-utility-d38e32c0ca3b76cd.yaml | 19 +++ test/utils/test_deserialization.py | 133 +++++++++++++++++- 5 files changed, 261 insertions(+), 32 deletions(-) create mode 100644 releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml diff --git a/haystack/tools/component_tool.py b/haystack/tools/component_tool.py index 2e1682a44c3..b2a7a51c561 100644 --- a/haystack/tools/component_tool.py +++ b/haystack/tools/component_tool.py @@ -3,7 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 from collections.abc import Callable -from typing import Any, get_args, get_origin +from typing import Any from pydantic import Field, TypeAdapter, create_model @@ -31,7 +31,7 @@ _serialize_outputs_to_state, _serialize_outputs_to_string, ) -from haystack.utils.type_serialization import _is_union_type +from haystack.utils.deserialization import _deserialize_from_dict, _resolve_from_dict_class logger = logging.getLogger(__name__) @@ -391,31 +391,10 @@ def _convert_param(self, param_value: Any, param_type: type) -> Any: :returns: The converted parameter value. """ - # We unwrap optional types so we can support types like messages: list[ChatMessage] | None - unwrapped_param_type = _unwrap_optional(param_type) - - # We handle union types (e.g. list[ChatMessage] | str) by extracting the list[T] arm that has - # from_dict, so the conversion below works the same as for plain list[T]. Other arms like str - # need no special handling and fall through to Pydantic or the plain return at the end. - if _is_union_type(unwrapped_param_type): - list_arms = [ - a - for a in get_args(unwrapped_param_type) - if get_origin(a) is list and get_args(a) and hasattr(get_args(a)[0], "from_dict") - ] - unwrapped_param_type = list_arms[0] if list_arms else unwrapped_param_type - - # We support calling from_dict on target types that have it, even if they are wrapped in a list. - # This allows us to support lists of dataclasses as well as single dataclass inputs. - target_type = ( - get_args(unwrapped_param_type)[0] if get_origin(unwrapped_param_type) is list else unwrapped_param_type - ) - if hasattr(target_type, "from_dict"): - if isinstance(param_value, list): - return [target_type.from_dict(item) if isinstance(item, dict) else item for item in param_value] - if isinstance(param_value, dict): - return target_type.from_dict(param_value) - return param_value - - # Use the original type for pydantic validation + # Types involving a from_dict-capable class (e.g. list[ChatMessage] | None) are deserialized via + # from_dict; everything else is validated with Pydantic. + target_class = _resolve_from_dict_class(param_type) + if target_class is not None: + return _deserialize_from_dict(param_value, target_class) + return TypeAdapter(param_type).validate_python(param_value) diff --git a/haystack/utils/__init__.py b/haystack/utils/__init__.py index 1095e49acbd..96d61269671 100644 --- a/haystack/utils/__init__.py +++ b/haystack/utils/__init__.py @@ -13,7 +13,7 @@ "base_serialization": ["_deserialize_value_with_schema", "_serialize_value_with_schema"], "callable_serialization": ["deserialize_callable", "serialize_callable"], "device": ["ComponentDevice", "Device", "DeviceMap", "DeviceType"], - "deserialization": ["deserialize_chatgenerator_inplace", "deserialize_component_inplace"], + "deserialization": ["coerce_pipeline_inputs", "deserialize_chatgenerator_inplace", "deserialize_component_inplace"], "filters": ["document_matches_filter"], "jinja2_extensions": ["Jinja2TimeExtension"], "jupyter": ["is_in_jupyter"], @@ -30,6 +30,7 @@ from .base_serialization import _serialize_value_with_schema as _serialize_value_with_schema from .callable_serialization import deserialize_callable as deserialize_callable from .callable_serialization import serialize_callable as serialize_callable + from .deserialization import coerce_pipeline_inputs as coerce_pipeline_inputs from .deserialization import deserialize_chatgenerator_inplace as deserialize_chatgenerator_inplace from .deserialization import deserialize_component_inplace as deserialize_component_inplace from .device import ComponentDevice as ComponentDevice diff --git a/haystack/utils/deserialization.py b/haystack/utils/deserialization.py index 90d2b9fceaa..f91631add55 100644 --- a/haystack/utils/deserialization.py +++ b/haystack/utils/deserialization.py @@ -2,10 +2,14 @@ # # SPDX-License-Identifier: Apache-2.0 -from typing import Any +from typing import TYPE_CHECKING, Any, get_args, get_origin from haystack.core.errors import DeserializationError from haystack.core.serialization import component_from_dict, import_class_by_name +from haystack.utils.type_serialization import _is_union_type + +if TYPE_CHECKING: + from haystack.core.pipeline.base import PipelineBase def deserialize_chatgenerator_inplace(data: dict[str, Any], key: str = "chat_generator") -> None: @@ -54,3 +58,98 @@ def deserialize_component_inplace(data: dict[str, Any], key: str = "chat_generat raise DeserializationError(f"Class '{serialized_component['type']}' not correctly imported") from e data[key] = component_from_dict(cls=component_class, data=serialized_component, name=key) + + +def _resolve_from_dict_class(type_: Any) -> Any: + """ + Resolve the class whose `from_dict` method can deserialize values of `type_`. + + Handles plain classes, `list[T]`, and optionals/unions of those. In unions, the first arm resolving to a + `from_dict`-capable class wins. + + :param type_: The type to inspect. + :returns: The resolved class, or None if `type_` involves no `from_dict`-capable class. + """ + if _is_union_type(type_): + for arm in get_args(type_): + resolved = _resolve_from_dict_class(arm) + if resolved is not None: + return resolved + return None + if get_origin(type_) is list: + args = get_args(type_) + return _resolve_from_dict_class(args[0]) if args else None + return type_ if hasattr(type_, "from_dict") else None + + +def _deserialize_from_dict(value: Any, target_class: Any) -> Any: + """ + Deserialize `value` using `target_class.from_dict`. + + Dictionaries are deserialized directly, lists element-wise (leaving non-dictionary items untouched). Any other + value is returned unchanged. + + :param value: The value to deserialize. + :param target_class: The class providing the `from_dict` method. + :returns: The deserialized value. + """ + if isinstance(value, dict): + return target_class.from_dict(value) + if isinstance(value, list): + return [target_class.from_dict(item) if isinstance(item, dict) else item for item in value] + return value + + +def _coerce_input_value(value: Any, socket_types: list[Any]) -> Any: + for type_ in socket_types: + target_class = _resolve_from_dict_class(type_) + if target_class is not None: + return _deserialize_from_dict(value, target_class) + return value + + +def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> dict[str, Any]: + """ + Deserialize serialized Haystack objects in pipeline input data, based on the pipeline's input socket types. + + For every provided value whose input socket type involves a class exposing a `from_dict` method + (such as `ChatMessage` or `Document`), plain dictionaries are converted into instances of that class. + Socket types of the form `T`, `list[T]`, and optionals/unions of those are supported. Values that are + already deserialized, or whose socket types involve no `from_dict`-capable class, are returned unchanged. + + Like `Pipeline.run`, `data` accepts the nested format (`{"component_name": {"input_name": value}}`) and + the flat format (`{"input_name": value}`). The same format detection rules apply and the returned + dictionary keeps the format of `data`. + + Usage example: + ```python + serialized_inputs = {"prompt_builder": {"template": [message.to_dict() for message in messages]}} + inputs = coerce_pipeline_inputs(pipeline, serialized_inputs) + result = pipeline.run(inputs) + ``` + + :param pipeline: The pipeline whose input socket types drive the coercion. + :param data: The pipeline input data, in nested or flat format. + :returns: A new dictionary with the same structure as `data` and deserialized values. + :raises Exception: Any error raised by a `from_dict` method if a dictionary does not match the expected format. + """ + # mirrors the format detection in Pipeline.run: nested if all values are dictionaries + if all(isinstance(value, dict) for value in data.values()): + available_inputs = pipeline.inputs(include_components_with_connected_inputs=True) + coerced: dict[str, Any] = {} + for component_name, component_inputs in data.items(): + sockets = available_inputs.get(component_name, {}) + coerced[component_name] = { + input_name: _coerce_input_value(value, [sockets[input_name]["type"]] if input_name in sockets else []) + for input_name, value in component_inputs.items() + } + return coerced + + # flat format: input names are matched across components like in Pipeline.run + available_inputs = pipeline.inputs() + return { + input_name: _coerce_input_value( + value, [sockets[input_name]["type"] for sockets in available_inputs.values() if input_name in sockets] + ) + for input_name, value in data.items() + } diff --git a/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml b/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml new file mode 100644 index 00000000000..fe451ff1ac7 --- /dev/null +++ b/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml @@ -0,0 +1,19 @@ +--- +features: + - | + Added the ``coerce_pipeline_inputs`` utility function (importable from ``haystack.utils``). + It deserializes serialized Haystack objects in pipeline input data based on the pipeline's input + socket types: for every provided value whose socket type involves a class exposing ``from_dict`` + (such as ``ChatMessage`` or ``Document``), plain dictionaries are converted into instances of that + class. Socket types of the form ``T``, ``list[T]``, and optionals/unions of those are supported, + and both the nested (``{"component_name": {"input_name": value}}``) and flat + (``{"input_name": value}``) input formats are accepted. This removes the need for API layers to + guess which pipeline inputs contain serialized ``ChatMessage`` objects before calling + ``Pipeline.run``: + + .. code-block:: python + + from haystack.utils import coerce_pipeline_inputs + + inputs = coerce_pipeline_inputs(pipeline, serialized_inputs) + result = pipeline.run(inputs) diff --git a/test/utils/test_deserialization.py b/test/utils/test_deserialization.py index 320f313075d..01738af84e5 100644 --- a/test/utils/test_deserialization.py +++ b/test/utils/test_deserialization.py @@ -3,11 +3,17 @@ # SPDX-License-Identifier: Apache-2.0 +from typing import Any + import pytest +from pydantic import BaseModel +from haystack import Document, Pipeline, component from haystack.components.generators.chat.openai import OpenAIChatGenerator +from haystack.components.joiners import BranchJoiner from haystack.core.errors import DeserializationError -from haystack.utils.deserialization import deserialize_component_inplace +from haystack.dataclasses import ChatMessage +from haystack.utils.deserialization import coerce_pipeline_inputs, deserialize_component_inplace class ChatGeneratorWithoutFromDict: @@ -15,6 +21,27 @@ def to_dict(self): return {"type": "test_deserialization.ChatGeneratorWithoutFromDict"} +@component +class MessageListEcho: + @component.output_types(messages=list[ChatMessage]) + def run(self, messages: list[ChatMessage], top_k: int = 3, config: dict[str, Any] | None = None) -> dict[str, Any]: + return {"messages": messages} + + +@component +class SingleMessageEcho: + @component.output_types(message=ChatMessage) + def run(self, message: ChatMessage, query: str = "") -> dict[str, Any]: + return {"message": message} + + +@component +class DocumentListEcho: + @component.output_types(documents=list[Document]) + def run(self, documents: list[Document] | None = None) -> dict[str, Any]: + return {"documents": documents or []} + + class TestDeserializeComponentInplace: def test_deserialize_component_inplace(self, monkeypatch): monkeypatch.setenv("OPENAI_API_KEY", "test-api-key") @@ -49,3 +76,107 @@ def test_component_no_from_dict_method(self): data = {"chat_generator": chat_generator.to_dict()} deserialize_component_inplace(data) assert isinstance(data["chat_generator"], ChatGeneratorWithoutFromDict) + + +class TestCoercePipelineInputs: + @pytest.fixture + def pipeline(self): + pipe = Pipeline() + pipe.add_component("messages_echo", MessageListEcho()) + pipe.add_component("single_echo", SingleMessageEcho()) + pipe.add_component("documents_echo", DocumentListEcho()) + return pipe + + def test_nested_format(self, pipeline): + messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] + data: dict[str, Any] = {"messages_echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"messages_echo": {"messages": messages, "top_k": 5}} + assert isinstance(data["messages_echo"]["messages"][0], dict) + + def test_flat_format(self, pipeline): + messages = [ChatMessage.from_user("Hi")] + data = {"messages": [message.to_dict() for message in messages], "top_k": 5} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"messages": messages, "top_k": 5} + + def test_single_message_socket(self, pipeline): + message = ChatMessage.from_user("Hi") + data = {"single_echo": {"message": message.to_dict(), "query": "some query"}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"single_echo": {"message": message, "query": "some query"}} + + def test_optional_list_socket(self, pipeline): + documents = [Document(content="Hello"), Document(content="World")] + data = {"documents_echo": {"documents": [document.to_dict() for document in documents]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"documents_echo": {"documents": documents}} + + def test_instances_pass_through(self, pipeline): + messages = [ChatMessage.from_user("Hi")] + data = {"messages_echo": {"messages": messages}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced["messages_echo"]["messages"][0] is messages[0] + + def test_mixed_list(self, pipeline): + message = ChatMessage.from_user("Hi") + data = {"messages_echo": {"messages": [message, ChatMessage.from_assistant("Hello").to_dict()]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced["messages_echo"]["messages"] == [message, ChatMessage.from_assistant("Hello")] + + def test_non_coercible_values_untouched(self, pipeline): + data = {"messages_echo": {"messages": [], "top_k": 5, "config": {"a": 1}}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == data + + def test_unknown_inputs_untouched(self, pipeline): + message_dict = ChatMessage.from_user("Hi").to_dict() + data = {"unknown_input": [message_dict], "top_k": 5} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"unknown_input": [message_dict], "top_k": 5} + + def test_unknown_component_untouched(self, pipeline): + message_dict = ChatMessage.from_user("Hi").to_dict() + data = {"unknown_component": {"messages": [message_dict]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"unknown_component": {"messages": [message_dict]}} + + def test_variadic_socket(self): + pipe = Pipeline() + pipe.add_component("joiner", BranchJoiner(list[ChatMessage])) + messages = [ChatMessage.from_user("Hi")] + data = {"value": [message.to_dict() for message in messages], "unrelated": 1} + + coerced = coerce_pipeline_inputs(pipe, data) + + assert coerced["value"] == messages + + def test_pydantic_model_dump_payload(self, pipeline): + class Response(BaseModel): + messages: list[ChatMessage] + + messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] + dumped = Response(messages=messages).model_dump(mode="json") + data = {"messages_echo": {"messages": dumped["messages"]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"messages_echo": {"messages": messages}} From ca78fb743fcd77bb169b4fb9909aaae6ced525a9 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Tue, 21 Jul 2026 10:01:15 +0200 Subject: [PATCH 2/6] refactoring --- haystack/__init__.py | 3 +- haystack/core/serialization.py | 105 +++++++++- haystack/tools/component_tool.py | 3 +- haystack/utils/__init__.py | 3 +- haystack/utils/deserialization.py | 101 +--------- ...eline-inputs-utility-d38e32c0ca3b76cd.yaml | 4 +- test/core/test_serialization.py | 181 ++++++++++++++++++ test/utils/test_deserialization.py | 133 +------------ 8 files changed, 293 insertions(+), 240 deletions(-) diff --git a/haystack/__init__.py b/haystack/__init__.py index 730f8b19f51..282479f0c3b 100644 --- a/haystack/__init__.py +++ b/haystack/__init__.py @@ -14,7 +14,7 @@ from haystack.core.component import component from haystack.core.errors import ComponentError, DeserializationError from haystack.core.pipeline import Pipeline -from haystack.core.serialization import default_from_dict, default_to_dict +from haystack.core.serialization import coerce_pipeline_inputs, default_from_dict, default_to_dict from haystack.core.super_component.super_component import SuperComponent, super_component from haystack.dataclasses import Answer, Document, ExtractedAnswer, GeneratedAnswer from haystack.version import __version__ # noqa: F401 @@ -34,6 +34,7 @@ "Pipeline", "SuperComponent", "super_component", + "coerce_pipeline_inputs", "component", "default_from_dict", "default_to_dict", diff --git a/haystack/core/serialization.py b/haystack/core/serialization.py index fec9780eedd..3d2a622748a 100644 --- a/haystack/core/serialization.py +++ b/haystack/core/serialization.py @@ -5,7 +5,7 @@ import inspect from collections.abc import Callable, Iterable from dataclasses import dataclass -from typing import Any, TypeVar +from typing import TYPE_CHECKING, Any, TypeVar, get_args, get_origin from haystack.core.component.component import _hook_component_init from haystack.core.errors import DeserializationError, SerializationError @@ -16,7 +16,10 @@ from haystack.core.serialization_security import allow_deserialization_module as allow_deserialization_module from haystack.utils.auth import Secret from haystack.utils.device import ComponentDevice -from haystack.utils.type_serialization import _import_class_by_name +from haystack.utils.type_serialization import _import_class_by_name, _is_union_type + +if TYPE_CHECKING: + from haystack.core.pipeline.base import PipelineBase T = TypeVar("T") @@ -379,3 +382,101 @@ def import_class_by_name(fully_qualified_name: str) -> type[object]: :raises DeserializationError: If the module is not on the deserialization allowlist. """ return _import_class_by_name(fully_qualified_name) + + +def _resolve_from_dict_class(type_: Any) -> Any: + """ + Resolve the class whose `from_dict` method can deserialize values of `type_`. + + Handles plain classes, `list[T]`, and optionals/unions of those. In unions, the first arm resolving to a + `from_dict`-capable class wins. + + :param type_: The type to inspect. + :returns: The resolved class, or None if `type_` involves no `from_dict`-capable class. + """ + if _is_union_type(type_): + for arm in get_args(type_): + resolved = _resolve_from_dict_class(arm) + if resolved is not None: + return resolved + return None + if get_origin(type_) is list: + args = get_args(type_) + return _resolve_from_dict_class(args[0]) if args else None + return type_ if hasattr(type_, "from_dict") else None + + +def _deserialize_from_dict(value: Any, target_class: Any) -> Any: + """ + Deserialize `value` using `target_class.from_dict`. + + Dictionaries are deserialized directly, lists element-wise (leaving non-dictionary items untouched). Any other + value is returned unchanged. + + :param value: The value to deserialize. + :param target_class: The class providing the `from_dict` method. + :returns: The deserialized value. + """ + if isinstance(value, dict): + return target_class.from_dict(value) + if isinstance(value, list): + return [target_class.from_dict(item) if isinstance(item, dict) else item for item in value] + return value + + +def _coerce_input_value(value: Any, socket_types: list[Any]) -> Any: + for type_ in socket_types: + target_class = _resolve_from_dict_class(type_) + if target_class is not None: + return _deserialize_from_dict(value, target_class) + return value + + +def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> dict[str, Any]: + """ + Deserialize serialized Haystack objects in pipeline input data, based on the pipeline's input socket types. + + For every provided value whose input socket type involves a class exposing a `from_dict` method + (such as `ChatMessage` or `Document`), plain dictionaries are converted into instances of that class. + Socket types of the form `T`, `list[T]`, and optionals/unions of those are supported. Values that are + already deserialized, or whose socket types involve no `from_dict`-capable class, are returned unchanged. + + The dictionaries are expected to be in the component-native format produced by each object's `to_dict` + method. + + Like `Pipeline.run`, `data` accepts the nested format (`{"component_name": {"input_name": value}}`) and + the flat format (`{"input_name": value}`). The same format detection rules apply and the returned + dictionary keeps the format of `data`. + + Usage example: + ```python + serialized_inputs = {"agent": {"messages": [message.to_dict() for message in messages]}} + inputs = coerce_pipeline_inputs(pipeline, serialized_inputs) + result = pipeline.run(inputs) + ``` + + :param pipeline: The pipeline whose input socket types drive the coercion. + :param data: The pipeline input data, in nested or flat format. + :returns: A new dictionary with the same structure as `data` and deserialized values. + :raises Exception: Any error raised by a `from_dict` method if a dictionary does not match the expected format. + """ + # mirrors the format detection in Pipeline.run: nested if all values are dictionaries + if all(isinstance(value, dict) for value in data.values()): + available_inputs = pipeline.inputs(include_components_with_connected_inputs=True) + coerced: dict[str, Any] = {} + for component_name, component_inputs in data.items(): + sockets = available_inputs.get(component_name, {}) + coerced[component_name] = { + input_name: _coerce_input_value(value, [sockets[input_name]["type"]] if input_name in sockets else []) + for input_name, value in component_inputs.items() + } + return coerced + + # flat format: input names are matched across components like in Pipeline.run + available_inputs = pipeline.inputs() + return { + input_name: _coerce_input_value( + value, [sockets[input_name]["type"] for sockets in available_inputs.values() if input_name in sockets] + ) + for input_name, value in data.items() + } diff --git a/haystack/tools/component_tool.py b/haystack/tools/component_tool.py index b2a7a51c561..81876e7d5d0 100644 --- a/haystack/tools/component_tool.py +++ b/haystack/tools/component_tool.py @@ -11,6 +11,8 @@ from haystack.components.agents.state.state import State from haystack.core.component import Component from haystack.core.serialization import ( + _deserialize_from_dict, + _resolve_from_dict_class, component_from_dict, component_to_dict, generate_qualified_class_name, @@ -31,7 +33,6 @@ _serialize_outputs_to_state, _serialize_outputs_to_string, ) -from haystack.utils.deserialization import _deserialize_from_dict, _resolve_from_dict_class logger = logging.getLogger(__name__) diff --git a/haystack/utils/__init__.py b/haystack/utils/__init__.py index 96d61269671..1095e49acbd 100644 --- a/haystack/utils/__init__.py +++ b/haystack/utils/__init__.py @@ -13,7 +13,7 @@ "base_serialization": ["_deserialize_value_with_schema", "_serialize_value_with_schema"], "callable_serialization": ["deserialize_callable", "serialize_callable"], "device": ["ComponentDevice", "Device", "DeviceMap", "DeviceType"], - "deserialization": ["coerce_pipeline_inputs", "deserialize_chatgenerator_inplace", "deserialize_component_inplace"], + "deserialization": ["deserialize_chatgenerator_inplace", "deserialize_component_inplace"], "filters": ["document_matches_filter"], "jinja2_extensions": ["Jinja2TimeExtension"], "jupyter": ["is_in_jupyter"], @@ -30,7 +30,6 @@ from .base_serialization import _serialize_value_with_schema as _serialize_value_with_schema from .callable_serialization import deserialize_callable as deserialize_callable from .callable_serialization import serialize_callable as serialize_callable - from .deserialization import coerce_pipeline_inputs as coerce_pipeline_inputs from .deserialization import deserialize_chatgenerator_inplace as deserialize_chatgenerator_inplace from .deserialization import deserialize_component_inplace as deserialize_component_inplace from .device import ComponentDevice as ComponentDevice diff --git a/haystack/utils/deserialization.py b/haystack/utils/deserialization.py index f91631add55..90d2b9fceaa 100644 --- a/haystack/utils/deserialization.py +++ b/haystack/utils/deserialization.py @@ -2,14 +2,10 @@ # # SPDX-License-Identifier: Apache-2.0 -from typing import TYPE_CHECKING, Any, get_args, get_origin +from typing import Any from haystack.core.errors import DeserializationError from haystack.core.serialization import component_from_dict, import_class_by_name -from haystack.utils.type_serialization import _is_union_type - -if TYPE_CHECKING: - from haystack.core.pipeline.base import PipelineBase def deserialize_chatgenerator_inplace(data: dict[str, Any], key: str = "chat_generator") -> None: @@ -58,98 +54,3 @@ def deserialize_component_inplace(data: dict[str, Any], key: str = "chat_generat raise DeserializationError(f"Class '{serialized_component['type']}' not correctly imported") from e data[key] = component_from_dict(cls=component_class, data=serialized_component, name=key) - - -def _resolve_from_dict_class(type_: Any) -> Any: - """ - Resolve the class whose `from_dict` method can deserialize values of `type_`. - - Handles plain classes, `list[T]`, and optionals/unions of those. In unions, the first arm resolving to a - `from_dict`-capable class wins. - - :param type_: The type to inspect. - :returns: The resolved class, or None if `type_` involves no `from_dict`-capable class. - """ - if _is_union_type(type_): - for arm in get_args(type_): - resolved = _resolve_from_dict_class(arm) - if resolved is not None: - return resolved - return None - if get_origin(type_) is list: - args = get_args(type_) - return _resolve_from_dict_class(args[0]) if args else None - return type_ if hasattr(type_, "from_dict") else None - - -def _deserialize_from_dict(value: Any, target_class: Any) -> Any: - """ - Deserialize `value` using `target_class.from_dict`. - - Dictionaries are deserialized directly, lists element-wise (leaving non-dictionary items untouched). Any other - value is returned unchanged. - - :param value: The value to deserialize. - :param target_class: The class providing the `from_dict` method. - :returns: The deserialized value. - """ - if isinstance(value, dict): - return target_class.from_dict(value) - if isinstance(value, list): - return [target_class.from_dict(item) if isinstance(item, dict) else item for item in value] - return value - - -def _coerce_input_value(value: Any, socket_types: list[Any]) -> Any: - for type_ in socket_types: - target_class = _resolve_from_dict_class(type_) - if target_class is not None: - return _deserialize_from_dict(value, target_class) - return value - - -def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> dict[str, Any]: - """ - Deserialize serialized Haystack objects in pipeline input data, based on the pipeline's input socket types. - - For every provided value whose input socket type involves a class exposing a `from_dict` method - (such as `ChatMessage` or `Document`), plain dictionaries are converted into instances of that class. - Socket types of the form `T`, `list[T]`, and optionals/unions of those are supported. Values that are - already deserialized, or whose socket types involve no `from_dict`-capable class, are returned unchanged. - - Like `Pipeline.run`, `data` accepts the nested format (`{"component_name": {"input_name": value}}`) and - the flat format (`{"input_name": value}`). The same format detection rules apply and the returned - dictionary keeps the format of `data`. - - Usage example: - ```python - serialized_inputs = {"prompt_builder": {"template": [message.to_dict() for message in messages]}} - inputs = coerce_pipeline_inputs(pipeline, serialized_inputs) - result = pipeline.run(inputs) - ``` - - :param pipeline: The pipeline whose input socket types drive the coercion. - :param data: The pipeline input data, in nested or flat format. - :returns: A new dictionary with the same structure as `data` and deserialized values. - :raises Exception: Any error raised by a `from_dict` method if a dictionary does not match the expected format. - """ - # mirrors the format detection in Pipeline.run: nested if all values are dictionaries - if all(isinstance(value, dict) for value in data.values()): - available_inputs = pipeline.inputs(include_components_with_connected_inputs=True) - coerced: dict[str, Any] = {} - for component_name, component_inputs in data.items(): - sockets = available_inputs.get(component_name, {}) - coerced[component_name] = { - input_name: _coerce_input_value(value, [sockets[input_name]["type"]] if input_name in sockets else []) - for input_name, value in component_inputs.items() - } - return coerced - - # flat format: input names are matched across components like in Pipeline.run - available_inputs = pipeline.inputs() - return { - input_name: _coerce_input_value( - value, [sockets[input_name]["type"] for sockets in available_inputs.values() if input_name in sockets] - ) - for input_name, value in data.items() - } diff --git a/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml b/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml index fe451ff1ac7..ec2de08f211 100644 --- a/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml +++ b/releasenotes/notes/coerce-pipeline-inputs-utility-d38e32c0ca3b76cd.yaml @@ -1,7 +1,7 @@ --- features: - | - Added the ``coerce_pipeline_inputs`` utility function (importable from ``haystack.utils``). + Added the ``coerce_pipeline_inputs`` utility function (importable from ``haystack``). It deserializes serialized Haystack objects in pipeline input data based on the pipeline's input socket types: for every provided value whose socket type involves a class exposing ``from_dict`` (such as ``ChatMessage`` or ``Document``), plain dictionaries are converted into instances of that @@ -13,7 +13,7 @@ features: .. code-block:: python - from haystack.utils import coerce_pipeline_inputs + from haystack import coerce_pipeline_inputs inputs = coerce_pipeline_inputs(pipeline, serialized_inputs) result = pipeline.run(inputs) diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py index 8422ab23394..cfc8864810a 100644 --- a/test/core/test_serialization.py +++ b/test/core/test_serialization.py @@ -8,11 +8,15 @@ from unittest.mock import Mock import pytest +from pydantic import BaseModel +from haystack import Document +from haystack.components.joiners import BranchJoiner from haystack.core.component import component from haystack.core.errors import DeserializationError, SerializationError from haystack.core.pipeline import Pipeline from haystack.core.serialization import ( + coerce_pipeline_inputs, component_from_dict, component_to_dict, default_from_dict, @@ -20,6 +24,19 @@ generate_qualified_class_name, import_class_by_name, ) +from haystack.dataclasses import ( + ByteStream, + ChatMessage, + ComponentInfo, + GeneratedAnswer, + ImageContent, + ReasoningContent, + SparseEmbedding, + StreamingChunk, + TextContent, + ToolCall, + ToolCallResult, +) from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.testing import factory from haystack.utils import ComponentDevice, Secret @@ -644,3 +661,167 @@ def test_default_from_dict_rejects_unknown_nested_parameter(): # The message lists the accepted parameters (sorted) and tells the user how to fix it. assert "Valid parameters are: 'document_store', 'name'." in message assert "Correct the parameter name or remove it from the serialized data." in message + + +@component +class _TypedEcho: + """Echoes back a single input whose socket type is set at construction time.""" + + def __init__(self, type_: Any) -> None: + component.set_input_types(self, value=type_) + component.set_output_types(self, value=Any) + + def run(self, **kwargs: Any) -> dict[str, Any]: + return {"value": kwargs.get("value")} + + +@component +class _MultiInputEcho: + """A component with one coercible socket alongside plain, non-coercible sockets.""" + + @component.output_types(messages=list[ChatMessage]) + def run(self, messages: list[ChatMessage], top_k: int = 3, config: dict[str, Any] | None = None) -> dict[str, Any]: + return {"messages": messages} + + +# One instance per Haystack dataclass exposing to_dict/from_dict, used to check that coerce_pipeline_inputs +# round-trips each of them for the `T`, `list[T]`, and optional socket shapes. +DATACLASS_INSTANCES = [ + pytest.param(Document(content="Hello", meta={"k": "v"}), id="Document"), + pytest.param(ChatMessage.from_user("Hi"), id="ChatMessage"), + pytest.param( + ChatMessage.from_assistant("x", tool_calls=[ToolCall(tool_name="t", arguments={"a": 1}, id="1")]), + id="ChatMessage-with-tool-call", + ), + pytest.param(ByteStream(data=b"hello", mime_type="text/plain", meta={"k": "v"}), id="ByteStream"), + pytest.param(SparseEmbedding(indices=[0, 2], values=[0.1, 0.2]), id="SparseEmbedding"), + pytest.param(GeneratedAnswer(data="answer", query="q", documents=[Document(content="d")]), id="GeneratedAnswer"), + pytest.param(ToolCall(tool_name="t", arguments={"a": 1}, id="1"), id="ToolCall"), + pytest.param( + ToolCallResult(result="r", origin=ToolCall(tool_name="t", arguments={"a": 1}, id="1"), error=False), + id="ToolCallResult", + ), + pytest.param(TextContent(text="hi"), id="TextContent"), + pytest.param(ReasoningContent(reasoning_text="because"), id="ReasoningContent"), + pytest.param(ImageContent(base64_image="aGVsbG8=", mime_type="image/png"), id="ImageContent"), + pytest.param(ComponentInfo(type="some.Type", name="comp"), id="ComponentInfo"), + pytest.param(StreamingChunk(content="hi"), id="StreamingChunk"), +] + + +class TestCoercePipelineInputsRoundTrip: + """`to_dict()` output is coerced back to an equal instance for every from_dict-capable dataclass.""" + + @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + def test_single_socket(self, instance): + pipe = Pipeline() + pipe.add_component("echo", _TypedEcho(type(instance))) + + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": instance.to_dict()}}) + + assert coerced["echo"]["value"] == instance + + @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + def test_list_socket(self, instance): + pipe = Pipeline() + pipe.add_component("echo", _TypedEcho(list[type(instance)])) + + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [instance.to_dict()]}}) + + assert coerced["echo"]["value"] == [instance] + + @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + def test_optional_socket(self, instance): + pipe = Pipeline() + pipe.add_component("echo", _TypedEcho(type(instance) | None)) + + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": instance.to_dict()}}) + + assert coerced["echo"]["value"] == instance + + +class TestCoercePipelineInputs: + @pytest.fixture + def pipeline(self): + pipe = Pipeline() + pipe.add_component("echo", _MultiInputEcho()) + return pipe + + def test_nested_format(self, pipeline): + messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] + data = {"echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"echo": {"messages": messages, "top_k": 5}} + # the input data is not mutated + assert isinstance(data["echo"]["messages"][0], dict) + + def test_flat_format(self, pipeline): + messages = [ChatMessage.from_user("Hi")] + data = {"messages": [message.to_dict() for message in messages], "top_k": 5} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"messages": messages, "top_k": 5} + + def test_instances_pass_through(self, pipeline): + messages = [ChatMessage.from_user("Hi")] + data = {"echo": {"messages": messages}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced["echo"]["messages"][0] is messages[0] + + def test_mixed_list(self, pipeline): + message = ChatMessage.from_user("Hi") + data = {"echo": {"messages": [message, ChatMessage.from_assistant("Hello").to_dict()]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced["echo"]["messages"] == [message, ChatMessage.from_assistant("Hello")] + + def test_non_coercible_values_untouched(self, pipeline): + data = {"echo": {"messages": [], "top_k": 5, "config": {"a": 1}}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == data + + def test_unknown_inputs_untouched(self, pipeline): + message_dict = ChatMessage.from_user("Hi").to_dict() + data = {"unknown_input": [message_dict], "top_k": 5} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"unknown_input": [message_dict], "top_k": 5} + + def test_unknown_component_untouched(self, pipeline): + message_dict = ChatMessage.from_user("Hi").to_dict() + data = {"unknown_component": {"messages": [message_dict]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"unknown_component": {"messages": [message_dict]}} + + def test_variadic_socket(self): + pipe = Pipeline() + pipe.add_component("joiner", BranchJoiner(list[ChatMessage])) + messages = [ChatMessage.from_user("Hi")] + data = {"value": [message.to_dict() for message in messages], "unrelated": 1} + + coerced = coerce_pipeline_inputs(pipe, data) + + assert coerced["value"] == messages + + def test_pydantic_model_dump_payload(self, pipeline): + class Response(BaseModel): + messages: list[ChatMessage] + + messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] + dumped = Response(messages=messages).model_dump(mode="json") + data = {"echo": {"messages": dumped["messages"]}} + + coerced = coerce_pipeline_inputs(pipeline, data) + + assert coerced == {"echo": {"messages": messages}} diff --git a/test/utils/test_deserialization.py b/test/utils/test_deserialization.py index 01738af84e5..320f313075d 100644 --- a/test/utils/test_deserialization.py +++ b/test/utils/test_deserialization.py @@ -3,17 +3,11 @@ # SPDX-License-Identifier: Apache-2.0 -from typing import Any - import pytest -from pydantic import BaseModel -from haystack import Document, Pipeline, component from haystack.components.generators.chat.openai import OpenAIChatGenerator -from haystack.components.joiners import BranchJoiner from haystack.core.errors import DeserializationError -from haystack.dataclasses import ChatMessage -from haystack.utils.deserialization import coerce_pipeline_inputs, deserialize_component_inplace +from haystack.utils.deserialization import deserialize_component_inplace class ChatGeneratorWithoutFromDict: @@ -21,27 +15,6 @@ def to_dict(self): return {"type": "test_deserialization.ChatGeneratorWithoutFromDict"} -@component -class MessageListEcho: - @component.output_types(messages=list[ChatMessage]) - def run(self, messages: list[ChatMessage], top_k: int = 3, config: dict[str, Any] | None = None) -> dict[str, Any]: - return {"messages": messages} - - -@component -class SingleMessageEcho: - @component.output_types(message=ChatMessage) - def run(self, message: ChatMessage, query: str = "") -> dict[str, Any]: - return {"message": message} - - -@component -class DocumentListEcho: - @component.output_types(documents=list[Document]) - def run(self, documents: list[Document] | None = None) -> dict[str, Any]: - return {"documents": documents or []} - - class TestDeserializeComponentInplace: def test_deserialize_component_inplace(self, monkeypatch): monkeypatch.setenv("OPENAI_API_KEY", "test-api-key") @@ -76,107 +49,3 @@ def test_component_no_from_dict_method(self): data = {"chat_generator": chat_generator.to_dict()} deserialize_component_inplace(data) assert isinstance(data["chat_generator"], ChatGeneratorWithoutFromDict) - - -class TestCoercePipelineInputs: - @pytest.fixture - def pipeline(self): - pipe = Pipeline() - pipe.add_component("messages_echo", MessageListEcho()) - pipe.add_component("single_echo", SingleMessageEcho()) - pipe.add_component("documents_echo", DocumentListEcho()) - return pipe - - def test_nested_format(self, pipeline): - messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] - data: dict[str, Any] = {"messages_echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"messages_echo": {"messages": messages, "top_k": 5}} - assert isinstance(data["messages_echo"]["messages"][0], dict) - - def test_flat_format(self, pipeline): - messages = [ChatMessage.from_user("Hi")] - data = {"messages": [message.to_dict() for message in messages], "top_k": 5} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"messages": messages, "top_k": 5} - - def test_single_message_socket(self, pipeline): - message = ChatMessage.from_user("Hi") - data = {"single_echo": {"message": message.to_dict(), "query": "some query"}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"single_echo": {"message": message, "query": "some query"}} - - def test_optional_list_socket(self, pipeline): - documents = [Document(content="Hello"), Document(content="World")] - data = {"documents_echo": {"documents": [document.to_dict() for document in documents]}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"documents_echo": {"documents": documents}} - - def test_instances_pass_through(self, pipeline): - messages = [ChatMessage.from_user("Hi")] - data = {"messages_echo": {"messages": messages}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced["messages_echo"]["messages"][0] is messages[0] - - def test_mixed_list(self, pipeline): - message = ChatMessage.from_user("Hi") - data = {"messages_echo": {"messages": [message, ChatMessage.from_assistant("Hello").to_dict()]}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced["messages_echo"]["messages"] == [message, ChatMessage.from_assistant("Hello")] - - def test_non_coercible_values_untouched(self, pipeline): - data = {"messages_echo": {"messages": [], "top_k": 5, "config": {"a": 1}}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == data - - def test_unknown_inputs_untouched(self, pipeline): - message_dict = ChatMessage.from_user("Hi").to_dict() - data = {"unknown_input": [message_dict], "top_k": 5} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"unknown_input": [message_dict], "top_k": 5} - - def test_unknown_component_untouched(self, pipeline): - message_dict = ChatMessage.from_user("Hi").to_dict() - data = {"unknown_component": {"messages": [message_dict]}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"unknown_component": {"messages": [message_dict]}} - - def test_variadic_socket(self): - pipe = Pipeline() - pipe.add_component("joiner", BranchJoiner(list[ChatMessage])) - messages = [ChatMessage.from_user("Hi")] - data = {"value": [message.to_dict() for message in messages], "unrelated": 1} - - coerced = coerce_pipeline_inputs(pipe, data) - - assert coerced["value"] == messages - - def test_pydantic_model_dump_payload(self, pipeline): - class Response(BaseModel): - messages: list[ChatMessage] - - messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] - dumped = Response(messages=messages).model_dump(mode="json") - data = {"messages_echo": {"messages": dumped["messages"]}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"messages_echo": {"messages": messages}} From 453e7eedab6eb83dc4f0ee60d2a7c1372d36c6c0 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Wed, 22 Jul 2026 10:07:43 +0200 Subject: [PATCH 3/6] Add support for pydantic models and dataclasses --- haystack/core/serialization.py | 72 +++++++++++++++++++-------- haystack/tools/component_tool.py | 8 +-- test/core/test_serialization.py | 83 +++++++++++++++++++++++++------- 3 files changed, 121 insertions(+), 42 deletions(-) diff --git a/haystack/core/serialization.py b/haystack/core/serialization.py index 3d2a622748a..a0c945242ae 100644 --- a/haystack/core/serialization.py +++ b/haystack/core/serialization.py @@ -4,9 +4,11 @@ import inspect from collections.abc import Callable, Iterable -from dataclasses import dataclass +from dataclasses import dataclass, is_dataclass from typing import TYPE_CHECKING, Any, TypeVar, get_args, get_origin +from pydantic import BaseModel, TypeAdapter + from haystack.core.component.component import _hook_component_init from haystack.core.errors import DeserializationError, SerializationError @@ -384,49 +386,72 @@ def import_class_by_name(fully_qualified_name: str) -> type[object]: return _import_class_by_name(fully_qualified_name) -def _resolve_from_dict_class(type_: Any) -> Any: +def _resolve_coercible_class(type_: Any) -> Any: """ - Resolve the class whose `from_dict` method can deserialize values of `type_`. + Resolve the class that can deserialize values of `type_` from a plain dictionary. - Handles plain classes, `list[T]`, and optionals/unions of those. In unions, the first arm resolving to a - `from_dict`-capable class wins. + A class is coercible if it exposes a `from_dict` method (Haystack dataclasses and Components), is a Pydantic + model, or is a standard-library dataclass. Handles plain classes, `list[T]`, and optionals/unions of those. + In unions, the first arm resolving to a coercible class wins. :param type_: The type to inspect. - :returns: The resolved class, or None if `type_` involves no `from_dict`-capable class. + :returns: The resolved class, or None if `type_` involves no coercible class. """ if _is_union_type(type_): for arm in get_args(type_): - resolved = _resolve_from_dict_class(arm) + resolved = _resolve_coercible_class(arm) if resolved is not None: return resolved return None if get_origin(type_) is list: args = get_args(type_) - return _resolve_from_dict_class(args[0]) if args else None - return type_ if hasattr(type_, "from_dict") else None + return _resolve_coercible_class(args[0]) if args else None + if not isinstance(type_, type): + return None + if hasattr(type_, "from_dict") or issubclass(type_, BaseModel) or is_dataclass(type_): + return type_ + return None + + +def _deserialize_one(value: dict[str, Any], target_class: Any) -> Any: + """ + Deserialize a single dictionary into an instance of `target_class`. + + `from_dict`-capable classes take priority so Haystack objects keep their native deserialization. Pydantic + models and standard-library dataclasses are deserialized with Pydantic. + + :param value: The dictionary to deserialize. + :param target_class: The coercible class resolved for the value's socket type. + :returns: The deserialized instance. + """ + if hasattr(target_class, "from_dict"): + return target_class.from_dict(value) + if issubclass(target_class, BaseModel): + return target_class.model_validate(value) + return TypeAdapter(target_class).validate_python(value) def _deserialize_from_dict(value: Any, target_class: Any) -> Any: """ - Deserialize `value` using `target_class.from_dict`. + Deserialize `value` into instances of `target_class`. Dictionaries are deserialized directly, lists element-wise (leaving non-dictionary items untouched). Any other value is returned unchanged. :param value: The value to deserialize. - :param target_class: The class providing the `from_dict` method. + :param target_class: The coercible class resolved for the value's socket type. :returns: The deserialized value. """ if isinstance(value, dict): - return target_class.from_dict(value) + return _deserialize_one(value, target_class) if isinstance(value, list): - return [target_class.from_dict(item) if isinstance(item, dict) else item for item in value] + return [_deserialize_one(item, target_class) if isinstance(item, dict) else item for item in value] return value def _coerce_input_value(value: Any, socket_types: list[Any]) -> Any: for type_ in socket_types: - target_class = _resolve_from_dict_class(type_) + target_class = _resolve_coercible_class(type_) if target_class is not None: return _deserialize_from_dict(value, target_class) return value @@ -436,13 +461,17 @@ def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> di """ Deserialize serialized Haystack objects in pipeline input data, based on the pipeline's input socket types. - For every provided value whose input socket type involves a class exposing a `from_dict` method - (such as `ChatMessage` or `Document`), plain dictionaries are converted into instances of that class. - Socket types of the form `T`, `list[T]`, and optionals/unions of those are supported. Values that are - already deserialized, or whose socket types involve no `from_dict`-capable class, are returned unchanged. + For every provided value whose input socket type involves a coercible class, plain dictionaries are + converted into instances of that class. A class is coercible if it exposes a `from_dict` method (such as + `ChatMessage` or `Document`), is a Pydantic model, or is a standard-library dataclass. Socket types of the + form `T`, `list[T]`, and optionals/unions of those are supported. Values that are already deserialized, or + whose socket types involve no coercible class, are returned unchanged. - The dictionaries are expected to be in the component-native format produced by each object's `to_dict` - method. + The dictionaries are expected to be in the format produced by the class's serialization: `to_dict` for + `from_dict`-capable classes, `model_dump` for Pydantic models, and `dataclasses.asdict` for standard-library + dataclasses. A Pydantic model or dataclass may nest `from_dict`-capable Haystack objects: Pydantic + (de)serializes them via their fields, so a `model_dump` payload round-trips through `model_validate` without + going through `to_dict`/`from_dict`. Like `Pipeline.run`, `data` accepts the nested format (`{"component_name": {"input_name": value}}`) and the flat format (`{"input_name": value}`). The same format detection rules apply and the returned @@ -458,7 +487,8 @@ def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> di :param pipeline: The pipeline whose input socket types drive the coercion. :param data: The pipeline input data, in nested or flat format. :returns: A new dictionary with the same structure as `data` and deserialized values. - :raises Exception: Any error raised by a `from_dict` method if a dictionary does not match the expected format. + :raises Exception: Any error raised while deserializing a dictionary that does not match the expected format, + such as a `from_dict` error or a Pydantic `ValidationError`. """ # mirrors the format detection in Pipeline.run: nested if all values are dictionaries if all(isinstance(value, dict) for value in data.values()): diff --git a/haystack/tools/component_tool.py b/haystack/tools/component_tool.py index 81876e7d5d0..835ba9c7616 100644 --- a/haystack/tools/component_tool.py +++ b/haystack/tools/component_tool.py @@ -12,7 +12,7 @@ from haystack.core.component import Component from haystack.core.serialization import ( _deserialize_from_dict, - _resolve_from_dict_class, + _resolve_coercible_class, component_from_dict, component_to_dict, generate_qualified_class_name, @@ -392,9 +392,9 @@ def _convert_param(self, param_value: Any, param_type: type) -> Any: :returns: The converted parameter value. """ - # Types involving a from_dict-capable class (e.g. list[ChatMessage] | None) are deserialized via - # from_dict; everything else is validated with Pydantic. - target_class = _resolve_from_dict_class(param_type) + # Types involving a coercible class (e.g. list[ChatMessage] | None) are deserialized element-wise via + # that class; everything else is validated with Pydantic. + target_class = _resolve_coercible_class(param_type) if target_class is not None: return _deserialize_from_dict(param_value, target_class) diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py index cfc8864810a..cb35f1b6a69 100644 --- a/test/core/test_serialization.py +++ b/test/core/test_serialization.py @@ -4,6 +4,7 @@ import sys from copy import deepcopy +from dataclasses import asdict, dataclass from typing import Any from unittest.mock import Mock @@ -664,7 +665,7 @@ def test_default_from_dict_rejects_unknown_nested_parameter(): @component -class _TypedEcho: +class TypedEcho: """Echoes back a single input whose socket type is set at construction time.""" def __init__(self, type_: Any) -> None: @@ -676,7 +677,7 @@ def run(self, **kwargs: Any) -> dict[str, Any]: @component -class _MultiInputEcho: +class MultiInputEcho: """A component with one coercible socket alongside plain, non-coercible sockets.""" @component.output_types(messages=list[ChatMessage]) @@ -684,9 +685,35 @@ def run(self, messages: list[ChatMessage], top_k: int = 3, config: dict[str, Any return {"messages": messages} -# One instance per Haystack dataclass exposing to_dict/from_dict, used to check that coerce_pipeline_inputs -# round-trips each of them for the `T`, `list[T]`, and optional socket shapes. -DATACLASS_INSTANCES = [ +@dataclass +class PlainDataclass: + name: str + count: int + + +class PydanticModel(BaseModel): + a: int + b: str + + +class Conversation(BaseModel): + messages: list[ChatMessage] + documents: list[Document] + + +def serialize_for_client(instance: Any) -> dict[str, Any]: + """Serialize an instance the way a JSON client would, per its serialization scheme.""" + if isinstance(instance, BaseModel): + return instance.model_dump(mode="json") + if hasattr(instance, "to_dict"): + return instance.to_dict() + return asdict(instance) + + +# One instance per coercible class, used to check that coerce_pipeline_inputs round-trips each of them for the +# `T`, `list[T]`, and optional socket shapes. Covers Haystack dataclasses (from_dict), a Pydantic model, and a +# standard-library dataclass. +COERCIBLE_INSTANCES = [ pytest.param(Document(content="Hello", meta={"k": "v"}), id="Document"), pytest.param(ChatMessage.from_user("Hi"), id="ChatMessage"), pytest.param( @@ -706,36 +733,42 @@ def run(self, messages: list[ChatMessage], top_k: int = 3, config: dict[str, Any pytest.param(ImageContent(base64_image="aGVsbG8=", mime_type="image/png"), id="ImageContent"), pytest.param(ComponentInfo(type="some.Type", name="comp"), id="ComponentInfo"), pytest.param(StreamingChunk(content="hi"), id="StreamingChunk"), + pytest.param(PydanticModel(a=1, b="x"), id="pydantic-model"), + pytest.param(PlainDataclass(name="x", count=2), id="stdlib-dataclass"), + pytest.param( + Conversation(messages=[ChatMessage.from_user("Hi")], documents=[Document(content="D")]), + id="pydantic-model-nesting-haystack", + ), ] class TestCoercePipelineInputsRoundTrip: - """`to_dict()` output is coerced back to an equal instance for every from_dict-capable dataclass.""" + """Serialized output is coerced back to an equal instance for every coercible class.""" - @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) def test_single_socket(self, instance): pipe = Pipeline() - pipe.add_component("echo", _TypedEcho(type(instance))) + pipe.add_component("echo", TypedEcho(type(instance))) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": instance.to_dict()}}) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) assert coerced["echo"]["value"] == instance - @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) def test_list_socket(self, instance): pipe = Pipeline() - pipe.add_component("echo", _TypedEcho(list[type(instance)])) + pipe.add_component("echo", TypedEcho(list[type(instance)])) # type: ignore[misc] - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [instance.to_dict()]}}) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [serialize_for_client(instance)]}}) assert coerced["echo"]["value"] == [instance] - @pytest.mark.parametrize("instance", DATACLASS_INSTANCES) + @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) def test_optional_socket(self, instance): pipe = Pipeline() - pipe.add_component("echo", _TypedEcho(type(instance) | None)) + pipe.add_component("echo", TypedEcho(type(instance) | None)) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": instance.to_dict()}}) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) assert coerced["echo"]["value"] == instance @@ -744,12 +777,12 @@ class TestCoercePipelineInputs: @pytest.fixture def pipeline(self): pipe = Pipeline() - pipe.add_component("echo", _MultiInputEcho()) + pipe.add_component("echo", MultiInputEcho()) return pipe def test_nested_format(self, pipeline): messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] - data = {"echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} + data: dict[str, Any] = {"echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} coerced = coerce_pipeline_inputs(pipeline, data) @@ -825,3 +858,19 @@ class Response(BaseModel): coerced = coerce_pipeline_inputs(pipeline, data) assert coerced == {"echo": {"messages": messages}} + + def test_dict_any_field_from_pydantic_model_is_coercible(self, pipeline): + # An API-layer Pydantic model types the pipeline inputs as dict[str, Any], so Pydantic leaves the + # nested serialized objects as plain dicts. Passing that field to coerce_pipeline_inputs recovers the + # objects, because coercion is driven by the pipeline's socket types rather than the field's type. + class Request(BaseModel): + pipeline_inputs: dict[str, Any] + + messages = [ChatMessage.from_user("Hi")] + request = Request(pipeline_inputs={"echo": {"messages": [message.to_dict() for message in messages]}}) + dumped = request.model_dump(mode="json") + assert isinstance(dumped["pipeline_inputs"]["echo"]["messages"][0], dict) + + coerced = coerce_pipeline_inputs(pipeline, dumped["pipeline_inputs"]) + + assert coerced == {"echo": {"messages": messages}} From 799364bc15c91d2eed2a1942bcc95d34ae4e587d Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Wed, 22 Jul 2026 10:29:42 +0200 Subject: [PATCH 4/6] improvements --- test/core/test_serialization.py | 126 +++++++++++++------------------- 1 file changed, 49 insertions(+), 77 deletions(-) diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py index cb35f1b6a69..f9eb425f0e0 100644 --- a/test/core/test_serialization.py +++ b/test/core/test_serialization.py @@ -4,7 +4,7 @@ import sys from copy import deepcopy -from dataclasses import asdict, dataclass +from dataclasses import dataclass from typing import Any from unittest.mock import Mock @@ -28,12 +28,12 @@ from haystack.dataclasses import ( ByteStream, ChatMessage, - ComponentInfo, + ChatRole, + FileContent, GeneratedAnswer, ImageContent, ReasoningContent, SparseEmbedding, - StreamingChunk, TextContent, ToolCall, ToolCallResult, @@ -701,38 +701,60 @@ class Conversation(BaseModel): documents: list[Document] -def serialize_for_client(instance: Any) -> dict[str, Any]: - """Serialize an instance the way a JSON client would, per its serialization scheme.""" - if isinstance(instance, BaseModel): - return instance.model_dump(mode="json") - if hasattr(instance, "to_dict"): - return instance.to_dict() - return asdict(instance) +class SerializationEnvelope(BaseModel): + value: Any -# One instance per coercible class, used to check that coerce_pipeline_inputs round-trips each of them for the -# `T`, `list[T]`, and optional socket shapes. Covers Haystack dataclasses (from_dict), a Pydantic model, and a -# standard-library dataclass. +def serialize_for_client(instance: Any) -> Any: + """ + Serialize an instance the way a FastAPI app does: instances arrive inside a loosely-typed Pydantic field, + so Pydantic (not to_dict) produces the JSON payload. + """ + return SerializationEnvelope(value=instance).model_dump(mode="json")["value"] + + +TOOL_CALL = ToolCall(tool_name="t", arguments={"a": 1}, id="1") + COERCIBLE_INSTANCES = [ - pytest.param(Document(content="Hello", meta={"k": "v"}), id="Document"), - pytest.param(ChatMessage.from_user("Hi"), id="ChatMessage"), + # We create a fully loaded Document to test all fields serializations pytest.param( - ChatMessage.from_assistant("x", tool_calls=[ToolCall(tool_name="t", arguments={"a": 1}, id="1")]), - id="ChatMessage-with-tool-call", + Document( + content="Hello", + meta={"k": "v"}, + score=0.5, + embedding=[0.1, 0.2], + sparse_embedding=SparseEmbedding(indices=[0, 2], values=[0.1, 0.2]), + ), + id="Document", + ), + # We create a fully loaded ChatMessage to test all possible content type serializations + pytest.param( + ChatMessage( + _role=ChatRole.ASSISTANT, + _content=[ + TextContent(text="hi"), + TOOL_CALL, + ToolCallResult(result="r", origin=TOOL_CALL, error=False), + ImageContent(base64_image="aGVsbG8=", mime_type="image/png"), + ReasoningContent(reasoning_text="because"), + FileContent(base64_data="aGVsbG8=", mime_type="application/pdf"), + ], + _name="n", + _meta={"k": "v"}, + ), + id="ChatMessage", ), - pytest.param(ByteStream(data=b"hello", mime_type="text/plain", meta={"k": "v"}), id="ByteStream"), - pytest.param(SparseEmbedding(indices=[0, 2], values=[0.1, 0.2]), id="SparseEmbedding"), - pytest.param(GeneratedAnswer(data="answer", query="q", documents=[Document(content="d")]), id="GeneratedAnswer"), - pytest.param(ToolCall(tool_name="t", arguments={"a": 1}, id="1"), id="ToolCall"), pytest.param( - ToolCallResult(result="r", origin=ToolCall(tool_name="t", arguments={"a": 1}, id="1"), error=False), - id="ToolCallResult", + ByteStream(data=b"hello", mime_type="text/plain", meta={"k": "v"}), + id="ByteStream", + marks=pytest.mark.xfail( + reason="Pydantic serializes the bytes field as a JSON string that ByteStream.from_dict cannot rebuild", + raises=TypeError, + strict=True, + ), ), - pytest.param(TextContent(text="hi"), id="TextContent"), - pytest.param(ReasoningContent(reasoning_text="because"), id="ReasoningContent"), + pytest.param(GeneratedAnswer(data="answer", query="q", documents=[Document(content="d")]), id="GeneratedAnswer"), pytest.param(ImageContent(base64_image="aGVsbG8=", mime_type="image/png"), id="ImageContent"), - pytest.param(ComponentInfo(type="some.Type", name="comp"), id="ComponentInfo"), - pytest.param(StreamingChunk(content="hi"), id="StreamingChunk"), pytest.param(PydanticModel(a=1, b="x"), id="pydantic-model"), pytest.param(PlainDataclass(name="x", count=2), id="stdlib-dataclass"), pytest.param( @@ -749,27 +771,21 @@ class TestCoercePipelineInputsRoundTrip: def test_single_socket(self, instance): pipe = Pipeline() pipe.add_component("echo", TypedEcho(type(instance))) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) - assert coerced["echo"]["value"] == instance @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) def test_list_socket(self, instance): pipe = Pipeline() pipe.add_component("echo", TypedEcho(list[type(instance)])) # type: ignore[misc] - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [serialize_for_client(instance)]}}) - assert coerced["echo"]["value"] == [instance] @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) def test_optional_socket(self, instance): pipe = Pipeline() pipe.add_component("echo", TypedEcho(type(instance) | None)) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) - assert coerced["echo"]["value"] == instance @@ -783,9 +799,7 @@ def pipeline(self): def test_nested_format(self, pipeline): messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] data: dict[str, Any] = {"echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"echo": {"messages": messages, "top_k": 5}} # the input data is not mutated assert isinstance(data["echo"]["messages"][0], dict) @@ -793,48 +807,36 @@ def test_nested_format(self, pipeline): def test_flat_format(self, pipeline): messages = [ChatMessage.from_user("Hi")] data = {"messages": [message.to_dict() for message in messages], "top_k": 5} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"messages": messages, "top_k": 5} def test_instances_pass_through(self, pipeline): messages = [ChatMessage.from_user("Hi")] data = {"echo": {"messages": messages}} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced["echo"]["messages"][0] is messages[0] def test_mixed_list(self, pipeline): message = ChatMessage.from_user("Hi") data = {"echo": {"messages": [message, ChatMessage.from_assistant("Hello").to_dict()]}} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced["echo"]["messages"] == [message, ChatMessage.from_assistant("Hello")] def test_non_coercible_values_untouched(self, pipeline): data = {"echo": {"messages": [], "top_k": 5, "config": {"a": 1}}} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == data def test_unknown_inputs_untouched(self, pipeline): message_dict = ChatMessage.from_user("Hi").to_dict() data = {"unknown_input": [message_dict], "top_k": 5} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"unknown_input": [message_dict], "top_k": 5} def test_unknown_component_untouched(self, pipeline): message_dict = ChatMessage.from_user("Hi").to_dict() data = {"unknown_component": {"messages": [message_dict]}} - coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"unknown_component": {"messages": [message_dict]}} def test_variadic_socket(self): @@ -842,35 +844,5 @@ def test_variadic_socket(self): pipe.add_component("joiner", BranchJoiner(list[ChatMessage])) messages = [ChatMessage.from_user("Hi")] data = {"value": [message.to_dict() for message in messages], "unrelated": 1} - coerced = coerce_pipeline_inputs(pipe, data) - assert coerced["value"] == messages - - def test_pydantic_model_dump_payload(self, pipeline): - class Response(BaseModel): - messages: list[ChatMessage] - - messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] - dumped = Response(messages=messages).model_dump(mode="json") - data = {"echo": {"messages": dumped["messages"]}} - - coerced = coerce_pipeline_inputs(pipeline, data) - - assert coerced == {"echo": {"messages": messages}} - - def test_dict_any_field_from_pydantic_model_is_coercible(self, pipeline): - # An API-layer Pydantic model types the pipeline inputs as dict[str, Any], so Pydantic leaves the - # nested serialized objects as plain dicts. Passing that field to coerce_pipeline_inputs recovers the - # objects, because coercion is driven by the pipeline's socket types rather than the field's type. - class Request(BaseModel): - pipeline_inputs: dict[str, Any] - - messages = [ChatMessage.from_user("Hi")] - request = Request(pipeline_inputs={"echo": {"messages": [message.to_dict() for message in messages]}}) - dumped = request.model_dump(mode="json") - assert isinstance(dumped["pipeline_inputs"]["echo"]["messages"][0], dict) - - coerced = coerce_pipeline_inputs(pipeline, dumped["pipeline_inputs"]) - - assert coerced == {"echo": {"messages": messages}} From 5fd5248ceb710785b5eb7bd284ba1f63ba4c818b Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Thu, 23 Jul 2026 11:04:43 +0200 Subject: [PATCH 5/6] PR comments --- haystack/core/serialization.py | 52 ++++++++++++++++++++------------ haystack/tools/component_tool.py | 12 ++++---- test/core/test_serialization.py | 48 +++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 25 deletions(-) diff --git a/haystack/core/serialization.py b/haystack/core/serialization.py index a0c945242ae..ade1b13d304 100644 --- a/haystack/core/serialization.py +++ b/haystack/core/serialization.py @@ -386,31 +386,27 @@ def import_class_by_name(fully_qualified_name: str) -> type[object]: return _import_class_by_name(fully_qualified_name) -def _resolve_coercible_class(type_: Any) -> Any: +def _coercible_classes(type_: Any) -> set[type]: """ - Resolve the class that can deserialize values of `type_` from a plain dictionary. + Collect the distinct classes involved in `type_` that can deserialize a plain dictionary. A class is coercible if it exposes a `from_dict` method (Haystack dataclasses and Components), is a Pydantic - model, or is a standard-library dataclass. Handles plain classes, `list[T]`, and optionals/unions of those. - In unions, the first arm resolving to a coercible class wins. + model, or is a standard-library dataclass. Follows `list[T]` and optionals/unions of those. :param type_: The type to inspect. - :returns: The resolved class, or None if `type_` involves no coercible class. + :returns: The set of coercible classes `type_` involves (empty if none). """ if _is_union_type(type_): + classes: set[type] = set() for arm in get_args(type_): - resolved = _resolve_coercible_class(arm) - if resolved is not None: - return resolved - return None + classes |= _coercible_classes(arm) + return classes if get_origin(type_) is list: args = get_args(type_) - return _resolve_coercible_class(args[0]) if args else None - if not isinstance(type_, type): - return None - if hasattr(type_, "from_dict") or issubclass(type_, BaseModel) or is_dataclass(type_): - return type_ - return None + return _coercible_classes(args[0]) if args else set() + if isinstance(type_, type) and (hasattr(type_, "from_dict") or issubclass(type_, BaseModel) or is_dataclass(type_)): + return {type_} + return set() def _deserialize_one(value: dict[str, Any], target_class: Any) -> Any: @@ -449,11 +445,25 @@ def _deserialize_from_dict(value: Any, target_class: Any) -> Any: return value +def _needs_coercion(value: Any) -> bool: + """Whether `value` carries dictionaries that could be deserialized (a dict, or a list containing one).""" + return isinstance(value, dict) or (isinstance(value, list) and any(isinstance(item, dict) for item in value)) + + def _coerce_input_value(value: Any, socket_types: list[Any]) -> Any: + if not _needs_coercion(value): + return value for type_ in socket_types: - target_class = _resolve_coercible_class(type_) - if target_class is not None: - return _deserialize_from_dict(value, target_class) + classes = _coercible_classes(type_) + if len(classes) > 1: + names = ", ".join(sorted(cls.__name__ for cls in classes)) + raise DeserializationError( + f"Cannot coerce input for socket type '{type_}': it has multiple deserializable members " + f"({names}) and the serialized payload carries no type information to choose between them. " + f"Provide this input as an already-deserialized object." + ) + if classes: + return _deserialize_from_dict(value, next(iter(classes))) return value @@ -465,7 +475,9 @@ def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> di converted into instances of that class. A class is coercible if it exposes a `from_dict` method (such as `ChatMessage` or `Document`), is a Pydantic model, or is a standard-library dataclass. Socket types of the form `T`, `list[T]`, and optionals/unions of those are supported. Values that are already deserialized, or - whose socket types involve no coercible class, are returned unchanged. + whose socket types involve no coercible class, are returned unchanged. A socket type that involves more than + one distinct coercible class (such as `GeneratedAnswer | ExtractedAnswer`) is ambiguous, since the payload + carries no type information to select one; coercing a dictionary against it raises a `DeserializationError`. The dictionaries are expected to be in the format produced by the class's serialization: `to_dict` for `from_dict`-capable classes, `model_dump` for Pydantic models, and `dataclasses.asdict` for standard-library @@ -487,6 +499,8 @@ def coerce_pipeline_inputs(pipeline: "PipelineBase", data: dict[str, Any]) -> di :param pipeline: The pipeline whose input socket types drive the coercion. :param data: The pipeline input data, in nested or flat format. :returns: A new dictionary with the same structure as `data` and deserialized values. + :raises DeserializationError: If a socket type involves more than one distinct coercible class and the + corresponding value is a serialized dictionary. :raises Exception: Any error raised while deserializing a dictionary that does not match the expected format, such as a `from_dict` error or a Pydantic `ValidationError`. """ diff --git a/haystack/tools/component_tool.py b/haystack/tools/component_tool.py index 835ba9c7616..11eaa46371a 100644 --- a/haystack/tools/component_tool.py +++ b/haystack/tools/component_tool.py @@ -11,8 +11,8 @@ from haystack.components.agents.state.state import State from haystack.core.component import Component from haystack.core.serialization import ( + _coercible_classes, _deserialize_from_dict, - _resolve_coercible_class, component_from_dict, component_to_dict, generate_qualified_class_name, @@ -392,10 +392,10 @@ def _convert_param(self, param_value: Any, param_type: type) -> Any: :returns: The converted parameter value. """ - # Types involving a coercible class (e.g. list[ChatMessage] | None) are deserialized element-wise via - # that class; everything else is validated with Pydantic. - target_class = _resolve_coercible_class(param_type) - if target_class is not None: - return _deserialize_from_dict(param_value, target_class) + # Types involving a single coercible class (e.g. list[ChatMessage] | None) are deserialized element-wise + # via that class; everything else is validated with Pydantic. + classes = _coercible_classes(param_type) + if len(classes) == 1: + return _deserialize_from_dict(param_value, next(iter(classes))) return TypeAdapter(param_type).validate_python(param_value) diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py index f9eb425f0e0..647beef1e31 100644 --- a/test/core/test_serialization.py +++ b/test/core/test_serialization.py @@ -12,6 +12,8 @@ from pydantic import BaseModel from haystack import Document +from haystack.components.converters.txt import TextFileToDocument +from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.joiners import BranchJoiner from haystack.core.component import component from haystack.core.errors import DeserializationError, SerializationError @@ -29,6 +31,7 @@ ByteStream, ChatMessage, ChatRole, + ExtractedAnswer, FileContent, GeneratedAnswer, ImageContent, @@ -846,3 +849,48 @@ def test_variadic_socket(self): data = {"value": [message.to_dict() for message in messages], "unrelated": 1} coerced = coerce_pipeline_inputs(pipe, data) assert coerced["value"] == messages + + def test_chat_generator_messages_socket(self): + # OpenAIChatGenerator's `messages` socket is `list[ChatMessage] | str`: the list of serialized messages + # is coerced, since ChatMessage is the only coercible member of the union. + pipe = Pipeline() + pipe.add_component("generator", OpenAIChatGenerator(api_key=Secret.from_token("test-key"))) + messages = [ChatMessage.from_user("Hi"), ChatMessage.from_system("Sys")] + data = {"generator": {"messages": [message.to_dict() for message in messages]}} + coerced = coerce_pipeline_inputs(pipe, data) + assert coerced == {"generator": {"messages": messages}} + + def test_chat_generator_messages_socket_string_passes_through(self): + # The `str` arm of `list[ChatMessage] | str` is not coercible, so a bare string is left untouched. + pipe = Pipeline() + pipe.add_component("generator", OpenAIChatGenerator(api_key=Secret.from_token("test-key"))) + data = {"generator": {"messages": "just a string"}} + coerced = coerce_pipeline_inputs(pipe, data) + assert coerced == {"generator": {"messages": "just a string"}} + + def test_converter_sources_socket(self): + # A converter's `sources` socket is `list[str | Path | ByteStream]`: ByteStream is the only coercible + # member, so a serialized ByteStream is coerced while plain string paths pass through. + pipe = Pipeline() + pipe.add_component("converter", TextFileToDocument()) + byte_stream = ByteStream(data=b"hello", mime_type="text/plain") + data = {"converter": {"sources": ["/path/to/file.txt", byte_stream.to_dict()]}} + coerced = coerce_pipeline_inputs(pipe, data) + assert coerced["converter"]["sources"] == ["/path/to/file.txt", byte_stream] + + def test_ambiguous_union_socket_raises(self): + # A socket that involves more than one coercible class cannot be disambiguated from the payload. + pipe = Pipeline() + pipe.add_component("echo", TypedEcho(GeneratedAnswer | ExtractedAnswer)) + answer = GeneratedAnswer(data="a", query="q", documents=[]) + data = {"echo": {"value": answer.to_dict()}} + with pytest.raises(DeserializationError, match="multiple deserializable members"): + coerce_pipeline_inputs(pipe, data) + + def test_ambiguous_union_socket_passes_through_instances(self): + # An already-deserialized value needs no coercion, so an ambiguous socket does not raise for it. + pipe = Pipeline() + pipe.add_component("echo", TypedEcho(GeneratedAnswer | ExtractedAnswer)) + answer = GeneratedAnswer(data="a", query="q", documents=[]) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": answer}}) + assert coerced["echo"]["value"] is answer From 8cf82ef8bac37ebed5f262ea0ee264b27c323d74 Mon Sep 17 00:00:00 2001 From: Sebastian Husch Lee Date: Thu, 23 Jul 2026 11:18:45 +0200 Subject: [PATCH 6/6] expand tests --- test/core/test_serialization.py | 151 ++++++++++++++++++++------------ 1 file changed, 95 insertions(+), 56 deletions(-) diff --git a/test/core/test_serialization.py b/test/core/test_serialization.py index 647beef1e31..ac6e5446cf3 100644 --- a/test/core/test_serialization.py +++ b/test/core/test_serialization.py @@ -708,18 +708,28 @@ class SerializationEnvelope(BaseModel): value: Any -def serialize_for_client(instance: Any) -> Any: +def serialize_with_haystack(instance: Any) -> Any: + """Serialize the way a client using Haystack's own methods does: via `to_dict`.""" + return instance.to_dict() + + +def serialize_with_pydantic(instance: Any) -> Any: """ - Serialize an instance the way a FastAPI app does: instances arrive inside a loosely-typed Pydantic field, - so Pydantic (not to_dict) produces the JSON payload. + Serialize the way a FastAPI app does: instances arrive inside a loosely-typed Pydantic field, so Pydantic + (not `to_dict`) produces the JSON payload. """ return SerializationEnvelope(value=instance).model_dump(mode="json")["value"] TOOL_CALL = ToolCall(tool_name="t", arguments={"a": 1}, id="1") -COERCIBLE_INSTANCES = [ - # We create a fully loaded Document to test all fields serializations +# A fully loaded ByteStream. It round-trips through `to_dict` (a list of ints) but not through Pydantic, which +# serializes the bytes field as a JSON string that ByteStream.from_dict cannot rebuild. +BYTE_STREAM = ByteStream(data=b"hello", mime_type="text/plain", meta={"k": "v"}) + +# Instances serializable by both a Haystack `to_dict` and a Pydantic `model_dump`. Fully loaded where possible: +# a Document with all non-bytes fields and a ChatMessage with every content-part type. +SHARED_ROUND_TRIP_INSTANCES = [ pytest.param( Document( content="Hello", @@ -730,7 +740,6 @@ def serialize_for_client(instance: Any) -> Any: ), id="Document", ), - # We create a fully loaded ChatMessage to test all possible content type serializations pytest.param( ChatMessage( _role=ChatRole.ASSISTANT, @@ -747,61 +756,57 @@ def serialize_for_client(instance: Any) -> Any: ), id="ChatMessage", ), - pytest.param( - ByteStream(data=b"hello", mime_type="text/plain", meta={"k": "v"}), - id="ByteStream", - marks=pytest.mark.xfail( - reason="Pydantic serializes the bytes field as a JSON string that ByteStream.from_dict cannot rebuild", - raises=TypeError, - strict=True, - ), - ), pytest.param(GeneratedAnswer(data="answer", query="q", documents=[Document(content="d")]), id="GeneratedAnswer"), pytest.param(ImageContent(base64_image="aGVsbG8=", mime_type="image/png"), id="ImageContent"), - pytest.param(PydanticModel(a=1, b="x"), id="pydantic-model"), - pytest.param(PlainDataclass(name="x", count=2), id="stdlib-dataclass"), - pytest.param( - Conversation(messages=[ChatMessage.from_user("Hi")], documents=[Document(content="D")]), - id="pydantic-model-nesting-haystack", - ), ] -class TestCoercePipelineInputsRoundTrip: - """Serialized output is coerced back to an equal instance for every coercible class.""" +def pytest_generate_tests(metafunc): + # Drives the round-trip tests off each subclass's `instances` attribute, so the two serialization methods + # can cover different sets of instances while sharing the same test bodies. + if "coercible_instance" in metafunc.fixturenames and metafunc.cls is not None: + metafunc.parametrize("coercible_instance", metafunc.cls.instances) - @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) - def test_single_socket(self, instance): - pipe = Pipeline() - pipe.add_component("echo", TypedEcho(type(instance))) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) - assert coerced["echo"]["value"] == instance - @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) - def test_list_socket(self, instance): - pipe = Pipeline() - pipe.add_component("echo", TypedEcho(list[type(instance)])) # type: ignore[misc] - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [serialize_for_client(instance)]}}) - assert coerced["echo"]["value"] == [instance] +class CoercePipelineInputsTests: + """ + Shared `coerce_pipeline_inputs` tests. Subclasses set `serialize` (the client-side serialization method) and + `instances` (the round-trip cases valid for that method). + """ - @pytest.mark.parametrize("instance", COERCIBLE_INSTANCES) - def test_optional_socket(self, instance): - pipe = Pipeline() - pipe.add_component("echo", TypedEcho(type(instance) | None)) - coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": serialize_for_client(instance)}}) - assert coerced["echo"]["value"] == instance + @staticmethod + def serialize(instance: Any) -> Any: + raise NotImplementedError + instances: list[Any] = [] -class TestCoercePipelineInputs: @pytest.fixture def pipeline(self): pipe = Pipeline() pipe.add_component("echo", MultiInputEcho()) return pipe + def test_single_socket(self, coercible_instance): + pipe = Pipeline() + pipe.add_component("echo", TypedEcho(type(coercible_instance))) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": self.serialize(coercible_instance)}}) + assert coerced["echo"]["value"] == coercible_instance + + def test_list_socket(self, coercible_instance): + pipe = Pipeline() + pipe.add_component("echo", TypedEcho(list[type(coercible_instance)])) # type: ignore[misc] + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": [self.serialize(coercible_instance)]}}) + assert coerced["echo"]["value"] == [coercible_instance] + + def test_optional_socket(self, coercible_instance): + pipe = Pipeline() + pipe.add_component("echo", TypedEcho(type(coercible_instance) | None)) + coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": self.serialize(coercible_instance)}}) + assert coerced["echo"]["value"] == coercible_instance + def test_nested_format(self, pipeline): messages = [ChatMessage.from_user("Hi"), ChatMessage.from_assistant("Hello")] - data: dict[str, Any] = {"echo": {"messages": [message.to_dict() for message in messages], "top_k": 5}} + data: dict[str, Any] = {"echo": {"messages": [self.serialize(message) for message in messages], "top_k": 5}} coerced = coerce_pipeline_inputs(pipeline, data) assert coerced == {"echo": {"messages": messages, "top_k": 5}} # the input data is not mutated @@ -809,7 +814,7 @@ def test_nested_format(self, pipeline): def test_flat_format(self, pipeline): messages = [ChatMessage.from_user("Hi")] - data = {"messages": [message.to_dict() for message in messages], "top_k": 5} + data = {"messages": [self.serialize(message) for message in messages], "top_k": 5} coerced = coerce_pipeline_inputs(pipeline, data) assert coerced == {"messages": messages, "top_k": 5} @@ -821,9 +826,10 @@ def test_instances_pass_through(self, pipeline): def test_mixed_list(self, pipeline): message = ChatMessage.from_user("Hi") - data = {"echo": {"messages": [message, ChatMessage.from_assistant("Hello").to_dict()]}} + other = ChatMessage.from_assistant("Hello") + data = {"echo": {"messages": [message, self.serialize(other)]}} coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced["echo"]["messages"] == [message, ChatMessage.from_assistant("Hello")] + assert coerced["echo"]["messages"] == [message, other] def test_non_coercible_values_untouched(self, pipeline): data = {"echo": {"messages": [], "top_k": 5, "config": {"a": 1}}} @@ -831,22 +837,22 @@ def test_non_coercible_values_untouched(self, pipeline): assert coerced == data def test_unknown_inputs_untouched(self, pipeline): - message_dict = ChatMessage.from_user("Hi").to_dict() - data = {"unknown_input": [message_dict], "top_k": 5} + message = self.serialize(ChatMessage.from_user("Hi")) + data = {"unknown_input": [message], "top_k": 5} coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"unknown_input": [message_dict], "top_k": 5} + assert coerced == {"unknown_input": [message], "top_k": 5} def test_unknown_component_untouched(self, pipeline): - message_dict = ChatMessage.from_user("Hi").to_dict() - data = {"unknown_component": {"messages": [message_dict]}} + message = self.serialize(ChatMessage.from_user("Hi")) + data = {"unknown_component": {"messages": [message]}} coerced = coerce_pipeline_inputs(pipeline, data) - assert coerced == {"unknown_component": {"messages": [message_dict]}} + assert coerced == {"unknown_component": {"messages": [message]}} def test_variadic_socket(self): pipe = Pipeline() pipe.add_component("joiner", BranchJoiner(list[ChatMessage])) messages = [ChatMessage.from_user("Hi")] - data = {"value": [message.to_dict() for message in messages], "unrelated": 1} + data = {"value": [self.serialize(message) for message in messages], "unrelated": 1} coerced = coerce_pipeline_inputs(pipe, data) assert coerced["value"] == messages @@ -856,7 +862,7 @@ def test_chat_generator_messages_socket(self): pipe = Pipeline() pipe.add_component("generator", OpenAIChatGenerator(api_key=Secret.from_token("test-key"))) messages = [ChatMessage.from_user("Hi"), ChatMessage.from_system("Sys")] - data = {"generator": {"messages": [message.to_dict() for message in messages]}} + data = {"generator": {"messages": [self.serialize(message) for message in messages]}} coerced = coerce_pipeline_inputs(pipe, data) assert coerced == {"generator": {"messages": messages}} @@ -870,7 +876,8 @@ def test_chat_generator_messages_socket_string_passes_through(self): def test_converter_sources_socket(self): # A converter's `sources` socket is `list[str | Path | ByteStream]`: ByteStream is the only coercible - # member, so a serialized ByteStream is coerced while plain string paths pass through. + # member, so a serialized ByteStream is coerced while plain string paths pass through. ByteStream is + # serialized with `to_dict` regardless of the client, since it has no lossless Pydantic JSON form. pipe = Pipeline() pipe.add_component("converter", TextFileToDocument()) byte_stream = ByteStream(data=b"hello", mime_type="text/plain") @@ -883,7 +890,7 @@ def test_ambiguous_union_socket_raises(self): pipe = Pipeline() pipe.add_component("echo", TypedEcho(GeneratedAnswer | ExtractedAnswer)) answer = GeneratedAnswer(data="a", query="q", documents=[]) - data = {"echo": {"value": answer.to_dict()}} + data = {"echo": {"value": self.serialize(answer)}} with pytest.raises(DeserializationError, match="multiple deserializable members"): coerce_pipeline_inputs(pipe, data) @@ -894,3 +901,35 @@ def test_ambiguous_union_socket_passes_through_instances(self): answer = GeneratedAnswer(data="a", query="q", documents=[]) coerced = coerce_pipeline_inputs(pipe, {"echo": {"value": answer}}) assert coerced["echo"]["value"] is answer + + +class TestCoercePipelineInputsHaystackSerialization(CoercePipelineInputsTests): + """Inputs serialized with Haystack's own `to_dict`.""" + + serialize = staticmethod(serialize_with_haystack) + instances = [*SHARED_ROUND_TRIP_INSTANCES, pytest.param(BYTE_STREAM, id="ByteStream")] + + +class TestCoercePipelineInputsPydanticSerialization(CoercePipelineInputsTests): + """Inputs serialized with a Pydantic `model_dump`, as a FastAPI app does.""" + + serialize = staticmethod(serialize_with_pydantic) + instances = [ + *SHARED_ROUND_TRIP_INSTANCES, + pytest.param( + BYTE_STREAM, + id="ByteStream", + marks=pytest.mark.xfail( + reason="Pydantic serializes the bytes field as a JSON string that ByteStream.from_dict cannot rebuild", + raises=TypeError, + strict=True, + ), + ), + # These types have no `to_dict`, so they only arise under Pydantic serialization. + pytest.param(PydanticModel(a=1, b="x"), id="pydantic-model"), + pytest.param(PlainDataclass(name="x", count=2), id="stdlib-dataclass"), + pytest.param( + Conversation(messages=[ChatMessage.from_user("Hi")], documents=[Document(content="D")]), + id="pydantic-model-nesting-haystack", + ), + ]