feat: add ImageContent dataclass to include images in ChatMessage + OpenAI support (deepset-ai#9626)

Author: anakin87

docs/pydoc/config/data_classess_api.yml

@@ -2,7 +2,7 @@ loaders:
   - type: haystack_pydoc_tools.loaders.CustomPythonLoader
     search_path: [../../../haystack/dataclasses]
     modules:
-      ["answer", "byte_stream", "chat_message", "document", "sparse_embedding", "streaming_chunk"]
+      ["answer", "byte_stream", "chat_message", "document", "image_content", "sparse_embedding", "streaming_chunk"]
     ignore_when_discovered: ["__init__"]
 processors:
   - type: filter

docs/pydoc/config_docusaurus/data_classess_api.yml

@@ -2,7 +2,7 @@ loaders:
   - type: haystack_pydoc_tools.loaders.CustomPythonLoader
     search_path: [../../../haystack/dataclasses]
     modules:
-      ["answer", "byte_stream", "chat_message", "document", "sparse_embedding", "streaming_chunk"]
+      ["answer", "byte_stream", "chat_message", "document", "image_content", "sparse_embedding", "streaming_chunk"]
     ignore_when_discovered: ["__init__"]
 processors:
   - type: filter

haystack/dataclasses/__init__.py

@@ -11,6 +11,7 @@
     "answer": ["Answer", "ExtractedAnswer", "GeneratedAnswer"],
     "byte_stream": ["ByteStream"],
     "chat_message": ["ChatMessage", "ChatRole", "TextContent", "ToolCall", "ToolCallResult"],
+    "image_content": ["ImageContent"],
     "document": ["Document"],
     "sparse_embedding": ["SparseEmbedding"],
     "state": ["State"],
@@ -37,6 +38,7 @@
     from .chat_message import ToolCall as ToolCall
     from .chat_message import ToolCallResult as ToolCallResult
     from .document import Document as Document
+    from .image_content import ImageContent as ImageContent
     from .sparse_embedding import SparseEmbedding as SparseEmbedding
     from .streaming_chunk import AsyncStreamingCallbackT as AsyncStreamingCallbackT
     from .streaming_chunk import ComponentInfo as ComponentInfo

haystack/dataclasses/chat_message.py

@@ -8,6 +8,7 @@
 from typing import Any, Dict, List, Optional, Sequence, Union
 
 from haystack import logging
+from haystack.dataclasses.image_content import ImageContent
 
 logger = logging.getLogger(__name__)
 
@@ -86,7 +87,33 @@ class TextContent:
     text: str
 
 
-ChatMessageContentT = Union[TextContent, ToolCall, ToolCallResult]
+ChatMessageContentT = Union[TextContent, ToolCall, ToolCallResult, ImageContent]
+
+
+def _deserialize_content_part(part: Dict[str, Any]) -> ChatMessageContentT:
+    """
+    Deserialize a single content part of a serialized ChatMessage.
+
+    :param part:
+        A dictionary representing a single content part of a serialized ChatMessage.
+    :returns:
+        A ChatMessageContentT object.
+    :raises ValueError:
+        If the part is not a valid ChatMessageContentT object.
+    """
+    if "text" in part:
+        return TextContent(text=part["text"])
+    if "tool_call" in part:
+        return ToolCall(**part["tool_call"])
+    if "tool_call_result" in part:
+        result = part["tool_call_result"]["result"]
+        origin = ToolCall(**part["tool_call_result"]["origin"])
+        error = part["tool_call_result"]["error"]
+        tcr = ToolCallResult(result=result, origin=origin, error=error)
+        return tcr
+    if "image" in part:
+        return ImageContent(**part["image"])
+    raise ValueError(f"Unsupported content part in the serialized ChatMessage: `{part}`")
 
 
 def _deserialize_content(serialized_content: List[Dict[str, Any]]) -> List[ChatMessageContentT]:
@@ -99,30 +126,29 @@ def _deserialize_content(serialized_content: List[Dict[str, Any]]) -> List[ChatM
     :returns:
         Deserialized `content` field as a list of `ChatMessageContentT` objects.
     """
-    content: List[ChatMessageContentT] = []
-
-    for part in serialized_content:
-        if "text" in part:
-            content.append(TextContent(text=part["text"]))
-        elif "tool_call" in part:
-            content.append(ToolCall(**part["tool_call"]))
-        elif "tool_call_result" in part:
-            result = part["tool_call_result"]["result"]
-            origin = ToolCall(**part["tool_call_result"]["origin"])
-            error = part["tool_call_result"]["error"]
-            tcr = ToolCallResult(result=result, origin=origin, error=error)
-            content.append(tcr)
-        else:
-            raise ValueError(
-                f"Unsupported content part in the serialized ChatMessage: {part}. "
-                "The `content` field of the serialized ChatMessage must be a list of dictionaries, where each "
-                "dictionary contains one of these keys: 'text', 'tool_call', or 'tool_call_result'. "
-                f"Valid formats: [{{'text': 'Hello'}}, "
-                f"{{'tool_call': {{'tool_name': 'search', 'arguments': {{}}, 'id': 'call_123'}}}}, "
-                f"{{'tool_call_result': {{'result': 'data', 'origin': {{...}}, 'error': false}}}}]"
-            )
+    return [_deserialize_content_part(part) for part in serialized_content]
 
-    return content
+
+def _serialize_content_part(part: ChatMessageContentT) -> Dict[str, Any]:
+    """
+    Serialize a single content part of a ChatMessage.
+
+    :param part:
+        A ChatMessageContentT object.
+    :returns:
+        A dictionary representing the content part.
+    :raises TypeError:
+        If the part is not a valid ChatMessageContentT object.
+    """
+    if isinstance(part, TextContent):
+        return {"text": part.text}
+    elif isinstance(part, ToolCall):
+        return {"tool_call": asdict(part)}
+    elif isinstance(part, ToolCallResult):
+        return {"tool_call_result": asdict(part)}
+    elif isinstance(part, ImageContent):
+        return {"image": asdict(part)}
+    raise TypeError(f"Unsupported type in ChatMessage content: `{type(part).__name__}` for `{part}`.")
 
 
 @dataclass
@@ -252,6 +278,22 @@ def tool_call_result(self) -> Optional[ToolCallResult]:
             return tool_call_results[0]
         return None
 
+    @property
+    def images(self) -> List[ImageContent]:
+        """
+        Returns the list of all images contained in the message.
+        """
+        return [content for content in self._content if isinstance(content, ImageContent)]
+
+    @property
+    def image(self) -> Optional[ImageContent]:
+        """
+        Returns the first image contained in the message.
+        """
+        if images := self.images:
+            return images[0]
+        return None
+
     def is_from(self, role: Union[ChatRole, str]) -> bool:
         """
         Check if the message is from a specific role.
@@ -264,16 +306,44 @@ def is_from(self, role: Union[ChatRole, str]) -> bool:
         return self._role == role
 
     @classmethod
-    def from_user(cls, text: str, meta: Optional[Dict[str, Any]] = None, name: Optional[str] = None) -> "ChatMessage":
+    def from_user(
+        cls,
+        text: Optional[str] = None,
+        meta: Optional[Dict[str, Any]] = None,
+        name: Optional[str] = None,
+        *,
+        content_parts: Optional[Sequence[Union[TextContent, str, ImageContent]]] = None,
+    ) -> "ChatMessage":
         """
         Create a message from the user.
 
-        :param text: The text content of the message.
+        :param text: The text content of the message. Specify this or content_parts.
         :param meta: Additional metadata associated with the message.
         :param name: An optional name for the participant. This field is only supported by OpenAI.
+        :param content_parts: A list of content parts to include in the message. Specify this or text.
         :returns: A new ChatMessage instance.
         """
-        return cls(_role=ChatRole.USER, _content=[TextContent(text=text)], _meta=meta or {}, _name=name)
+        if not text and not content_parts:
+            raise ValueError("Either text or content_parts must be provided.")
+        if text and content_parts:
+            raise ValueError("Only one of text or content_parts can be provided.")
+
+        content: Sequence[Union[TextContent, ImageContent]] = []
+
+        if text is not None:
+            content = [TextContent(text=text)]
+        elif content_parts is not None:
+            content = [TextContent(el) if isinstance(el, str) else el for el in content_parts]
+            if not any(isinstance(el, TextContent) for el in content):
+                raise ValueError("The user message must contain at least one textual part.")
+
+            unsupported_parts = [el for el in content if not isinstance(el, (ImageContent, TextContent))]
+            if unsupported_parts:
+                raise ValueError(
+                    f"The user message must contain only text or image parts. Unsupported parts: {unsupported_parts}"
+                )
+
+        return cls(_role=ChatRole.USER, _content=content, _meta=meta or {}, _name=name)
 
     @classmethod
     def from_system(cls, text: str, meta: Optional[Dict[str, Any]] = None, name: Optional[str] = None) -> "ChatMessage":
@@ -343,18 +413,8 @@ def to_dict(self) -> Dict[str, Any]:
         serialized["role"] = self._role.value
         serialized["meta"] = self._meta
         serialized["name"] = self._name
-        content: List[Dict[str, Any]] = []
-        for part in self._content:
-            if isinstance(part, TextContent):
-                content.append({"text": part.text})
-            elif isinstance(part, ToolCall):
-                content.append({"tool_call": asdict(part)})
-            elif isinstance(part, ToolCallResult):
-                content.append({"tool_call_result": asdict(part)})
-            else:
-                raise TypeError(f"Unsupported type in ChatMessage content: `{type(part).__name__}` for `{part}`.")
 
-        serialized["content"] = content
+        serialized["content"] = [_serialize_content_part(part) for part in self._content]
         return serialized
 
     @classmethod
@@ -426,30 +486,56 @@ def to_openai_dict_format(self, require_tool_call_ids: bool = True) -> Dict[str,
                 "A `ChatMessage` must contain at least one `TextContent`, `ToolCall`, or `ToolCallResult`."
             )
         if len(text_contents) + len(tool_call_results) > 1:
-            raise ValueError("A `ChatMessage` can only contain one `TextContent` or one `ToolCallResult`.")
+            raise ValueError(
+                "For OpenAI compatibility, a `ChatMessage` can only contain one `TextContent` or one `ToolCallResult`."
+            )
 
         openai_msg: Dict[str, Any] = {"role": self._role.value}
 
         # Add name field if present
         if self._name is not None:
             openai_msg["name"] = self._name
 
+        # user message
+        if openai_msg["role"] == "user":
+            if len(self._content) == 1:
+                openai_msg["content"] = self.text
+                return openai_msg
+
+            # if the user message contains a list of text and images, OpenAI expects a list of dictionaries
+            content = []
+            for part in self._content:
+                if isinstance(part, TextContent):
+                    content.append({"type": "text", "text": part.text})
+                elif isinstance(part, ImageContent):
+                    image_item: Dict[str, Any] = {
+                        "type": "image_url",
+                        # If no MIME type is provided, default to JPEG.
+                        # OpenAI API appears to tolerate MIME type mismatches.
+                        "image_url": {"url": f"data:{part.mime_type or 'image/jpeg'};base64,{part.base64_image}"},
+                    }
+                    if part.detail:
+                        image_item["image_url"]["detail"] = part.detail
+                    content.append(image_item)
+            openai_msg["content"] = content
+            return openai_msg
+
+        # tool message
         if tool_call_results:
             result = tool_call_results[0]
             openai_msg["content"] = result.result
             if result.origin.id is not None:
                 openai_msg["tool_call_id"] = result.origin.id
             elif require_tool_call_ids:
                 raise ValueError("`ToolCall` must have a non-null `id` attribute to be used with OpenAI.")
-
             # OpenAI does not provide a way to communicate errors in tool invocations, so we ignore the error field
             return openai_msg
 
+        # system and assistant messages
         if text_contents:
             openai_msg["content"] = text_contents[0]
         if tool_calls:
             openai_tool_calls = []
-
             for tc in tool_calls:
                 openai_tool_call = {
                     "type": "function",
@@ -461,7 +547,6 @@ def to_openai_dict_format(self, require_tool_call_ids: bool = True) -> Dict[str,
                 elif require_tool_call_ids:
                     raise ValueError("`ToolCall` must have a non-null `id` attribute to be used with OpenAI.")
                 openai_tool_calls.append(openai_tool_call)
-
             openai_msg["tool_calls"] = openai_tool_calls
         return openai_msg