wehub-resource-sync
c56bef871b
Sync docs with Docusaurus / sync (push) Waiting to run
Tests / Check if changed (push) Waiting to run
Tests / format (push) Blocked by required conditions
Tests / check-imports (push) Blocked by required conditions
Tests / Unit / macos-latest (push) Blocked by required conditions
Tests / Unit / ubuntu-latest (push) Blocked by required conditions
Tests / Unit / windows-latest (push) Blocked by required conditions
Tests / mypy (push) Blocked by required conditions
Tests / Integration / ubuntu-latest (push) Blocked by required conditions
Tests / Integration / macos-latest (push) Blocked by required conditions
Tests / Integration / windows-latest (push) Blocked by required conditions
Tests / notify-slack-on-failure (push) Blocked by required conditions
Tests / Mark tests as completed (push) Blocked by required conditions
Update Platform Components Table / update (push) Waiting to run
CodeQL / Analyze (python) (push) Waiting to run
Docker image release / Build base image (push) Waiting to run
817 lines
30 KiB
Python
817 lines
30 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
import json
|
|
from collections.abc import Sequence
|
|
from dataclasses import asdict, dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from haystack import logging
|
|
from haystack.dataclasses.file_content import FileContent
|
|
from haystack.dataclasses.image_content import ImageContent
|
|
from haystack.utils.dataclasses import _warn_on_inplace_mutation
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ChatRole(str, Enum):
|
|
"""
|
|
Enumeration representing the roles within a chat.
|
|
"""
|
|
|
|
#: The user role. A message from the user contains only text.
|
|
USER = "user"
|
|
|
|
#: The system role. A message from the system contains only text.
|
|
SYSTEM = "system"
|
|
|
|
#: The assistant role. A message from the assistant can contain text and Tool calls. It can also store metadata.
|
|
ASSISTANT = "assistant"
|
|
|
|
#: The tool role. A message from a tool contains the result of a Tool invocation.
|
|
TOOL = "tool"
|
|
|
|
@staticmethod
|
|
def from_str(string: str) -> "ChatRole":
|
|
"""
|
|
Convert a string to a ChatRole enum.
|
|
"""
|
|
enum_map = {e.value: e for e in ChatRole}
|
|
role = enum_map.get(string)
|
|
if role is None:
|
|
msg = f"Unknown chat role '{string}'. Supported roles are: {list(enum_map.keys())}"
|
|
raise ValueError(msg)
|
|
return role
|
|
|
|
|
|
@_warn_on_inplace_mutation
|
|
@dataclass
|
|
class TextContent:
|
|
"""
|
|
The textual content of a chat message.
|
|
|
|
:param text: The text content of the message.
|
|
"""
|
|
|
|
text: str
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Convert TextContent into a dictionary.
|
|
"""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "TextContent":
|
|
"""
|
|
Create a TextContent from a dictionary.
|
|
"""
|
|
return TextContent(**data)
|
|
|
|
|
|
@_warn_on_inplace_mutation
|
|
@dataclass
|
|
class ToolCall:
|
|
"""
|
|
Represents a Tool call prepared by the model, usually contained in an assistant message.
|
|
|
|
:param id: The ID of the Tool call.
|
|
:param tool_name: The name of the Tool to call.
|
|
:param arguments: The arguments to call the Tool with.
|
|
:param extra: Dictionary of extra information about the Tool call. Use to store provider-specific
|
|
information. To avoid serialization issues, values should be JSON serializable.
|
|
"""
|
|
|
|
tool_name: str
|
|
arguments: dict[str, Any]
|
|
id: str | None = None # noqa: A003
|
|
extra: dict[str, Any] | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Convert ToolCall into a dictionary.
|
|
|
|
:returns: A dictionary with keys 'tool_name', 'arguments', 'id', and 'extra'.
|
|
"""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ToolCall":
|
|
"""
|
|
Creates a new ToolCall object from a dictionary.
|
|
|
|
:param data:
|
|
The dictionary to build the ToolCall object.
|
|
:returns:
|
|
The created object.
|
|
"""
|
|
return ToolCall(**data)
|
|
|
|
|
|
ToolCallResultContentT = str | Sequence[TextContent | ImageContent | FileContent]
|
|
|
|
|
|
@_warn_on_inplace_mutation
|
|
@dataclass
|
|
class ToolCallResult:
|
|
"""
|
|
Represents the result of a Tool invocation.
|
|
|
|
:param result: The result of the Tool invocation.
|
|
:param origin: The Tool call that produced this result.
|
|
:param error: Whether the Tool invocation resulted in an error.
|
|
"""
|
|
|
|
result: ToolCallResultContentT
|
|
origin: ToolCall
|
|
error: bool
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Converts ToolCallResult into a dictionary.
|
|
|
|
:returns: A dictionary with keys 'result', 'origin', and 'error'.
|
|
"""
|
|
serialized = asdict(self)
|
|
if isinstance(self.result, list):
|
|
if not all(isinstance(part, (TextContent, ImageContent, FileContent)) for part in self.result):
|
|
raise ValueError(
|
|
"ToolCallResult result must be a string or a list of TextContent, ImageContent, or FileContent"
|
|
)
|
|
serialized["result"] = [_serialize_content_part(part) for part in self.result]
|
|
return serialized
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ToolCallResult":
|
|
"""
|
|
Creates a ToolCallResult from a dictionary.
|
|
|
|
:param data:
|
|
The dictionary to build the ToolCallResult object.
|
|
:returns:
|
|
The created object.
|
|
"""
|
|
if not all(x in data for x in ["result", "origin", "error"]):
|
|
raise ValueError(
|
|
"Fields `result`, `origin`, `error` are required for ToolCallResult deserialization. "
|
|
f"Received dictionary with keys {list(data.keys())}"
|
|
)
|
|
|
|
result = data["result"]
|
|
if isinstance(result, list):
|
|
result = [_deserialize_content_part(part) for part in result]
|
|
|
|
return ToolCallResult(result=result, origin=ToolCall.from_dict(data["origin"]), error=data["error"])
|
|
|
|
|
|
@_warn_on_inplace_mutation
|
|
@dataclass
|
|
class ReasoningContent:
|
|
"""
|
|
Represents the optional reasoning content prepared by the model, usually contained in an assistant message.
|
|
|
|
:param reasoning_text: The reasoning text produced by the model.
|
|
:param extra: Dictionary of extra information about the reasoning content. Use to store provider-specific
|
|
information. To avoid serialization issues, values should be JSON serializable.
|
|
"""
|
|
|
|
reasoning_text: str
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Convert ReasoningContent into a dictionary.
|
|
|
|
:returns: A dictionary with keys 'reasoning_text', and 'extra'.
|
|
"""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ReasoningContent":
|
|
"""
|
|
Creates a new ReasoningContent object from a dictionary.
|
|
|
|
:param data:
|
|
The dictionary to build the ReasoningContent object.
|
|
:returns:
|
|
The created object.
|
|
"""
|
|
return ReasoningContent(**data)
|
|
|
|
|
|
ChatMessageContentT = TextContent | ToolCall | ToolCallResult | ImageContent | ReasoningContent | FileContent
|
|
|
|
_CONTENT_PART_CLASSES_TO_SERIALIZATION_KEYS: dict[type[ChatMessageContentT], str] = {
|
|
TextContent: "text",
|
|
ToolCall: "tool_call",
|
|
ToolCallResult: "tool_call_result",
|
|
ImageContent: "image",
|
|
ReasoningContent: "reasoning",
|
|
FileContent: "file",
|
|
}
|
|
|
|
|
|
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.
|
|
"""
|
|
# handle flat text format separately
|
|
if "text" in part:
|
|
return TextContent.from_dict(part)
|
|
|
|
for cls, serialization_key in _CONTENT_PART_CLASSES_TO_SERIALIZATION_KEYS.items():
|
|
if serialization_key in part:
|
|
return cls.from_dict(part[serialization_key])
|
|
|
|
# NOTE: this verbose error message provides guidance to LLMs when creating invalid messages during agent runs
|
|
msg = (
|
|
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', 'image', 'file', 'reasoning', 'tool_call', or 'tool_call_result'. "
|
|
"Valid formats: [{'text': 'Hello'}, {'image': {'base64_image': '...', ...}}, "
|
|
"{'file': {'base64_data': '...', ...}}, {'reasoning': {'reasoning_text': 'I think...', 'extra': {...}}}, "
|
|
"{'tool_call': {'tool_name': 'search', 'arguments': {}, 'id': 'call_123'}}, "
|
|
"{'tool_call_result': {'result': 'data', 'origin': {...}, 'error': false}}]"
|
|
)
|
|
raise ValueError(msg)
|
|
|
|
|
|
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.
|
|
"""
|
|
serialization_key = _CONTENT_PART_CLASSES_TO_SERIALIZATION_KEYS.get(type(part))
|
|
if serialization_key is None:
|
|
raise TypeError(f"Unsupported type in ChatMessage content: `{type(part).__name__}` for `{part}`.")
|
|
|
|
# handle flat text format separately
|
|
if isinstance(part, TextContent):
|
|
return part.to_dict()
|
|
|
|
return {serialization_key: part.to_dict()}
|
|
|
|
|
|
@_warn_on_inplace_mutation
|
|
@dataclass
|
|
class ChatMessage:
|
|
"""
|
|
Represents a message in a LLM chat conversation.
|
|
|
|
Use the `from_assistant`, `from_user`, `from_system`, and `from_tool` class methods to create a ChatMessage.
|
|
"""
|
|
|
|
_role: ChatRole
|
|
_content: Sequence[ChatMessageContentT]
|
|
_name: str | None = None
|
|
_meta: dict[str, Any] = field(default_factory=dict, hash=False)
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._content)
|
|
|
|
@property
|
|
def role(self) -> ChatRole:
|
|
"""
|
|
Returns the role of the entity sending the message.
|
|
"""
|
|
return self._role
|
|
|
|
@property
|
|
def meta(self) -> dict[str, Any]:
|
|
"""
|
|
Returns the metadata associated with the message.
|
|
"""
|
|
return self._meta
|
|
|
|
@property
|
|
def name(self) -> str | None:
|
|
"""
|
|
Returns the name associated with the message.
|
|
"""
|
|
return self._name
|
|
|
|
@property
|
|
def texts(self) -> list[str]:
|
|
"""
|
|
Returns the list of all texts contained in the message.
|
|
"""
|
|
return [content.text for content in self._content if isinstance(content, TextContent)]
|
|
|
|
@property
|
|
def text(self) -> str | None:
|
|
"""
|
|
Returns the first text contained in the message.
|
|
"""
|
|
if texts := self.texts:
|
|
return texts[0]
|
|
return None
|
|
|
|
@property
|
|
def tool_calls(self) -> list[ToolCall]:
|
|
"""
|
|
Returns the list of all Tool calls contained in the message.
|
|
"""
|
|
return [content for content in self._content if isinstance(content, ToolCall)]
|
|
|
|
@property
|
|
def tool_call(self) -> ToolCall | None:
|
|
"""
|
|
Returns the first Tool call contained in the message.
|
|
"""
|
|
if tool_calls := self.tool_calls:
|
|
return tool_calls[0]
|
|
return None
|
|
|
|
@property
|
|
def tool_call_results(self) -> list[ToolCallResult]:
|
|
"""
|
|
Returns the list of all Tool call results contained in the message.
|
|
"""
|
|
return [content for content in self._content if isinstance(content, ToolCallResult)]
|
|
|
|
@property
|
|
def tool_call_result(self) -> ToolCallResult | None:
|
|
"""
|
|
Returns the first Tool call result contained in the message.
|
|
"""
|
|
if tool_call_results := self.tool_call_results:
|
|
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) -> ImageContent | None:
|
|
"""
|
|
Returns the first image contained in the message.
|
|
"""
|
|
if images := self.images:
|
|
return images[0]
|
|
return None
|
|
|
|
@property
|
|
def files(self) -> list[FileContent]:
|
|
"""
|
|
Returns the list of all files contained in the message.
|
|
"""
|
|
return [content for content in self._content if isinstance(content, FileContent)]
|
|
|
|
@property
|
|
def file(self) -> FileContent | None:
|
|
"""
|
|
Returns the first file contained in the message.
|
|
"""
|
|
if files := self.files:
|
|
return files[0]
|
|
return None
|
|
|
|
@property
|
|
def reasonings(self) -> list[ReasoningContent]:
|
|
"""
|
|
Returns the list of all reasoning contents contained in the message.
|
|
"""
|
|
return [content for content in self._content if isinstance(content, ReasoningContent)]
|
|
|
|
@property
|
|
def reasoning(self) -> ReasoningContent | None:
|
|
"""
|
|
Returns the first reasoning content contained in the message.
|
|
"""
|
|
if reasonings := self.reasonings:
|
|
return reasonings[0]
|
|
return None
|
|
|
|
def is_from(self, role: ChatRole | str) -> bool:
|
|
"""
|
|
Check if the message is from a specific role.
|
|
|
|
:param role: The role to check against.
|
|
:returns: True if the message is from the specified role, False otherwise.
|
|
"""
|
|
if isinstance(role, str):
|
|
role = ChatRole.from_str(role)
|
|
return self._role == role
|
|
|
|
@classmethod
|
|
def from_user(
|
|
cls,
|
|
text: str | None = None,
|
|
meta: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
*,
|
|
content_parts: Sequence[TextContent | str | ImageContent | FileContent] | None = None,
|
|
) -> "ChatMessage":
|
|
"""
|
|
Create a message from the user.
|
|
|
|
: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.
|
|
:raises ValueError: If neither or both of text and content_parts are provided, or if content_parts is empty.
|
|
:raises TypeError: If a content part is not a str, TextContent, ImageContent, or FileContent.
|
|
"""
|
|
if text is None and content_parts is None:
|
|
raise ValueError("Either text or content_parts must be provided.")
|
|
if text is not None and content_parts is not None:
|
|
raise ValueError("Only one of text or content_parts can be provided.")
|
|
|
|
content: list[TextContent | ImageContent | FileContent] = []
|
|
|
|
if text is not None:
|
|
content = [TextContent(text=text)]
|
|
elif content_parts is not None:
|
|
for part in content_parts:
|
|
if isinstance(part, str):
|
|
content.append(TextContent(text=part))
|
|
elif isinstance(part, (TextContent, ImageContent, FileContent)):
|
|
content.append(part)
|
|
else:
|
|
raise TypeError(f"The user message must contain only text or image parts. Unsupported part: {part}")
|
|
if len(content) == 0:
|
|
raise ValueError("The user message must contain at least one content part (text, image, file).")
|
|
|
|
return cls(_role=ChatRole.USER, _content=content, _meta=meta or {}, _name=name)
|
|
|
|
@classmethod
|
|
def from_system(cls, text: str, meta: dict[str, Any] | None = None, name: str | None = None) -> "ChatMessage":
|
|
"""
|
|
Create a message from the system.
|
|
|
|
:param text: The text content of the message.
|
|
:param meta: Additional metadata associated with the message.
|
|
:param name: An optional name for the participant. This field is only supported by OpenAI.
|
|
:returns: A new ChatMessage instance.
|
|
"""
|
|
return cls(_role=ChatRole.SYSTEM, _content=[TextContent(text=text)], _meta=meta or {}, _name=name)
|
|
|
|
@classmethod
|
|
def from_assistant(
|
|
cls,
|
|
text: str | None = None,
|
|
meta: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
tool_calls: list[ToolCall] | None = None,
|
|
*,
|
|
reasoning: str | ReasoningContent | None = None,
|
|
) -> "ChatMessage":
|
|
"""
|
|
Create a message from the assistant.
|
|
|
|
:param text: The text content of the message.
|
|
:param meta: Additional metadata associated with the message.
|
|
:param name: An optional name for the participant. This field is only supported by OpenAI.
|
|
:param tool_calls: The Tool calls to include in the message.
|
|
:param reasoning: The reasoning content to include in the message.
|
|
:returns: A new ChatMessage instance.
|
|
:raises TypeError: If `reasoning` is not a string or ReasoningContent object.
|
|
"""
|
|
content: list[ChatMessageContentT] = []
|
|
if reasoning:
|
|
if isinstance(reasoning, str):
|
|
content.append(ReasoningContent(reasoning_text=reasoning))
|
|
elif isinstance(reasoning, ReasoningContent):
|
|
content.append(reasoning)
|
|
else:
|
|
raise TypeError(f"reasoning must be a string or a ReasoningContent object, got {type(reasoning)}")
|
|
if text is not None:
|
|
content.append(TextContent(text=text))
|
|
if tool_calls:
|
|
content.extend(tool_calls)
|
|
|
|
return cls(_role=ChatRole.ASSISTANT, _content=content, _meta=meta or {}, _name=name)
|
|
|
|
@classmethod
|
|
def from_tool(
|
|
cls,
|
|
tool_result: ToolCallResultContentT,
|
|
origin: ToolCall,
|
|
error: bool = False,
|
|
meta: dict[str, Any] | None = None,
|
|
) -> "ChatMessage":
|
|
"""
|
|
Create a message from a Tool.
|
|
|
|
:param tool_result: The result of the Tool invocation.
|
|
:param origin: The Tool call that produced this result.
|
|
:param error: Whether the Tool invocation resulted in an error.
|
|
:param meta: Additional metadata associated with the message.
|
|
:returns: A new ChatMessage instance.
|
|
"""
|
|
return cls(
|
|
_role=ChatRole.TOOL,
|
|
_content=[ToolCallResult(result=tool_result, origin=origin, error=error)],
|
|
_meta=meta or {},
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Converts ChatMessage into a dictionary.
|
|
|
|
:returns:
|
|
Serialized version of the object.
|
|
"""
|
|
|
|
serialized: dict[str, Any] = {}
|
|
serialized["role"] = self._role.value
|
|
serialized["meta"] = self._meta
|
|
serialized["name"] = self._name
|
|
|
|
serialized["content"] = [_serialize_content_part(part) for part in self._content]
|
|
return serialized
|
|
|
|
def _to_trace_dict(self) -> dict[str, Any]:
|
|
"""
|
|
Convert the ChatMessage to a dictionary representation for tracing.
|
|
|
|
For Image Content objects, the base64_image is replaced with a placeholder string to avoid sending large
|
|
payloads to the tracing backend.
|
|
|
|
:returns:
|
|
Serialized version of the object only for tracing purposes.
|
|
"""
|
|
|
|
serialized: dict[str, Any] = {}
|
|
serialized["role"] = self._role.value
|
|
serialized["meta"] = self._meta
|
|
serialized["name"] = self._name
|
|
|
|
serialized["content"] = []
|
|
for part in self._content:
|
|
serialized_part = _serialize_content_part(part)
|
|
if isinstance(part, ImageContent):
|
|
serialized_part["image"] = part._to_trace_dict()
|
|
elif isinstance(part, FileContent):
|
|
serialized_part["file"] = part._to_trace_dict()
|
|
serialized["content"].append(serialized_part)
|
|
|
|
return serialized
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ChatMessage":
|
|
"""
|
|
Creates a new ChatMessage object from a dictionary.
|
|
|
|
:param data:
|
|
The dictionary to build the ChatMessage object.
|
|
:returns:
|
|
The created object.
|
|
:raises ValueError: If the `role` field is missing from the dictionary.
|
|
:raises TypeError: If the `content` field is not a list or string.
|
|
"""
|
|
|
|
# NOTE: this verbose error message provides guidance to LLMs when creating invalid messages during agent runs
|
|
if "role" not in data and "_role" not in data:
|
|
raise ValueError(
|
|
"The `role` field is required in the message dictionary. "
|
|
f"Expected a dictionary with 'role' field containing one of: {[role.value for role in ChatRole]}. "
|
|
f"Common roles are 'user' (for user messages) and 'assistant' (for AI responses). "
|
|
f"Received dictionary with keys: {list(data.keys())}"
|
|
)
|
|
|
|
if "content" in data:
|
|
init_params: dict[str, Any] = {
|
|
"_role": ChatRole(data["role"]),
|
|
"_name": data.get("name"),
|
|
"_meta": data.get("meta") or {},
|
|
}
|
|
|
|
if isinstance(data["content"], list):
|
|
# current format - the serialized `content` field is a list of dictionaries
|
|
init_params["_content"] = [_deserialize_content_part(part) for part in data["content"]]
|
|
elif isinstance(data["content"], str):
|
|
# pre 2.9.0 format - the `content` field is a string
|
|
init_params["_content"] = [TextContent(text=data["content"])]
|
|
else:
|
|
raise TypeError(f"Unsupported content type in serialized ChatMessage: `{(data['content'])}`")
|
|
return cls(**init_params)
|
|
|
|
if "_content" in data:
|
|
# format for versions >=2.9.0 and <2.12.0 - the serialized `_content` field is a list of dictionaries
|
|
return cls(
|
|
_role=ChatRole(data["_role"]),
|
|
_content=[_deserialize_content_part(part) for part in data["_content"]],
|
|
_name=data.get("_name"),
|
|
_meta=data.get("_meta") or {},
|
|
)
|
|
|
|
raise ValueError(f"Missing 'content' or '_content' in serialized ChatMessage: `{data}`")
|
|
|
|
def to_openai_dict_format(self, require_tool_call_ids: bool = True) -> dict[str, Any]:
|
|
"""
|
|
Convert a ChatMessage to the dictionary format expected by OpenAI's Chat Completions API.
|
|
|
|
:param require_tool_call_ids:
|
|
If True (default), enforces that each Tool Call includes a non-null `id` attribute.
|
|
Set to False to allow Tool Calls without `id`, which may be suitable for shallow OpenAI-compatible APIs.
|
|
:returns:
|
|
The ChatMessage in the format expected by OpenAI's Chat Completions API.
|
|
|
|
:raises ValueError:
|
|
If the message format is invalid, or if `require_tool_call_ids` is True and any Tool Call is missing an
|
|
`id` attribute.
|
|
"""
|
|
if not self.texts and not self.tool_calls and not self.tool_call_results and not self.images and not self.files:
|
|
raise ValueError(
|
|
"A `ChatMessage` must contain at least one `TextContent`, `ToolCall`, "
|
|
"`ToolCallResult`, `ImageContent`, or `FileContent`."
|
|
)
|
|
if len(self.tool_call_results) > 0 and len(self._content) > 1:
|
|
raise ValueError(
|
|
"For OpenAI compatibility, a `ChatMessage` with a `ToolCallResult` cannot contain any other content."
|
|
)
|
|
|
|
openai_msg: dict[str, Any] = {"role": self._role.value}
|
|
|
|
if self._name is not None:
|
|
openai_msg["name"] = self._name
|
|
|
|
if openai_msg["role"] == "user":
|
|
return self._user_message_to_openai(openai_msg)
|
|
|
|
if self.tool_call_results:
|
|
return self._tool_result_message_to_openai(openai_msg, require_tool_call_ids)
|
|
|
|
return self._system_assistant_message_to_openai(openai_msg, require_tool_call_ids)
|
|
|
|
def _user_message_to_openai(self, openai_msg: dict[str, Any]) -> dict[str, Any]:
|
|
"""Build OpenAI dict for a user message."""
|
|
if len(self._content) == 1 and isinstance(self._content[0], TextContent):
|
|
openai_msg["content"] = self.text
|
|
return openai_msg
|
|
|
|
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)
|
|
elif isinstance(part, FileContent):
|
|
file_item: dict[str, Any] = {
|
|
"type": "file",
|
|
"file": {
|
|
"file_data": f"data:{part.mime_type or 'application/pdf'};base64,{part.base64_data}",
|
|
# Filename is optional but if not provided, OpenAI expects a file_id of a previous file upload.
|
|
# We use a dummy filename.
|
|
"filename": part.filename or "filename",
|
|
},
|
|
}
|
|
content.append(file_item)
|
|
openai_msg["content"] = content
|
|
return openai_msg
|
|
|
|
def _tool_result_message_to_openai(self, openai_msg: dict[str, Any], require_tool_call_ids: bool) -> dict[str, Any]:
|
|
"""Build OpenAI dict for a tool result message."""
|
|
result = self.tool_call_results[0]
|
|
if isinstance(result.result, str):
|
|
openai_msg["content"] = result.result
|
|
# OpenAI Chat Completions API does not support multimodal tool results
|
|
elif isinstance(result.result, list) and all(isinstance(part, TextContent) for part in result.result):
|
|
openai_msg["content"] = [{"type": "text", "text": part.text} for part in result.result]
|
|
else:
|
|
raise ValueError(
|
|
f"Unsupported tool result: {result}. If you need to pass images in tool results, "
|
|
"use OpenAI Responses API instead."
|
|
)
|
|
|
|
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
|
|
|
|
def _system_assistant_message_to_openai(
|
|
self, openai_msg: dict[str, Any], require_tool_call_ids: bool
|
|
) -> dict[str, Any]:
|
|
"""Build OpenAI dict for system and assistant messages."""
|
|
# OpenAI Chat Completions API does not support reasoning content, so we ignore it
|
|
if self.texts:
|
|
openai_msg["content"] = self.texts[0]
|
|
if self.tool_calls:
|
|
openai_tool_calls = []
|
|
for tc in self.tool_calls:
|
|
openai_tool_call = {
|
|
"type": "function",
|
|
# We disable ensure_ascii so special chars like emojis are not converted
|
|
"function": {"name": tc.tool_name, "arguments": json.dumps(tc.arguments, ensure_ascii=False)},
|
|
}
|
|
if tc.id is not None:
|
|
openai_tool_call["id"] = tc.id
|
|
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
|
|
|
|
@staticmethod
|
|
def _validate_openai_message(message: dict[str, Any]) -> None:
|
|
"""
|
|
Validate that a message dictionary follows OpenAI's Chat API format.
|
|
|
|
:param message: The message dictionary to validate
|
|
:raises ValueError: If the message format is invalid
|
|
"""
|
|
if "role" not in message:
|
|
raise ValueError("The `role` field is required in the message dictionary.")
|
|
|
|
role = message["role"]
|
|
content = message.get("content")
|
|
tool_calls = message.get("tool_calls")
|
|
|
|
if role not in ["assistant", "user", "system", "developer", "tool"]:
|
|
raise ValueError(f"Unsupported role: {role}")
|
|
|
|
if role == "assistant":
|
|
if not content and not tool_calls:
|
|
raise ValueError("For assistant messages, either `content` or `tool_calls` must be present.")
|
|
if tool_calls:
|
|
for tc in tool_calls:
|
|
if "function" not in tc:
|
|
raise ValueError("Tool calls must contain the `function` field")
|
|
elif not content:
|
|
raise ValueError(f"The `content` field is required for {role} messages.")
|
|
|
|
@classmethod
|
|
def from_openai_dict_format(cls, message: dict[str, Any]) -> "ChatMessage":
|
|
"""
|
|
Create a ChatMessage from a dictionary in the format expected by OpenAI's Chat API.
|
|
|
|
NOTE: While OpenAI's API requires `tool_call_id` in both tool calls and tool messages, this method
|
|
accepts messages without it to support shallow OpenAI-compatible APIs.
|
|
If you plan to use the resulting ChatMessage with OpenAI, you must include `tool_call_id` or you'll
|
|
encounter validation errors.
|
|
|
|
:param message:
|
|
The OpenAI dictionary to build the ChatMessage object.
|
|
:returns:
|
|
The created ChatMessage object.
|
|
|
|
:raises ValueError:
|
|
If the message dictionary is missing required fields.
|
|
"""
|
|
cls._validate_openai_message(message)
|
|
|
|
role = message["role"]
|
|
content = message.get("content")
|
|
name = message.get("name")
|
|
tool_calls = message.get("tool_calls")
|
|
tool_call_id = message.get("tool_call_id")
|
|
|
|
if role == "assistant":
|
|
haystack_tool_calls = None
|
|
if tool_calls:
|
|
haystack_tool_calls = []
|
|
for tc in tool_calls:
|
|
haystack_tc = ToolCall(
|
|
id=tc.get("id"),
|
|
tool_name=tc["function"]["name"],
|
|
arguments=json.loads(tc["function"]["arguments"]),
|
|
)
|
|
haystack_tool_calls.append(haystack_tc)
|
|
return cls.from_assistant(text=content, name=name, tool_calls=haystack_tool_calls)
|
|
|
|
assert content is not None # ensured by _validate_openai_message, but we need to make mypy happy
|
|
|
|
if role == "user":
|
|
return cls.from_user(text=content, name=name)
|
|
if role in ["system", "developer"]:
|
|
return cls.from_system(text=content, name=name)
|
|
|
|
if isinstance(content, list):
|
|
if not all("text" in el for el in content):
|
|
raise ValueError("To be used with OpenAI, tool results must be a string or a list of TextContent")
|
|
content = [TextContent(text=el["text"]) for el in content]
|
|
return cls.from_tool(
|
|
tool_result=content, origin=ToolCall(id=tool_call_id, tool_name="", arguments={}), error=False
|
|
)
|