模块工具

Tool

用于表示语言模型可以为其准备调用的工具的数据类。

文本属性的准确定义，例如name 和description 对语言模型正确准备调用很重要。

参数:

name: 工具的名称。
description: 工具的描述。
parameters: 定义工具所需参数的 JSON schema。
function: 调用工具时将调用的函数。
outputs_to_string: 可选字典，定义如何将工具输出转换为字符串。如果提供了 source，则仅将指定的输出键发送到处理程序。如果省略 source，则将整个工具结果发送到处理程序。示例

{
    "source": "docs", "handler": format_documents
}

inputs_from_state: 可选字典，将 state 键映射到工具参数名称。示例{"repository": "repo"} 将 state 的 "repository" 映射到工具的 "repo" 参数。
outputs_to_state: 可选字典，定义工具输出如何映射到 state 中的键以及可选的处理程序。如果提供了 source，则仅将指定的输出键发送到处理程序。示例

{
    "documents": {"source": "docs", "handler": custom_handler}
}

如果省略 source，则将整个工具结果发送到处理程序。示例

{
    "documents": {"handler": custom_handler}
}

Tool.tool_spec

@property
def tool_spec() -> dict[str, Any]

返回将由语言模型使用的工具规范。

Tool.invoke

def invoke(**kwargs: Any) -> Any

使用提供的关键字参数调用工具。

Tool.to_dict

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

将工具序列化为字典。

返回值:

包含序列化数据的字典。

Tool.from_dict

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

从字典反序列化工具。

参数:

data: 要反序列化的字典。

返回值:

反序列化的工具。

模块 from_function

create_tool_from_function

def create_tool_from_function(
        function: Callable,
        name: Optional[str] = None,
        description: Optional[str] = None,
        inputs_from_state: Optional[dict[str, str]] = None,
        outputs_to_state: Optional[dict[str, dict[str,
                                                  Any]]] = None) -> "Tool"

从函数创建工具实例。

允许自定义工具名称和描述。对于更简单的用例，请考虑使用@tool 装饰器。

使用示例

from typing import Annotated, Literal
from haystack.tools import create_tool_from_function

def get_weather(
    city: Annotated[str, "the city for which to get the weather"] = "Munich",
    unit: Annotated[Literal["Celsius", "Fahrenheit"], "the unit for the temperature"] = "Celsius"):
    '''A simple function to get the current weather for a location.'''
    return f"Weather report for {city}: 20 {unit}, sunny"

tool = create_tool_from_function(get_weather)

print(tool)
>>> Tool(name='get_weather', description='A simple function to get the current weather for a location.',
>>> parameters={
>>> 'type': 'object',
>>> 'properties': {
>>>     'city': {'type': 'string', 'description': 'the city for which to get the weather', 'default': 'Munich'},
>>>     'unit': {
>>>         'type': 'string',
>>>         'enum': ['Celsius', 'Fahrenheit'],
>>>         'description': 'the unit for the temperature',
>>>         'default': 'Celsius',
>>>     },
>>>     }
>>> },
>>> function=<function get_weather at 0x7f7b3a8a9b80>)

参数:

function: 要转换为工具的函数。该函数必须包含所有参数的类型提示。预期函数具有基本的 Python 输入类型（str、int、float、bool、list、dict、tuple）。其他输入类型可能有效，但不保证。如果参数使用typing.Annotated 进行注解，其元数据将用作参数描述。
name: 工具的名称。如果未提供，将使用函数的名称。
description: 工具的描述。如果未提供，将使用函数的文档字符串。要故意将描述留空，请传递一个空字符串。
inputs_from_state: 可选字典，将 state 键映射到工具参数名称。示例{"repository": "repo"} 将 state 的 "repository" 映射到工具的 "repo" 参数。
outputs_to_state: 可选字典，定义工具输出如何映射到 state 和消息处理。示例

{
    "documents": {"source": "docs", "handler": custom_handler},
    "message": {"source": "summary", "handler": format_summary}
}

引发:

ValueError: 如果函数中的任何参数缺少类型提示。
SchemaGenerationError: 如果在生成工具的 JSON schema 时出错。

返回值:

从函数创建的工具。

tool

def tool(
    function: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    inputs_from_state: Optional[dict[str, str]] = None,
    outputs_to_state: Optional[dict[str, dict[str, Any]]] = None
) -> Union[Tool, Callable[[Callable], Tool]]

将函数转换为工具的装饰器。

可以使用或不使用参数：@tool # 无参数 def my_function(): ...

@tool(name="custom_name") # 带参数 def my_function(): ...

使用示例

from typing import Annotated, Literal
from haystack.tools import tool

@tool
def get_weather(
    city: Annotated[str, "the city for which to get the weather"] = "Munich",
    unit: Annotated[Literal["Celsius", "Fahrenheit"], "the unit for the temperature"] = "Celsius"):
    '''A simple function to get the current weather for a location.'''
    return f"Weather report for {city}: 20 {unit}, sunny"

print(get_weather)
>>> Tool(name='get_weather', description='A simple function to get the current weather for a location.',
>>> parameters={
>>> 'type': 'object',
>>> 'properties': {
>>>     'city': {'type': 'string', 'description': 'the city for which to get the weather', 'default': 'Munich'},
>>>     'unit': {
>>>         'type': 'string',
>>>         'enum': ['Celsius', 'Fahrenheit'],
>>>         'description': 'the unit for the temperature',
>>>         'default': 'Celsius',
>>>     },
>>>     }
>>> },
>>> function=<function get_weather at 0x7f7b3a8a9b80>)

参数:

function: 要装饰的函数（不带参数时）
name: 工具的可选自定义名称
description: 可选自定义描述
inputs_from_state: 可选字典，将 state 键映射到工具参数名称
outputs_to_state: 可选字典，定义工具输出如何映射到 state 和消息处理

返回值:

工具实例或将创建工具实例的装饰器函数

模块 component_tool

ComponentTool

一个包装 Haystack 组件的工具，允许 LLM 使用它们作为工具。

ComponentTool 从组件的输入套接字自动生成 LLM 兼容的工具 schema，这些套接字是从组件的run 方法签名和类型提示派生的。

主要功能

从组件输入套接字自动生成 LLM 工具调用 schema
组件输入的类型转换和验证
支持的类型
数据类
数据类列表
基本类型（str、int、float、bool、dict）
基本类型列表
从组件类名自动生成名称
从组件文档字符串提取描述

要使用 ComponentTool，您首先需要一个 Haystack 组件——无论是现有的还是您创建的新组件。通过将组件传递给 ComponentTool 构造函数，可以从组件创建 ComponentTool。下面是一个从现有的 SerperDevWebSearch 组件创建 ComponentTool 的示例。

用法示例

from haystack import component, Pipeline
from haystack.tools import ComponentTool
from haystack.components.websearch import SerperDevWebSearch
from haystack.utils import Secret
from haystack.components.tools.tool_invoker import ToolInvoker
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

# Create a SerperDev search component
search = SerperDevWebSearch(api_key=Secret.from_env_var("SERPERDEV_API_KEY"), top_k=3)

# Create a tool from the component
tool = ComponentTool(
    component=search,
    name="web_search",  # Optional: defaults to "serper_dev_web_search"
    description="Search the web for current information on any topic"  # Optional: defaults to component docstring
)

# Create pipeline with OpenAIChatGenerator and ToolInvoker
pipeline = Pipeline()
pipeline.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini", tools=[tool]))
pipeline.add_component("tool_invoker", ToolInvoker(tools=[tool]))

# Connect components
pipeline.connect("llm.replies", "tool_invoker.messages")

message = ChatMessage.from_user("Use the web search tool to find information about Nikola Tesla")

# Run pipeline
result = pipeline.run({"llm": {"messages": [message]}})

print(result)

ComponentTool.init

def __init__(
    component: Component,
    name: Optional[str] = None,
    description: Optional[str] = None,
    parameters: Optional[dict[str, Any]] = None,
    *,
    outputs_to_string: Optional[dict[str, Union[str, Callable[[Any],
                                                              str]]]] = None,
    inputs_from_state: Optional[dict[str, str]] = None,
    outputs_to_state: Optional[dict[str, dict[str, Union[str,
                                                         Callable]]]] = None
) -> None

从 Haystack 组件创建工具实例。

参数:

component: 要包装为工具的 Haystack 组件。
name: 工具的可选名称（默认为组件类名的 snake_case）。
description: 可选描述（默认为组件的文档字符串）。
parameters: 定义工具所需参数的 JSON schema。如果未提供，则回退到组件的 run 方法签名中定义的参数。
outputs_to_string: 可选字典，定义如何将工具输出转换为字符串。如果提供了 source，则仅将指定的输出键发送到处理程序。如果省略 source，则将整个工具结果发送到处理程序。示例

{
    "source": "docs", "handler": format_documents
}

inputs_from_state: 可选字典，将 state 键映射到工具参数名称。示例{"repository": "repo"} 将 state 的 "repository" 映射到工具的 "repo" 参数。
outputs_to_state: 可选字典，定义工具输出如何映射到 state 中的键以及可选的处理程序。如果提供了 source，则仅将指定的输出键发送到处理程序。示例

{
    "documents": {"source": "docs", "handler": custom_handler}
}

如果省略 source，则将整个工具结果发送到处理程序。示例

{
    "documents": {"handler": custom_handler}
}

引发:

ValueError: 如果组件无效或 schema 生成失败。

ComponentTool.to_dict

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

将 ComponentTool 序列化为字典。

ComponentTool.from_dict

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

从字典反序列化 ComponentTool。

ComponentTool.tool_spec

@property
def tool_spec() -> dict[str, Any]

返回将由语言模型使用的工具规范。

ComponentTool.invoke

def invoke(**kwargs: Any) -> Any

使用提供的关键字参数调用工具。

模块 toolset

Toolkits (工具集)

一组相关的工具，可以作为一个整体进行使用和管理。

Toolset有两个主要目的

将相关工具分组：Toolset允许您将相关工具组织到单个集合中，从而更轻松地在 Haystack 管道中将它们作为一个整体进行管理和使用。

示例:

from haystack.tools import Tool, Toolset
from haystack.components.tools import ToolInvoker

# Define math functions
def add_numbers(a: int, b: int) -> int:
    return a + b

def subtract_numbers(a: int, b: int) -> int:
    return a - b

# Create tools with proper schemas
add_tool = Tool(
    name="add",
    description="Add two numbers",
    parameters={
        "type": "object",
        "properties": {
            "a": {"type": "integer"},
            "b": {"type": "integer"}
        },
        "required": ["a", "b"]
    },
    function=add_numbers
)

subtract_tool = Tool(
    name="subtract",
    description="Subtract b from a",
    parameters={
        "type": "object",
        "properties": {
            "a": {"type": "integer"},
            "b": {"type": "integer"}
        },
        "required": ["a", "b"]
    },
    function=subtract_numbers
)

# Create a toolset with the math tools
math_toolset = Toolset([add_tool, subtract_tool])

# Use the toolset with a ToolInvoker or ChatGenerator component
invoker = ToolInvoker(tools=math_toolset)

动态加载工具的基类：通过继承 Toolset，您可以创建从 OpenAPI URL、MCP 服务器或其他资源等外部源动态加载工具的实现。

示例:

from haystack.core.serialization import generate_qualified_class_name
from haystack.tools import Tool, Toolset
from haystack.components.tools import ToolInvoker

class CalculatorToolset(Toolset):
    '''A toolset for calculator operations.'''

    def __init__(self):
        tools = self._create_tools()
        super().__init__(tools)

    def _create_tools(self):
        # These Tool instances are obviously defined statically and for illustration purposes only.
        # In a real-world scenario, you would dynamically load tools from an external source here.
        tools = []
        add_tool = Tool(
            name="add",
            description="Add two numbers",
            parameters={
                "type": "object",
                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                "required": ["a", "b"],
            },
            function=lambda a, b: a + b,
        )

        multiply_tool = Tool(
            name="multiply",
            description="Multiply two numbers",
            parameters={
                "type": "object",
                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                "required": ["a", "b"],
            },
            function=lambda a, b: a * b,
        )

        tools.append(add_tool)
        tools.append(multiply_tool)

        return tools

    def to_dict(self):
        return {
            "type": generate_qualified_class_name(type(self)),
            "data": {},  # no data to serialize as we define the tools dynamically
        }

    @classmethod
    def from_dict(cls, data):
        return cls()  # Recreate the tools dynamically during deserialization

# Create the dynamic toolset and use it with ToolInvoker
calculator_toolset = CalculatorToolset()
invoker = ToolInvoker(tools=calculator_toolset)

Toolset实现了集合接口（iter、contains、len、getitem），使其行为类似于工具列表。这使其与期望可迭代工具的组件兼容，例如 ToolInvoker 或 Haystack 聊天生成器。

实现用于动态加载工具的自定义 Toolset 子类时

在 init 方法中执行动态加载
如果您的工具是动态定义的，请覆盖 to_dict() 和 from_dict() 方法
如果您的工具从外部源加载，请序列化端点描述符而不是工具实例

Toolset.__post_init__

def __post_init__()

初始化后验证并设置工具集。

这处理了在初始化过程中提供工具的情况。

Toolset.iter

def __iter__() -> Iterator[Tool]

返回此 Toolset 中工具的迭代器。

这允许 Toolset 在需要工具列表的地方使用。

返回值:

一个生成工具实例的迭代器

Toolset.contains

def __contains__(item: Any) -> bool

检查工具是否在此 Toolset 中。

支持按以下方式检查

工具实例：tool in toolset
工具名称："tool_name" in toolset

参数:

item: 工具实例或工具名称字符串

返回值:

如果包含则为 True，否则为 False

Toolset.add

def add(tool: Union[Tool, "Toolset"]) -> None

添加一个新工具或合并另一个 Toolset。

参数:

tool: 要添加的工具实例或另一个 Toolset

引发:

ValueError: 如果添加工具会导致重复的工具名称
TypeError: 如果提供的对象不是 Tool 或 Toolset

Toolset.to_dict

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

将 Toolset 序列化为字典。

返回值:

Toolset 的字典表示。注意：对于子类实现者：默认实现非常适合静态解析工具的情况。然而，如果您的 Toolset 子类从外部源（例如 MCP 服务器、OpenAPI URL 或本地 OpenAPI 规范）动态解析工具实例，您应该考虑序列化端点描述符而不是工具实例本身。此策略保留了 Toolset 的动态性质，并最大限度地减少了序列化大量工具对象可能带来的开销。此外，通过序列化描述符，您可以确保反序列化过程能够准确地重建工具实例，即使它们在上次序列化后已被修改或删除。未能序列化描述符可能会导致加载过时或不正确的工具配置，从而可能导致错误或意外行为。

Toolset.from_dict

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

从字典反序列化 Toolset。

参数:

data: Toolset 的字典表示

返回值:

一个新的 Toolset 实例

Toolset.add

def __add__(other: Union[Tool, "Toolset", list[Tool]]) -> "Toolset"

将此 Toolset 与另一个工具、Toolset或工具列表连接起来。

参数:

other: 要连接的另一个工具、Toolset或工具列表

引发:

TypeError: 如果 other 参数不是 Tool、Toolset或工具列表
ValueError: 如果组合会导致重复的工具名称

返回值:

一个包含所有工具的新 Toolset

Toolset.len

def __len__() -> int

返回此 Toolset 中的工具数量。

返回值:

工具数量

Toolset.getitem

def __getitem__(index)

按索引获取工具。

参数:

index: 要获取的工具的索引

返回值:

指定索引处的工具

模块 工具

Tool

Tool.tool_spec

Tool.invoke

Tool.to_dict

Tool.from_dict

模块 from_function

create_tool_from_function

使用示例

tool

使用示例

模块 component_tool

ComponentTool

用法示例

ComponentTool.__init__

ComponentTool.to_dict

ComponentTool.from_dict

ComponentTool.tool_spec

ComponentTool.invoke

模块 toolset

Toolkits (工具集)

Toolset.__post_init__

Toolset.__iter__

Toolset.__contains__

Toolset.add

Toolset.to_dict

Toolset.from_dict

Toolset.__add__

Toolset.__len__

Toolset.__getitem__

模块工具

ComponentTool.init

Toolset.iter

Toolset.contains

Toolset.add

Toolset.len

Toolset.getitem