feat: Pipelines - support connection and conversion between ChatMessage and str (#10507)

* draft

* optimizations

* more

* simplify

* better types

* better types

* license + improvs

* test + improvements + types

* revert unneeded changes

* fix lint

* remove 3.10 guards

* relnote

* fix

* improvements from feedback

* revert wrong change + more tests

Author: anakin87

haystack/core/pipeline/base.py

@@ -21,6 +21,7 @@
     PipelineDrawingError,
     PipelineError,
     PipelineMaxComponentRuns,
+    PipelineRuntimeError,
     PipelineUnmarshalError,
     PipelineValidationError,
 )
@@ -39,7 +40,7 @@
     component_to_dict,
     generate_qualified_class_name,
 )
-from haystack.core.type_utils import _safe_get_origin, _type_name, _types_are_compatible
+from haystack.core.type_utils import _convert_value, _safe_get_origin, _type_name, _types_are_compatible
 from haystack.marshal import Marshaller, YamlMarshaller
 from haystack.utils import is_in_jupyter, type_serialization
 
@@ -499,11 +500,25 @@ def connect(self, sender: str, receiver: str) -> "PipelineBase":  # noqa: PLR091
             [receiver_socket] if receiver_socket else list(receiver_sockets.values())
         )
 
+        is_strict_match = True
+
         # Find all possible connections between these two components
-        possible_connections = []
+        possible_connections: list[tuple[OutputSocket, InputSocket, bool]] = []
         for sender_sock, receiver_sock in itertools.product(sender_socket_candidates, receiver_socket_candidates):
-            if _types_are_compatible(sender_sock.type, receiver_sock.type, self._connection_type_validation):
-                possible_connections.append((sender_sock, receiver_sock))
+            is_compat, is_strict = _types_are_compatible(
+                sender_sock.type, receiver_sock.type, self._connection_type_validation
+            )
+            if is_compat:
+                possible_connections.append((sender_sock, receiver_sock, is_strict))
+
+        # If there are multiple possibilities, prioritize strict matches over convertible ones.
+        # This ensures backward compatibility: previously, pipelines did not allow type conversion.
+        if len(possible_connections) > 1 and self._connection_type_validation:
+            strict_matches = [
+                (out_sock, in_sock, is_strict) for out_sock, in_sock, is_strict in possible_connections if is_strict
+            ]
+            if strict_matches:
+                possible_connections[:] = strict_matches
 
         # We need this status for error messages, since we might need it in multiple places we calculate it here
         status = _connections_status(
@@ -532,11 +547,14 @@ def connect(self, sender: str, receiver: str) -> "PipelineBase":  # noqa: PLR091
             # There's only one possible connection, use it
             sender_socket = possible_connections[0][0]
             receiver_socket = possible_connections[0][1]
+            is_strict_match = possible_connections[0][2]
 
         if len(possible_connections) > 1:
             # There are multiple possible connection, let's try to match them by name
             name_matches = [
-                (out_sock, in_sock) for out_sock, in_sock in possible_connections if in_sock.name == out_sock.name
+                (out_sock, in_sock, is_strict)
+                for out_sock, in_sock, is_strict in possible_connections
+                if in_sock.name == out_sock.name
             ]
             if len(name_matches) != 1:
                 # There's are either no matches or more than one, we can't pick one reliably
@@ -552,6 +570,7 @@ def connect(self, sender: str, receiver: str) -> "PipelineBase":  # noqa: PLR091
             # Get the only possible match
             sender_socket = name_matches[0][0]
             receiver_socket = name_matches[0][1]
+            is_strict_match = name_matches[0][2]
 
         # Connection must be valid on both sender/receiver sides
         if not sender_socket or not receiver_socket or not sender_component_name or not receiver_component_name:
@@ -613,6 +632,7 @@ def connect(self, sender: str, receiver: str) -> "PipelineBase":  # noqa: PLR091
             from_socket=sender_socket,
             to_socket=receiver_socket,
             mandatory=receiver_socket.is_mandatory,
+            convert=not is_strict_match,
         )
         return self
 
@@ -1017,22 +1037,23 @@ def from_template(
             msg += f"Source:\n{rendered}"
             raise PipelineUnmarshalError(msg)
 
-    def _find_receivers_from(self, component_name: str) -> list[tuple[str, OutputSocket, InputSocket]]:
+    def _find_receivers_from(self, component_name: str) -> list[tuple[str, OutputSocket, InputSocket, bool]]:
         """
         Utility function to find all Components that receive input from `component_name`.
 
         :param component_name:
             Name of the sender Component
 
         :returns:
-            List of tuples containing name of the receiver Component and sender OutputSocket
-            and receiver InputSocket instances
+            List of tuples containing name of the receiver Component, sender OutputSocket,
+            receiver InputSocket instances, and a boolean indicating if conversion is needed.
         """
         res = []
         for _, receiver_name, connection in self.graph.edges(nbunch=component_name, data=True):
             sender_socket: OutputSocket = connection["from_socket"]
             receiver_socket: InputSocket = connection["to_socket"]
-            res.append((receiver_name, sender_socket, receiver_socket))
+            convert: bool = connection.get("convert", False)
+            res.append((receiver_name, sender_socket, receiver_socket, convert))
         return res
 
     @staticmethod
@@ -1277,12 +1298,13 @@ def _tiebreak_waiting_components(
 
         return component_name, topological_sort
 
-    @staticmethod
     def _write_component_outputs(
+        self,
+        *,
         component_name: str,
         component_outputs: Mapping[str, Any],
         inputs: dict[str, Any],
-        receivers: list[tuple[str, OutputSocket, InputSocket]],
+        receivers: list[tuple[str, OutputSocket, InputSocket, bool]],
         include_outputs_from: set[str],
     ) -> Mapping[str, Any]:
         """
@@ -1291,16 +1313,40 @@ def _write_component_outputs(
         :param component_name: The name of the component.
         :param component_outputs: The outputs of the component.
         :param inputs: The current global input state.
-        :param receivers: List of tuples containing name of the receiver Component and sender OutputSocket
-            and receiver InputSocket instances.
+        :param receivers: List of tuples containing name of the receiver Component, sender OutputSocket,
+            receiver InputSocket instances, and a boolean indicating if conversion is needed.
         :param include_outputs_from: Set of component names that should always return an output from the pipeline.
         """
-        for receiver_name, sender_socket, receiver_socket in receivers:
+        for receiver_name, sender_socket, receiver_socket, convert in receivers:
             # We either get the value that was produced by the actor or we use the _NO_OUTPUT_PRODUCED class to indicate
             # that the sender did not produce an output for this socket.
             # This allows us to track if a predecessor already ran but did not produce an output.
             value = component_outputs.get(sender_socket.name, _NO_OUTPUT_PRODUCED)
 
+            if value is not _NO_OUTPUT_PRODUCED and convert:
+                try:
+                    value = _convert_value(
+                        value=value, sender_type=sender_socket.type, receiver_type=receiver_socket.type
+                    )
+                except Exception as e:
+                    sender_node = self.graph.nodes.get(component_name)
+                    sender_instance = sender_node.get("instance") if sender_node else None
+                    sender_type_name = type(sender_instance).__name__ if sender_instance else "unknown"
+
+                    receiver_node = self.graph.nodes.get(receiver_name)
+                    receiver_instance = receiver_node.get("instance") if receiver_node else None
+                    receiver_type_name = type(receiver_instance).__name__ if receiver_instance else "unknown"
+
+                    msg = (
+                        f"Failed to perform conversion between components:\n"
+                        f"Sender component: '{component_name}' (type: '{sender_type_name}')\n"
+                        f"Sender socket: '{sender_socket.name}'\n"
+                        f"Receiver component: '{receiver_name}' (type: '{receiver_type_name}')\n"
+                        f"Receiver socket: '{receiver_socket.name}'\n"
+                        f"Error: {e}"
+                    )
+                    raise PipelineRuntimeError(component_name=None, component_type=None, message=msg) from e
+
             if receiver_name not in inputs:
                 inputs[receiver_name] = {}
 
@@ -1332,7 +1378,7 @@ def _write_component_outputs(
 
         # We prune outputs that were consumed by any receiving sockets.
         # All remaining outputs will be added to the final outputs of the pipeline.
-        consumed_outputs = {sender_socket.name for _, sender_socket, __ in receivers}
+        consumed_outputs = {sender_socket.name for _, sender_socket, __, ___ in receivers}
         pruned_outputs = {key: value for key, value in component_outputs.items() if key not in consumed_outputs}
 
         return pruned_outputs
@@ -1431,7 +1477,7 @@ def _merge_super_component_pipelines(self) -> tuple[networkx.MultiDiGraph, dict[
                     # find a matching input socket in the entry point
                     entry_point_sockets = internal_graph.nodes[entry_point]["input_sockets"]
                     for socket_name, socket in entry_point_sockets.items():
-                        if _types_are_compatible(sender_socket.type, socket.type, self._connection_type_validation):
+                        if _types_are_compatible(sender_socket.type, socket.type, self._connection_type_validation)[0]:
                             merged_graph.add_edge(
                                 sender,
                                 entry_point,
@@ -1449,7 +1495,9 @@ def _merge_super_component_pipelines(self) -> tuple[networkx.MultiDiGraph, dict[
                     # find a matching output socket in the exit point
                     exit_point_sockets = internal_graph.nodes[exit_point]["output_sockets"]
                     for socket_name, socket in exit_point_sockets.items():
-                        if _types_are_compatible(socket.type, receiver_socket.type, self._connection_type_validation):
+                        if _types_are_compatible(socket.type, receiver_socket.type, self._connection_type_validation)[
+                            0
+                        ]:
                             merged_graph.add_edge(
                                 exit_point,
                                 receiver,

haystack/core/type_utils.py

@@ -4,24 +4,33 @@
 
 import collections.abc
 from types import NoneType, UnionType
-from typing import Any, TypeVar, Union, get_args, get_origin
+from typing import Any, Union, get_args, get_origin
 
-T = TypeVar("T")
+from haystack.dataclasses import ChatMessage
 
 
-def _types_are_compatible(sender: type | UnionType, receiver: type | UnionType, type_validation: bool = True) -> bool:
+def _types_are_compatible(
+    sender: type | UnionType, receiver: type | UnionType, type_validation: bool = True
+) -> tuple[bool, bool]:
     """
     Determines if two types are compatible based on the specified validation mode.
 
     :param sender: The sender type.
     :param receiver: The receiver type.
     :param type_validation: Whether to perform strict type validation.
-    :return: True if the types are compatible, False otherwise.
+    :return: A tuple where the first element is True if the types are compatible, and the second
+             element is True if they are strictly compatible.
     """
-    if type_validation:
-        return _strict_types_are_compatible(sender, receiver)
-    else:
-        return True
+    if not type_validation:
+        return True, True
+
+    if _strict_types_are_compatible(sender, receiver):
+        return True, True
+
+    if _types_are_convertible(sender, receiver):
+        return True, False
+
+    return False, False
 
 
 def _safe_get_origin(_type: type | UnionType) -> type | None:
@@ -43,13 +52,60 @@ def _safe_get_origin(_type: type | UnionType) -> type | None:
     return origin
 
 
-def _strict_types_are_compatible(sender, receiver):  # pylint: disable=too-many-return-statements
+def _contains_type(container: Any, target: Any) -> bool:
+    """Checks if the container type includes the target type"""
+    if container == target:
+        return True
+    return _safe_get_origin(container) is Union and target in get_args(container)
+
+
+def _types_are_convertible(sender: Any, receiver: Any) -> bool:
+    """
+    Checks whether the sender type is convertible to the receiver type.
+
+    ChatMessage is convertible to str and vice versa.
+    """
+    # Optional[T] must not connect to T
+    if _contains_type(sender, NoneType) and not _contains_type(receiver, NoneType):
+        return False
+
+    # if sender is a single type and receiver is a Union/Optional containing that type, they are convertible
+    # e.g. str is convertible to Optional[str] or Union[str, int] etc.
+    if _contains_type(receiver, sender):
+        return True
+
+    if _contains_type(sender, ChatMessage) and _contains_type(receiver, str):
+        return True
+
+    return _contains_type(sender, str) and _contains_type(receiver, ChatMessage)
+
+
+def _convert_value(value: Any, sender_type: Any, receiver_type: Any) -> Any:
+    """
+    Converts a value from the sender type to the receiver type.
+
+    :param value: The value to convert.
+    :param sender_type: The sender type.
+    :param receiver_type: The receiver type.
+    :return: The converted value.
+    """
+    if _contains_type(sender_type, ChatMessage) and _contains_type(receiver_type, str):
+        if value.text is None:
+            msg = "Cannot convert `ChatMessage` to `str` because it has no text. "
+            raise ValueError(msg)
+        return value.text
+
+    if _contains_type(sender_type, str) and _contains_type(receiver_type, ChatMessage):
+        return ChatMessage.from_user(value)
+
+    return value
+
+
+def _strict_types_are_compatible(sender: Any, receiver: Any) -> bool:  # pylint: disable=too-many-return-statements
     """
     Checks whether the sender type is equal to or a subtype of the receiver type under strict validation.
 
-    Note: this method has no pretense to perform proper type matching. It especially does not deal with aliasing of
-    typing classes such as `List` or `Dict` to their runtime counterparts `list` and `dict`. It also does not deal well
-    with "bare" types, so `List` is treated differently from `List[Any]`, even though they should be the same.
+    Note: this method has no pretense to perform complete type matching.
     Consider simplifying the typing of your components if you observe unexpected errors during component connection.
 
     :param sender: The sender type.

releasenotes/notes/pipeline-chatmsg-str-conversion-bf23ed0d3e1c7287.yaml

@@ -0,0 +1,10 @@
+---
+features:
+  - |
+    Pipelines now support connection and automatic conversion between ``ChatMessage`` and ``str`` types.
+    - When a ``str`` output is connected to a ``ChatMessage`` input, it is automatically converted to a user
+    ``ChatMessage``.
+    - When a ``ChatMessage`` output is connected to a ``str`` input, its ``text`` attribute is automatically
+      extracted. If ``text`` is ``None``, an informative ``PipelineRuntimeError`` is raised.
+    - To maintain backward compatibility, when multiple connections are available, strict type matching is prioritized
+      over conversion.