13972131258

Committed 20 Mar 2025 02:43PM UTC coverage: 90.021% (-0.03%) from 90.054%

Build # 13972131258

Build Type

Pull #9069

github

Committed by

web-flow

Commit Message

Merge 8371761b0 into 67ab3788e

Pull Request Pull Request #9069: refactor!: `ChatMessage` serialization-deserialization updates

Run Details

9833 of 10923 relevant lines covered (90.02%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.42

haystack/components/tools/tool_invoker.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import json
from typing import Any, Dict, List

from haystack import component, default_from_dict, default_to_dict
from haystack.dataclasses.chat_message import ChatMessage, ToolCall
from haystack.tools.tool import Tool, ToolInvocationError, _check_duplicate_tool_names, deserialize_tools_inplace

_TOOL_INVOCATION_FAILURE = "Tool invocation failed with error: {error}."
_TOOL_NOT_FOUND = "Tool {tool_name} not found in the list of tools. Available tools are: {available_tools}."
_TOOL_RESULT_CONVERSION_FAILURE = (
    "Failed to convert tool result to string using '{conversion_function}'. Error: {error}."
)


class ToolNotFoundException(Exception):
    """
    Exception raised when a tool is not found in the list of available tools.
    """

    pass


class StringConversionError(Exception):
    """
    Exception raised when the conversion of a tool result to a string fails.
    """

    pass


@component
class ToolInvoker:
    """
    Invokes tools based on prepared tool calls and returns the results as a list of ChatMessage objects.

    At initialization, the ToolInvoker component is provided with a list of available tools.
    At runtime, the component processes a list of ChatMessage object containing tool calls
    and invokes the corresponding tools.
    The results of the tool invocations are returned as a list of ChatMessage objects with tool role.

    Usage example:
    ```python
    from haystack.dataclasses import ChatMessage, ToolCall, Tool
    from haystack.components.tools import ToolInvoker

    # Tool definition
    def dummy_weather_function(city: str):
        return f"The weather in {city} is 20 degrees."

    parameters = {"type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"]}

    tool = Tool(name="weather_tool",
                description="A tool to get the weather",
                function=dummy_weather_function,
                parameters=parameters)

    # Usually, the ChatMessage with tool_calls is generated by a Language Model
    # Here, we create it manually for demonstration purposes
    tool_call = ToolCall(
        tool_name="weather_tool",
        arguments={"city": "Berlin"}
    )
    message = ChatMessage.from_assistant(tool_calls=[tool_call])

    # ToolInvoker initialization and run
    invoker = ToolInvoker(tools=[tool])
    result = invoker.run(messages=[message])

    print(result)
    ```

    ```
    >>  {
    >>      'tool_messages': [
    >>          ChatMessage(
    >>              _role=<ChatRole.TOOL: 'tool'>,
    >>              _content=[
    >>                  ToolCallResult(
    >>                      result='"The weather in Berlin is 20 degrees."',
    >>                      origin=ToolCall(
    >>                          tool_name='weather_tool',
    >>                          arguments={'city': 'Berlin'},
    >>                          id=None
    >>                      )
    >>                  )
    >>              ],
    >>              _meta={}
    >>          )
    >>      ]
    >>  }
    ```
    """

    def __init__(self, tools: List[Tool], raise_on_failure: bool = True, convert_result_to_json_string: bool = False):
        """
        Initialize the ToolInvoker component.

        :param tools:
            A list of tools that can be invoked.
        :param raise_on_failure:
            If True, the component will raise an exception in case of errors
            (tool not found, tool invocation errors, tool result conversion errors).
            If False, the component will return a ChatMessage object with `error=True`
            and a description of the error in `result`.
        :param convert_result_to_json_string:
            If True, the tool invocation result will be converted to a string using `json.dumps`.
            If False, the tool invocation result will be converted to a string using `str`.

        :raises ValueError:
            If no tools are provided or if duplicate tool names are found.
        """

        if not tools:
            raise ValueError("ToolInvoker requires at least one tool to be provided.")
        _check_duplicate_tool_names(tools)

        self.tools = tools
        self._tools_with_names = dict(zip([tool.name for tool in tools], tools))
        self.raise_on_failure = raise_on_failure
        self.convert_result_to_json_string = convert_result_to_json_string

    def _prepare_tool_result_message(self, result: Any, tool_call: ToolCall) -> ChatMessage:
        """
        Prepares a ChatMessage with the result of a tool invocation.

        :param result:
            The tool result.
        :returns:
            A ChatMessage object containing the tool result as a string.

        :raises
            StringConversionError: If the conversion of the tool result to a string fails
            and `raise_on_failure` is True.
        """
        error = False

        if self.convert_result_to_json_string:
            try:
                # We disable ensure_ascii so special chars like emojis are not converted
                tool_result_str = json.dumps(result, ensure_ascii=False)
            except Exception as e:
                if self.raise_on_failure:
                    raise StringConversionError("Failed to convert tool result to string using `json.dumps`") from e
                tool_result_str = _TOOL_RESULT_CONVERSION_FAILURE.format(error=e, conversion_function="json.dumps")
                error = True
            return ChatMessage.from_tool(tool_result=tool_result_str, error=error, origin=tool_call)

        try:
            tool_result_str = str(result)
        except Exception as e:
            if self.raise_on_failure:
                raise StringConversionError("Failed to convert tool result to string using `str`") from e
            tool_result_str = _TOOL_RESULT_CONVERSION_FAILURE.format(error=e, conversion_function="str")
            error = True
        return ChatMessage.from_tool(tool_result=tool_result_str, error=error, origin=tool_call)

    @component.output_types(tool_messages=List[ChatMessage])
    def run(self, messages: List[ChatMessage]) -> Dict[str, Any]:
        """
        Processes ChatMessage objects containing tool calls and invokes the corresponding tools, if available.

        :param messages:
            A list of ChatMessage objects.
        :returns:
            A dictionary with the key `tool_messages` containing a list of ChatMessage objects with tool role.
            Each ChatMessage objects wraps the result of a tool invocation.

        :raises ToolNotFoundException:
            If the tool is not found in the list of available tools and `raise_on_failure` is True.
        :raises ToolInvocationError:
            If the tool invocation fails and `raise_on_failure` is True.
        :raises StringConversionError:
            If the conversion of the tool result to a string fails and `raise_on_failure` is True.
        """
        tool_messages = []

        for message in messages:
            tool_calls = message.tool_calls
            if not tool_calls:
                continue

            for tool_call in tool_calls:
                tool_name = tool_call.tool_name
                tool_arguments = tool_call.arguments

                if not tool_name in self._tools_with_names:
                    msg = _TOOL_NOT_FOUND.format(tool_name=tool_name, available_tools=self._tools_with_names.keys())
                    if self.raise_on_failure:
                        raise ToolNotFoundException(msg)
                    tool_messages.append(ChatMessage.from_tool(tool_result=msg, origin=tool_call, error=True))
                    continue

                tool_to_invoke = self._tools_with_names[tool_name]
                try:
                    tool_result = tool_to_invoke.invoke(**tool_arguments)
                except ToolInvocationError as e:
                    if self.raise_on_failure:
                        raise e
                    msg = _TOOL_INVOCATION_FAILURE.format(error=e)
                    tool_messages.append(ChatMessage.from_tool(tool_result=msg, origin=tool_call, error=True))
                    continue

                tool_message = self._prepare_tool_result_message(tool_result, tool_call)
                tool_messages.append(tool_message)

        return {"tool_messages": tool_messages}

    def to_dict(self) -> Dict[str, Any]:
        """
        Serializes the component to a dictionary.

        :returns:
            Dictionary with serialized data.
        """
        serialized_tools = [tool.to_dict() for tool in self.tools]
        return default_to_dict(
            self,
            tools=serialized_tools,
            raise_on_failure=self.raise_on_failure,
            convert_result_to_json_string=self.convert_result_to_json_string,
        )

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "ToolInvoker":
        """
        Deserializes the component from a dictionary.

        :param data:
            The dictionary to deserialize from.
        :returns:
            The deserialized component.
        """
        deserialize_tools_inplace(data["init_parameters"], key="tools")
        return default_from_dict(cls, data)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	import json	1✔
6	from typing import Any, Dict, List	1✔
7
8	from haystack import component, default_from_dict, default_to_dict	1✔
9	from haystack.dataclasses.chat_message import ChatMessage, ToolCall	1✔
10	from haystack.tools.tool import Tool, ToolInvocationError, _check_duplicate_tool_names, deserialize_tools_inplace	1✔
11
12	_TOOL_INVOCATION_FAILURE = "Tool invocation failed with error: {error}."	1✔
13	_TOOL_NOT_FOUND = "Tool {tool_name} not found in the list of tools. Available tools are: {available_tools}."	1✔
14	_TOOL_RESULT_CONVERSION_FAILURE = (	1✔
15	"Failed to convert tool result to string using '{conversion_function}'. Error: {error}."
16	)
17
18
19	class ToolNotFoundException(Exception):	1✔
20	"""
21	Exception raised when a tool is not found in the list of available tools.
22	"""
23
24	pass	1✔
25
26
27	class StringConversionError(Exception):	1✔
28	"""
29	Exception raised when the conversion of a tool result to a string fails.
30	"""
31
32	pass	1✔
33
34
35	@component	1✔
36	class ToolInvoker:	1✔
37	"""
38	Invokes tools based on prepared tool calls and returns the results as a list of ChatMessage objects.
39
40	At initialization, the ToolInvoker component is provided with a list of available tools.
41	At runtime, the component processes a list of ChatMessage object containing tool calls
42	and invokes the corresponding tools.
43	The results of the tool invocations are returned as a list of ChatMessage objects with tool role.
44
45	Usage example:
46	```python
47	from haystack.dataclasses import ChatMessage, ToolCall, Tool
48	from haystack.components.tools import ToolInvoker
49
50	# Tool definition
51	def dummy_weather_function(city: str):
52	return f"The weather in {city} is 20 degrees."
53
54	parameters = {"type": "object",
55	"properties": {"city": {"type": "string"}},
56	"required": ["city"]}
57
58	tool = Tool(name="weather_tool",
59	description="A tool to get the weather",
60	function=dummy_weather_function,
61	parameters=parameters)
62
63	# Usually, the ChatMessage with tool_calls is generated by a Language Model
64	# Here, we create it manually for demonstration purposes
65	tool_call = ToolCall(
66	tool_name="weather_tool",
67	arguments={"city": "Berlin"}
68	)
69	message = ChatMessage.from_assistant(tool_calls=[tool_call])
70
71	# ToolInvoker initialization and run
72	invoker = ToolInvoker(tools=[tool])
73	result = invoker.run(messages=[message])
74
75	print(result)
76	```
77
78	```
79	>> {
80	>> 'tool_messages': [
81	>> ChatMessage(
82	>> _role=<ChatRole.TOOL: 'tool'>,
83	>> _content=[
84	>> ToolCallResult(
85	>> result='"The weather in Berlin is 20 degrees."',
86	>> origin=ToolCall(
87	>> tool_name='weather_tool',
88	>> arguments={'city': 'Berlin'},
89	>> id=None
90	>> )
91	>> )
92	>> ],
93	>> _meta={}
94	>> )
95	>> ]
96	>> }
97	```
98	"""
99
100	def __init__(self, tools: List[Tool], raise_on_failure: bool = True, convert_result_to_json_string: bool = False):	1✔
101	"""
102	Initialize the ToolInvoker component.
103
104	:param tools:
105	A list of tools that can be invoked.
106	:param raise_on_failure:
107	If True, the component will raise an exception in case of errors
108	(tool not found, tool invocation errors, tool result conversion errors).
109	If False, the component will return a ChatMessage object with `error=True`
110	and a description of the error in `result`.
111	:param convert_result_to_json_string:
112	If True, the tool invocation result will be converted to a string using `json.dumps`.
113	If False, the tool invocation result will be converted to a string using `str`.
114
115	:raises ValueError:
116	If no tools are provided or if duplicate tool names are found.
117	"""
118
119	if not tools:	1✔
120	raise ValueError("ToolInvoker requires at least one tool to be provided.")	1✔
121	_check_duplicate_tool_names(tools)	1✔
122
123	self.tools = tools	1✔
124	self._tools_with_names = dict(zip([tool.name for tool in tools], tools))	1✔
125	self.raise_on_failure = raise_on_failure	1✔
126	self.convert_result_to_json_string = convert_result_to_json_string	1✔
127
128	def _prepare_tool_result_message(self, result: Any, tool_call: ToolCall) -> ChatMessage:	1✔
129	"""
130	Prepares a ChatMessage with the result of a tool invocation.
131
132	:param result:
133	The tool result.
134	:returns:
135	A ChatMessage object containing the tool result as a string.
136
137	:raises
138	StringConversionError: If the conversion of the tool result to a string fails
139	and `raise_on_failure` is True.
140	"""
141	error = False	1✔
142
143	if self.convert_result_to_json_string:	1✔
144	try:	1✔
145	# We disable ensure_ascii so special chars like emojis are not converted
146	tool_result_str = json.dumps(result, ensure_ascii=False)	1✔
147	except Exception as e:	1✔
148	if self.raise_on_failure:	1✔
149	raise StringConversionError("Failed to convert tool result to string using `json.dumps`") from e	1✔
150	tool_result_str = _TOOL_RESULT_CONVERSION_FAILURE.format(error=e, conversion_function="json.dumps")	1✔
151	error = True	1✔
152	return ChatMessage.from_tool(tool_result=tool_result_str, error=error, origin=tool_call)	1✔
153
154	try:	1✔
155	tool_result_str = str(result)	1✔
156	except Exception as e:	×
157	if self.raise_on_failure:	×
158	raise StringConversionError("Failed to convert tool result to string using `str`") from e	×
159	tool_result_str = _TOOL_RESULT_CONVERSION_FAILURE.format(error=e, conversion_function="str")	×
160	error = True	×
161	return ChatMessage.from_tool(tool_result=tool_result_str, error=error, origin=tool_call)	1✔
162
163	@component.output_types(tool_messages=List[ChatMessage])	1✔
164	def run(self, messages: List[ChatMessage]) -> Dict[str, Any]:	1✔
165	"""
166	Processes ChatMessage objects containing tool calls and invokes the corresponding tools, if available.
167
168	:param messages:
169	A list of ChatMessage objects.
170	:returns:
171	A dictionary with the key `tool_messages` containing a list of ChatMessage objects with tool role.
172	Each ChatMessage objects wraps the result of a tool invocation.
173
174	:raises ToolNotFoundException:
175	If the tool is not found in the list of available tools and `raise_on_failure` is True.
176	:raises ToolInvocationError:
177	If the tool invocation fails and `raise_on_failure` is True.
178	:raises StringConversionError:
179	If the conversion of the tool result to a string fails and `raise_on_failure` is True.
180	"""
181	tool_messages = []	1✔
182
183	for message in messages:	1✔
184	tool_calls = message.tool_calls	1✔
185	if not tool_calls:	1✔
186	continue	1✔
187
188	for tool_call in tool_calls:	1✔
189	tool_name = tool_call.tool_name	1✔
190	tool_arguments = tool_call.arguments	1✔
191
192	if not tool_name in self._tools_with_names:	1✔
193	msg = _TOOL_NOT_FOUND.format(tool_name=tool_name, available_tools=self._tools_with_names.keys())	1✔
194	if self.raise_on_failure:	1✔
195	raise ToolNotFoundException(msg)	1✔
196	tool_messages.append(ChatMessage.from_tool(tool_result=msg, origin=tool_call, error=True))	1✔
197	continue	1✔
198
199	tool_to_invoke = self._tools_with_names[tool_name]	1✔
200	try:	1✔
201	tool_result = tool_to_invoke.invoke(**tool_arguments)	1✔
202	except ToolInvocationError as e:	1✔
203	if self.raise_on_failure:	1✔
204	raise e	1✔
205	msg = _TOOL_INVOCATION_FAILURE.format(error=e)	1✔
206	tool_messages.append(ChatMessage.from_tool(tool_result=msg, origin=tool_call, error=True))	1✔
207	continue	1✔
208
209	tool_message = self._prepare_tool_result_message(tool_result, tool_call)	1✔
210	tool_messages.append(tool_message)	1✔
211
212	return {"tool_messages": tool_messages}	1✔
213
214	def to_dict(self) -> Dict[str, Any]:	1✔
215	"""
216	Serializes the component to a dictionary.
217
218	:returns:
219	Dictionary with serialized data.
220	"""
221	serialized_tools = [tool.to_dict() for tool in self.tools]	1✔
222	return default_to_dict(	1✔
223	self,
224	tools=serialized_tools,
225	raise_on_failure=self.raise_on_failure,
226	convert_result_to_json_string=self.convert_result_to_json_string,
227	)
228
229	@classmethod	1✔
230	def from_dict(cls, data: Dict[str, Any]) -> "ToolInvoker":	1✔
231	"""
232	Deserializes the component from a dictionary.
233
234	:param data:
235	The dictionary to deserialize from.
236	:returns:
237	The deserialized component.
238	"""
239	deserialize_tools_inplace(data["init_parameters"], key="tools")	1✔
240	return default_from_dict(cls, data)	1✔

deepset-ai / haystack / 13972131258

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous