模块 tool_invoker

ToolInvokerError

ToolInvoker 错误的基类异常。

ToolNotFoundException

在可用工具列表中找不到工具时引发的异常。

StringConversionError

将工具结果转换为字符串失败时引发的异常。

ToolOutputMergeError

将工具输出合并到状态失败时引发的异常。

ToolInvoker

根据准备好的工具调用来调用工具，并将结果作为 ChatMessage 对象列表返回。

还可以读取/写入共享的State。初始化时，ToolInvoker 组件会提供一个可用工具列表。运行时，该组件会处理包含工具调用的 ChatMessage 对象列表，并调用相应的工具。工具调用的结果将作为带有工具角色的 ChatMessage 对象列表返回。

使用示例

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={}
>>          )
>>      ]
>>  }

使用 Toolset 的示例

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)

<a id="tool_invoker.ToolInvoker.__init__"></a>

#### ToolInvoker.\_\_init\_\_

```python
def __init__(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,
             max_workers: int = 4)

初始化 ToolInvoker 组件。

参数:

tools: 可调用的工具列表或可以解析工具的 Toolset 实例。
raise_on_failure: 如果为 True，则组件将在出错时（未找到工具、工具调用错误、工具结果转换错误）引发异常。如果为 False，则组件将返回一个带有error=True 的 ChatMessage 对象，并在result.
convert_result_to_json_string: 如果为 True，则工具调用结果将使用json.dumps 转换为字符串。如果为 False，则工具调用结果将使用str.
streaming_callback: 一个将用于发出工具结果的回调函数。请注意，仅当结果可用时才会发出结果 — 它不会实时增量流式传输。
enable_streaming_callback_passthrough: 如果为 True，则streaming_callback 将传递给工具调用，前提是该工具支持此功能。这允许工具将其结果流式传输回客户端。请注意，这要求工具在其invoke 方法签名中具有streaming_callback 参数。如果为 False，则streaming_callback 将不会传递给工具调用。
max_workers: 线程池执行器中使用的最大工作线程数。这也决定了并发工具调用的最大数量。

引发:

ValueError: 如果未提供工具或发现重复的工具名称。

ToolInvoker.run

@component.output_types(tool_messages=list[ChatMessage], state=State)
def run(messages: list[ChatMessage],
        state: Optional[State] = None,
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        enable_streaming_callback_passthrough: Optional[bool] = None,
        tools: Optional[Union[list[Tool], Toolset]] = None) -> dict[str, Any]

处理包含工具调用的 ChatMessage 对象，并在可用时调用相应的工具。

参数:

messages: ChatMessage 对象列表。
state: 工具应使用的运行时状态。
streaming_callback: 一个将用于发出工具结果的回调函数。请注意，仅当结果可用时才会发出结果 — 它不会实时增量流式传输。
enable_streaming_callback_passthrough: 如果为 True，则streaming_callback 将传递给工具调用，前提是该工具支持此功能。这允许工具将其结果流式传输回客户端。请注意，这要求工具在其invoke 方法签名中具有streaming_callback 参数。如果为 False，则streaming_callback 将不会传递给工具调用。如果为 None，则将使用构造函数中的值。
tools: 用于工具调用器的工具列表。如果设置，将覆盖构造函数中设置的工具。

引发:

ToolNotFoundException: 如果在可用工具列表中找不到工具且raise_on_failure 为 True。
ToolInvocationError: 如果工具调用失败且raise_on_failure 为 True。
StringConversionError: 如果工具结果转换为字符串失败且raise_on_failure 为 True。
ToolOutputMergeError: 如果将工具输出合并到状态失败且raise_on_failure 为 True。

返回值:

一个字典，其中包含键tool_messages，其中包含一个工具角色的 ChatMessage 对象列表。每个 ChatMessage 对象都封装了工具调用的结果。

ToolInvoker.invoke_tool_safely

@staticmethod
async def invoke_tool_safely(
        executor: ThreadPoolExecutor, tool_to_invoke: Tool,
        final_args: dict[str, Any]) -> Union[ToolInvocationError, Any]

安全地调用工具并进行适当的异常处理。

ToolInvoker.run_async

@component.output_types(tool_messages=list[ChatMessage], state=State)
async def run_async(
        messages: list[ChatMessage],
        state: Optional[State] = None,
        streaming_callback: Optional[StreamingCallbackT] = None,
        *,
        enable_streaming_callback_passthrough: Optional[bool] = None,
        tools: Optional[Union[list[Tool], Toolset]] = None) -> dict[str, Any]

异步处理包含工具调用的 ChatMessage 对象。

将并发执行多个工具调用。

参数:

messages: ChatMessage 对象列表。
state: 工具应使用的运行时状态。
streaming_callback: 一个异步回调函数，将用于发出工具结果。请注意，仅当结果可用时才会发出结果 — 它不会实时增量流式传输。
enable_streaming_callback_passthrough: 如果为 True，则streaming_callback 将传递给工具调用，前提是该工具支持此功能。这允许工具将其结果流式传输回客户端。请注意，这要求工具在其invoke 方法签名中具有streaming_callback 参数。如果为 False，则streaming_callback 将不会传递给工具调用。如果为 None，则将使用构造函数中的值。
tools: 用于工具调用器的工具列表。如果设置，将覆盖构造函数中设置的工具。

引发:

ToolNotFoundException: 如果在可用工具列表中找不到工具且raise_on_failure 为 True。
ToolInvocationError: 如果工具调用失败且raise_on_failure 为 True。
StringConversionError: 如果工具结果转换为字符串失败且raise_on_failure 为 True。
ToolOutputMergeError: 如果将工具输出合并到状态失败且raise_on_failure 为 True。

返回值:

一个字典，其中包含键tool_messages，其中包含一个工具角色的 ChatMessage 对象列表。每个 ChatMessage 对象都封装了工具调用的结果。

ToolInvoker.to_dict

def to_dict() -> dict[str, Any]

将组件序列化为字典。

返回值:

包含序列化数据的字典。

ToolInvoker.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "ToolInvoker"

从字典反序列化组件。

参数:

data: 要反序列化的字典。

返回值:

反序列化后的组件。