15825051859

Committed 23 Jun 2025 01:05PM UTC coverage: 90.176% (-0.005%) from 90.181%

Build # 15825051859

Build Type

Pull #9536

github

Committed by

web-flow

Commit Message

Merge 0408d779d into 556dcc9e4

Pull Request Pull Request #9536: feat: Add `finish_reason` field to `StreamingChunk`

Run Details

11575 of 12836 relevant lines covered (90.18%)

0.9 hits per line

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

83.94

haystack/components/tools/tool_invoker.py

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

import asyncio
import inspect
import json
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import Any, Dict, List, Optional, Set, Union

from haystack import component, default_from_dict, default_to_dict, logging
from haystack.components.agents import State
from haystack.core.component.sockets import Sockets
from haystack.dataclasses import ChatMessage, ToolCall
from haystack.dataclasses.streaming_chunk import StreamingCallbackT, StreamingChunk, select_streaming_callback
from haystack.tools import (
    ComponentTool,
    Tool,
    Toolset,
    _check_duplicate_tool_names,
    deserialize_tools_or_toolset_inplace,
    serialize_tools_or_toolset,
)
from haystack.tools.errors import ToolInvocationError
from haystack.tracing.utils import _serializable_value
from haystack.utils.callable_serialization import deserialize_callable, serialize_callable

logger = logging.getLogger(__name__)


class ToolInvokerError(Exception):
    """Base exception class for ToolInvoker errors."""

    def __init__(self, message: str):
        super().__init__(message)


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

    def __init__(self, tool_name: str, available_tools: List[str]):
        message = f"Tool '{tool_name}' not found. Available tools: {', '.join(available_tools)}"
        super().__init__(message)


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

    def __init__(self, tool_name: str, conversion_function: str, error: Exception):
        message = f"Failed to convert tool result from tool {tool_name} using '{conversion_function}'. Error: {error}"
        super().__init__(message)


class ToolOutputMergeError(ToolInvokerError):
    """Exception raised when merging tool outputs into state fails."""

    pass


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

    Also handles reading/writing from a shared `State`.
    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
    from haystack.tools import 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={}
    >>          )
    >>      ]
    >>  }
    ```

    Usage example with a Toolset:
    ```python
    from haystack.dataclasses import ChatMessage, ToolCall
    from haystack.tools import Tool, Toolset
    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)

    # Create a Toolset
    toolset = Toolset([tool])

    # 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 with Toolset
    invoker = ToolInvoker(tools=toolset)
    result = invoker.run(messages=[message])

    print(result)
    """

    def __init__(
        self,
        tools: Union[List[Tool], Toolset],
        raise_on_failure: bool = True,
        convert_result_to_json_string: bool = False,
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        enable_streaming_callback_passthrough: bool = False,
        async_executor: Optional[ThreadPoolExecutor] = None,
    ):
        """
        Initialize the ToolInvoker component.

        :param tools:
            A list of tools that can be invoked or a Toolset instance that can resolve tools.
        :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`.
        :param streaming_callback:
            A callback function that will be called to emit tool results.
            Note that the result is only emitted once it becomes available — it is not
            streamed incrementally in real time.
        :param enable_streaming_callback_passthrough:
            If True, the `streaming_callback` will be passed to the tool invocation if the tool supports it.
            This allows tools to stream their results back to the client.
            Note that this requires the tool to have a `streaming_callback` parameter in its `invoke` method signature.
            If False, the `streaming_callback` will not be passed to the tool invocation.
        :param async_executor:
            Optional ThreadPoolExecutor to use for async calls. If not provided, a single-threaded executor will be
            initialized and used.
        :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.")

        # could be a Toolset instance or a list of Tools
        self.tools = tools
        self.streaming_callback = streaming_callback
        self.enable_streaming_callback_passthrough = enable_streaming_callback_passthrough

        # Convert Toolset to list for internal use
        if isinstance(tools, Toolset):
            converted_tools = list(tools)
        else:
            converted_tools = tools

        _check_duplicate_tool_names(converted_tools)
        tool_names = [tool.name for tool in converted_tools]
        duplicates = {name for name in tool_names if tool_names.count(name) > 1}
        if duplicates:
            raise ValueError(f"Duplicate tool names found: {duplicates}")

        self._tools_with_names = dict(zip(tool_names, converted_tools))
        self.raise_on_failure = raise_on_failure
        self.convert_result_to_json_string = convert_result_to_json_string
        self._owns_executor = async_executor is None
        self.executor = (
            ThreadPoolExecutor(thread_name_prefix=f"async-ToolInvoker-executor-{id(self)}", max_workers=1)
            if async_executor is None
            else async_executor
        )

    def __del__(self):
        """
        Cleanup when the instance is being destroyed.
        """
        if hasattr(self, "_owns_executor") and self._owns_executor and hasattr(self, "executor"):
            self.executor.shutdown(wait=True)

    def shutdown(self):
        """
        Explicitly shutdown the executor if we own it.
        """
        if self._owns_executor:
            self.executor.shutdown(wait=True)

    def _handle_error(self, error: Exception) -> str:
        """
        Handles errors by logging and either raising or returning a fallback error message.

        :param error: The exception instance.
        :returns: The fallback error message when `raise_on_failure` is False.
        :raises: The provided error if `raise_on_failure` is True.
        """
        logger.error("{error_exception}", error_exception=error)
        if self.raise_on_failure:
            # We re-raise the original error maintaining the exception chain
            raise error
        return str(error)

    def _default_output_to_string_handler(self, result: Any) -> str:
        """
        Default handler for converting a tool result to a string.

        :param result: The tool result to convert to a string.
        :returns: The converted tool result as a string.
        """
        # We iterate through all items in result and call to_dict() if present
        # Relevant for a few reasons:
        # - If using convert_result_to_json_string we'd rather convert Haystack objects to JSON serializable dicts
        # - If using default str() we prefer converting Haystack objects to dicts rather than relying on the
        #   __repr__ method
        serializable = _serializable_value(result)

        if self.convert_result_to_json_string:
            try:
                # We disable ensure_ascii so special chars like emojis are not converted
                str_result = json.dumps(serializable, ensure_ascii=False)
            except Exception as error:
                # If the result is not JSON serializable, we fall back to str
                logger.warning(
                    "Tool result is not JSON serializable. Falling back to str conversion. "
                    "Result: {result}\n"
                    "Error: {error}",
                    result=result,
                    err=error,
                )
                str_result = str(result)
            return str_result

        return str(serializable)

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

        :param result:
            The tool result.
        :param tool_call:
            The ToolCall object containing the tool name and arguments.
        :param tool_to_invoke:
            The Tool object that was invoked.
        :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.
        """
        source_key = None
        output_to_string_handler = None
        if tool_to_invoke.outputs_to_string is not None:
            if tool_to_invoke.outputs_to_string.get("source"):
                source_key = tool_to_invoke.outputs_to_string["source"]
            if tool_to_invoke.outputs_to_string.get("handler"):
                output_to_string_handler = tool_to_invoke.outputs_to_string["handler"]

        # If a source key is provided, we extract the result from the source key
        if source_key is not None:
            result_to_convert = result.get(source_key)
        else:
            result_to_convert = result

        # If no handler is provided, we use the default handler
        if output_to_string_handler is None:
            output_to_string_handler = self._default_output_to_string_handler

        error = False
        try:
            tool_result_str = output_to_string_handler(result_to_convert)
        except Exception as e:
            try:
                tool_result_str = self._handle_error(
                    StringConversionError(tool_call.tool_name, output_to_string_handler.__name__, e)
                )
                error = True
            except StringConversionError as conversion_error:
                # If _handle_error re-raises, this properly preserves the chain
                raise conversion_error from e
        return ChatMessage.from_tool(tool_result=tool_result_str, error=error, origin=tool_call)

    def _get_func_params(self, tool: Tool) -> Set:
        """
        Returns the function parameters of the tool's invoke method.

        This method inspects the tool's function signature to determine which parameters the tool accepts.
        """
        # ComponentTool wraps the function with a function that accepts kwargs, so we need to look at input sockets
        # to find out which parameters the tool accepts.
        if isinstance(tool, ComponentTool):
            # mypy doesn't know that ComponentMeta always adds __haystack_input__ to Component
            assert hasattr(tool._component, "__haystack_input__") and isinstance(
                tool._component.__haystack_input__, Sockets
            )
            func_params = set(tool._component.__haystack_input__._sockets_dict.keys())
        else:
            func_params = set(inspect.signature(tool.function).parameters.keys())

        return func_params

    def _inject_state_args(self, tool: Tool, llm_args: Dict[str, Any], state: State) -> Dict[str, Any]:
        """
        Combine LLM-provided arguments (llm_args) with state-based arguments.

        Tool arguments take precedence in the following order:
          - LLM overrides state if the same param is present in both
          - local tool.inputs mappings (if any)
          - function signature name matching
        """
        final_args = dict(llm_args)  # start with LLM-provided
        func_params = self._get_func_params(tool)

        # Determine the source of parameter mappings (explicit tool inputs or direct function parameters)
        # Typically, a "Tool" might have .inputs_from_state = {"state_key": "tool_param_name"}
        if hasattr(tool, "inputs_from_state") and isinstance(tool.inputs_from_state, dict):
            param_mappings = tool.inputs_from_state
        else:
            param_mappings = {name: name for name in func_params}

        # Populate final_args from state if not provided by LLM
        for state_key, param_name in param_mappings.items():
            if param_name not in final_args and state.has(state_key):
                final_args[param_name] = state.get(state_key)

        return final_args

    @staticmethod
    def _merge_tool_outputs(tool: Tool, result: Any, state: State) -> None:
        """
        Merges the tool result into the State.

        This method processes the output of a tool execution and integrates it into the global state.
        It also determines what message, if any, should be returned for further processing in a conversation.

        Processing Steps:
        1. If `result` is not a dictionary, nothing is stored into state and the full `result` is returned.
        2. If the `tool` does not define an `outputs_to_state` mapping nothing is stored into state.
           The return value in this case is simply the full `result` dictionary.
        3. If the tool defines an `outputs_to_state` mapping (a dictionary describing how the tool's output should be
           processed), the method delegates to `_handle_tool_outputs` to process the output accordingly.
           This allows certain fields in `result` to be mapped explicitly to state fields or formatted using custom
           handlers.

        :param tool: Tool instance containing optional `outputs_to_state` mapping to guide result processing.
        :param result: The output from tool execution. Can be a dictionary, or any other type.
        :param state: The global State object to which results should be merged.
        :returns: Three possible values:
            - A string message for conversation
            - The merged result dictionary
            - Or the raw result if not a dictionary
        """
        # If result is not a dictionary we exit
        if not isinstance(result, dict):
            return

        # If there is no specific `outputs_to_state` mapping, we exit
        if not hasattr(tool, "outputs_to_state") or not isinstance(tool.outputs_to_state, dict):
            return

        # Update the state with the tool outputs
        for state_key, config in tool.outputs_to_state.items():
            # Get the source key from the output config, otherwise use the entire result
            source_key = config.get("source", None)
            output_value = result if source_key is None else result.get(source_key)

            # Get the handler function, if any
            handler = config.get("handler", None)

            # Merge other outputs into the state
            state.set(state_key, output_value, handler_override=handler)

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

        :param messages:
            A list of ChatMessage objects.
        :param state: The runtime state that should be used by the tools.
        :param streaming_callback: A callback function that will be called to emit tool results.
            Note that the result is only emitted once it becomes available — it is not
            streamed incrementally in real time.
        :param enable_streaming_callback_passthrough:
            If True, the `streaming_callback` will be passed to the tool invocation if the tool supports it.
            This allows tools to stream their results back to the client.
            Note that this requires the tool to have a `streaming_callback` parameter in its `invoke` method signature.
            If False, the `streaming_callback` will not be passed to the tool invocation.
            If None, the value from the constructor will be used.
        :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.
        :raises ToolOutputMergeError:
            If merging tool outputs into state fails and `raise_on_failure` is True.
        """
        if state is None:
            state = State(schema={})

        resolved_enable_streaming_passthrough = (
            enable_streaming_callback_passthrough
            if enable_streaming_callback_passthrough is not None
            else self.enable_streaming_callback_passthrough
        )

        # Only keep messages with tool calls
        messages_with_tool_calls = [message for message in messages if message.tool_calls]
        streaming_callback = select_streaming_callback(
            init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=False
        )

        tool_messages = []
        for message in messages_with_tool_calls:
            for tool_call in message.tool_calls:
                tool_name = tool_call.tool_name

                # Check if the tool is available, otherwise return an error message
                if tool_name not in self._tools_with_names:
                    error_message = self._handle_error(
                        ToolNotFoundException(tool_name, list(self._tools_with_names.keys()))
                    )
                    tool_messages.append(ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True))
                    continue

                tool_to_invoke = self._tools_with_names[tool_name]

                # 1) Combine user + state inputs
                llm_args = tool_call.arguments.copy()
                final_args = self._inject_state_args(tool_to_invoke, llm_args, state)

                # Check whether to inject streaming_callback
                if (
                    resolved_enable_streaming_passthrough
                    and streaming_callback is not None
                    and "streaming_callback" not in final_args
                ):
                    invoke_params = self._get_func_params(tool_to_invoke)
                    if "streaming_callback" in invoke_params:
                        final_args["streaming_callback"] = streaming_callback

                # 2) Invoke the tool
                try:
                    tool_result = tool_to_invoke.invoke(**final_args)

                except ToolInvocationError as e:
                    error_message = self._handle_error(e)
                    tool_messages.append(ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True))
                    continue

                # 3) Merge outputs into state
                try:
                    self._merge_tool_outputs(tool_to_invoke, tool_result, state)
                except Exception as e:
                    try:
                        error_message = self._handle_error(
                            ToolOutputMergeError(f"Failed to merge tool outputs from tool {tool_name} into State: {e}")
                        )
                        tool_messages.append(
                            ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True)
                        )
                        continue
                    except ToolOutputMergeError as propagated_e:
                        # Re-raise with proper error chain
                        raise propagated_e from e

                # 4) Prepare the tool result ChatMessage message
                tool_messages.append(
                    self._prepare_tool_result_message(
                        result=tool_result, tool_call=tool_call, tool_to_invoke=tool_to_invoke
                    )
                )

                if streaming_callback is not None:
                    streaming_callback(
                        StreamingChunk(
                            content="",
                            index=len(tool_messages) - 1,
                            tool_call_result=tool_messages[-1].tool_call_results[0],
                            start=True,
                            meta={"tool_result": tool_messages[-1].tool_call_results[0].result, "tool_call": tool_call},
                        )
                    )

        # We stream one more chunk that contains a finish_reason if tool_messages were generated
        if len(tool_messages) > 0 and streaming_callback is not None:
            streaming_callback(
                StreamingChunk(
                    content="", finish_reason="tool_call_results", meta={"finish_reason": "tool_call_results"}
                )
            )

        return {"tool_messages": tool_messages, "state": state}

    @component.output_types(tool_messages=List[ChatMessage], state=State)
    async def run_async(
        self,
        messages: List[ChatMessage],
        state: Optional[State] = None,
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        enable_streaming_callback_passthrough: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """
        Asynchronously processes ChatMessage objects containing tool calls and invokes the corresponding tools.

        :param messages:
            A list of ChatMessage objects.
        :param state: The runtime state that should be used by the tools.
        :param streaming_callback: An asynchronous callback function that will be called to emit tool results.
            Note that the result is only emitted once it becomes available — it is not
            streamed incrementally in real time.
        :param enable_streaming_callback_passthrough:
            If True, the `streaming_callback` will be passed to the tool invocation if the tool supports it.
            This allows tools to stream their results back to the client.
            Note that this requires the tool to have a `streaming_callback` parameter in its `invoke` method signature.
            If False, the `streaming_callback` will not be passed to the tool invocation.
            If None, the value from the constructor will be used.
        :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.
        :raises ToolOutputMergeError:
            If merging tool outputs into state fails and `raise_on_failure` is True.
        """
        if state is None:
            state = State(schema={})

        resolved_enable_streaming_passthrough = (
            enable_streaming_callback_passthrough
            if enable_streaming_callback_passthrough is not None
            else self.enable_streaming_callback_passthrough
        )

        # Only keep messages with tool calls
        messages_with_tool_calls = [message for message in messages if message.tool_calls]
        streaming_callback = select_streaming_callback(
            init_callback=self.streaming_callback, runtime_callback=streaming_callback, requires_async=True
        )

        tool_messages = []
        for message in messages_with_tool_calls:
            for tool_call in message.tool_calls:
                tool_name = tool_call.tool_name

                # Check if the tool is available, otherwise return an error message
                if tool_name not in self._tools_with_names:
                    error_message = self._handle_error(
                        ToolNotFoundException(tool_name, list(self._tools_with_names.keys()))
                    )
                    tool_messages.append(ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True))
                    continue

                tool_to_invoke = self._tools_with_names[tool_name]

                # 1) Combine user + state inputs
                llm_args = tool_call.arguments.copy()
                final_args = self._inject_state_args(tool_to_invoke, llm_args, state)

                # Check whether to inject streaming_callback
                if (
                    resolved_enable_streaming_passthrough
                    and streaming_callback is not None
                    and "streaming_callback" not in final_args
                ):
                    invoke_params = self._get_func_params(tool_to_invoke)
                    if "streaming_callback" in invoke_params:
                        final_args["streaming_callback"] = streaming_callback

                # 2) Invoke the tool asynchronously
                try:
                    tool_result = await asyncio.get_running_loop().run_in_executor(
                        self.executor, partial(tool_to_invoke.invoke, **final_args)
                    )

                except ToolInvocationError as e:
                    error_message = self._handle_error(e)
                    tool_messages.append(ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True))
                    continue

                # 3) Merge outputs into state
                try:
                    self._merge_tool_outputs(tool_to_invoke, tool_result, state)
                except Exception as e:
                    try:
                        error_message = self._handle_error(
                            ToolOutputMergeError(f"Failed to merge tool outputs from tool {tool_name} into State: {e}")
                        )
                        tool_messages.append(
                            ChatMessage.from_tool(tool_result=error_message, origin=tool_call, error=True)
                        )
                        continue
                    except ToolOutputMergeError as propagated_e:
                        # Re-raise with proper error chain
                        raise propagated_e from e

                # 4) Prepare the tool result ChatMessage message
                tool_messages.append(
                    self._prepare_tool_result_message(
                        result=tool_result, tool_call=tool_call, tool_to_invoke=tool_to_invoke
                    )
                )

                if streaming_callback is not None:
                    await streaming_callback(
                        StreamingChunk(
                            content="",
                            index=len(tool_messages) - 1,
                            tool_call_result=tool_messages[-1].tool_call_results[0],
                            start=True,
                            meta={"tool_result": tool_messages[-1].tool_call_results[0].result, "tool_call": tool_call},
                        )
                    )

        # We stream one more chunk that contains a finish_reason if tool_messages were generated
        if len(tool_messages) > 0 and streaming_callback is not None:
            await streaming_callback(
                StreamingChunk(
                    content="", finish_reason="tool_call_results", meta={"finish_reason": "tool_call_results"}
                )
            )

        return {"tool_messages": tool_messages, "state": state}

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

        :returns:
            Dictionary with serialized data.
        """
        if self.streaming_callback is not None:
            streaming_callback = serialize_callable(self.streaming_callback)
        else:
            streaming_callback = None

        return default_to_dict(
            self,
            tools=serialize_tools_or_toolset(self.tools),
            raise_on_failure=self.raise_on_failure,
            convert_result_to_json_string=self.convert_result_to_json_string,
            streaming_callback=streaming_callback,
            enable_streaming_callback_passthrough=self.enable_streaming_callback_passthrough,
        )

    @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_or_toolset_inplace(data["init_parameters"], key="tools")
        if data["init_parameters"].get("streaming_callback") is not None:
            data["init_parameters"]["streaming_callback"] = deserialize_callable(
                data["init_parameters"]["streaming_callback"]
            )
        return default_from_dict(cls, data)

deepset-ai / haystack / 15825051859

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