Haystack
获取企业支持
开始
📘 **TELUS Agriculture & Consumer Goods** 如何通过 **Haystack Agents** 转变促销交易

定义和运行工具


在 Colab 中打开 下载
最后更新：2025年4月10日

在此 notebook 中，我们介绍了我们在 Haystack 中开发的用于工具/函数调用支持的功能。

  • 我们重构了 `ChatMessage` 数据类，使其更灵活、更具未来兼容性。
  • 我们引入了一些新的数据类：`ToolCall`、`ToolCallResult` 和 `Tool`。
  • 我们为 `OpenAIChatGenerator` 和其他聊天生成器添加了对工具的支持。
  • 我们引入了 `ToolInvoker` 组件，用于实际执行由语言模型准备的工具调用。

我们将首先介绍新功能，然后展示两个示例

  • 一个基本的助手，它可以通过使用天气工具或依赖自身知识来回答用户问题。
  • 助手的一个改进版本，可以在将工具的输出呈现给用户之前对其进行进一步处理。

有关涉及多个工具和人工干预的更复杂示例，请参阅本教程：使用函数调用构建聊天代理。

! pip install haystack-ai "sentence-transformers>=3.0.0"

import os
from getpass import getpass
from rich import print

if "OPENAI_API_KEY" not in os.environ:
  os.environ["OPENAI_API_KEY"] = getpass("Enter OpenAI API key:")

新的实验性功能

重构了 `ChatMessage` 数据类、`ToolCall` 和 `ToolCallResult`

`ChatMessage` 数据类已更新，因此 `content` 字段不再仅仅是字符串 (`str`)；它现在是一个列表，可以包含各种类型，包括 `TextContent`、`ToolCall` 和 `ToolCallResult`。

类方法 `from_user`、`from_system`、`from_assistant` 和 `from_tool`（新添加的）是创建 `ChatMessage` 实例的推荐方式。

此外，我们还引入了

  • `ToolCall` 数据类：表示由语言模型准备的工具调用。
  • `ToolCallResult` 数据类：表示工具调用的结果。

让我们看一些示例。

from haystack.dataclasses import ChatMessage, ToolCall

user_message = ChatMessage.from_user("What is the capital of Australia?")
print(user_message)

print(user_message.text)
print(user_message.texts)

print(user_message.tool_call)
print(user_message.tool_calls)

print(user_message.tool_call_result)
print(user_message.tool_call_results)

ChatMessage(
    _role=<ChatRole.USER: 'user'>,
    _content=[TextContent(text='What is the capital of Australia?')],
    _meta={}
)

What is the capital of Australia?

['What is the capital of Australia?']

None

[]

None

[]

assistant_message = ChatMessage.from_assistant("How can I assist you today?")
print(assistant_message)

print(assistant_message.text)
print(assistant_message.texts)

print(assistant_message.tool_call)
print(assistant_message.tool_calls)

print(assistant_message.tool_call_result)
print(assistant_message.tool_call_results)

ChatMessage(
    _role=<ChatRole.ASSISTANT: 'assistant'>,
    _content=[TextContent(text='How can I assist you today?')],
    _meta={}
)

How can I assist you today?

['How can I assist you today?']

None

[]

None

[]

tool_call = ToolCall(tool_name="weather_tool", arguments={"location": "Rome"})

assistant_message_w_tool_call = ChatMessage.from_assistant(tool_calls=[tool_call])

print(assistant_message_w_tool_call.text)
print(assistant_message_w_tool_call.texts)

print(assistant_message_w_tool_call.tool_call)
print(assistant_message_w_tool_call.tool_calls)

print(assistant_message_w_tool_call.tool_call_result)
print(assistant_message_w_tool_call.tool_call_results)

None

[]

ToolCall(tool_name='weather_tool', arguments={'location': 'Rome'}, id=None)

[ToolCall(tool_name='weather_tool', arguments={'location': 'Rome'}, id=None)]

None

[]

tool_message = ChatMessage.from_tool(tool_result="temperature: 25°C", origin=tool_call, error=False)

print(tool_message.text)
print(tool_message.texts)

print(tool_message.tool_call)
print(tool_message.tool_calls)

print(tool_message.tool_call_result)
print(tool_message.tool_call_results)

None

[]

None

[]

ToolCallResult(
    result='temperature: 25°C',
    origin=ToolCall(tool_name='weather_tool', arguments={'location': 'Rome'}, id=None),
    error=False
)

[
    ToolCallResult(
        result='temperature: 25°C',
        origin=ToolCall(tool_name='weather_tool', arguments={'location': 'Rome'}, id=None),
        error=False
    )
]

`Tool` 数据类

这代表语言模型可以准备调用的工具。

它具有以下属性

  • name
  • description
  • parameters：描述预期参数的 JSON schema
  • function：工具被调用时执行的可调用对象

对 `name` 和 `description` 等文本属性的准确定义对于语言模型正确准备调用非常重要。

`Tool` 暴露了一个 `tool_spec` 属性，返回用于语言模型的工具规范。

它还有一个 `invoke` 方法，可以使用提供的参数执行底层函数。

让我们看一个例子。

from haystack.tools import Tool

def add(a: int, b: int) -> int:
    return a + b


parameters = {
    "type": "object",
    "properties": {
        "a": {"type": "integer"},
        "b": {"type": "integer"}
    },
    "required": ["a", "b"]
}

add_tool = Tool(name="addition_tool",
            description="This tool adds two numbers",
            parameters=parameters,
            function=add)

print(add_tool.tool_spec)

print(add_tool.invoke(a=15, b=10))

{
    'name': 'addition_tool',
    'description': 'This tool adds two numbers',
    'parameters': {
        'type': 'object',
        'properties': {'a': {'type': 'integer'}, 'b': {'type': 'integer'}},
        'required': ['a', 'b']
    }
}

25

在 `OpenAIChatGenerator` 中支持工具

`OpenAIChatGenerator` 现在支持工具。您可以在初始化时或通过 `run` 方法传递工具，它将在适当的时候使用它们来准备工具调用。

这里有一些例子。

from haystack.components.generators.chat import OpenAIChatGenerator

# Initialize the Chat Generator with the addition tool
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", tools=[add_tool])

res=chat_generator.run([ChatMessage.from_user("10 + 238")])
print(res)

res=chat_generator.run([ChatMessage.from_user("What is the habitat of a lion?")])
print(res)

{
    'replies': [
        ChatMessage(
            _role=<ChatRole.ASSISTANT: 'assistant'>,
            _content=[
                ToolCall(
                    tool_name='addition_tool',
                    arguments={'a': 10, 'b': 238},
                    id='call_rbYtbCdW0UbWMfy2x0sgF1Ap'
                )
            ],
            _meta={
                'model': 'gpt-4o-mini-2024-07-18',
                'index': 0,
                'finish_reason': 'tool_calls',
                'usage': {'completion_tokens': 19, 'prompt_tokens': 51, 'total_tokens': 70}
            }
        )
    ]
}

{
    'replies': [
        ChatMessage(
            _role=<ChatRole.ASSISTANT: 'assistant'>,
            _content=[
                TextContent(
                    text='Lions primarily inhabit grasslands, savannas, and open woodlands. They are generally 
found in sub-Saharan Africa, where they prefer areas that provide ample prey and suitable shelter. Historically, 
lions were also present in parts of Europe and Asia, but their range has significantly diminished, and they are now
largely confined to Africa. The presence of water sources and accessible prey are critical factors in their habitat
selection.'
                )
            ],
            _meta={
                'model': 'gpt-4o-mini-2024-07-18',
                'index': 0,
                'finish_reason': 'stop',
                'usage': {'completion_tokens': 84, 'prompt_tokens': 55, 'total_tokens': 139}
            }
        )
    ]
}

# Initialize the Chat Generator without tools
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini")

res=chat_generator.run([ChatMessage.from_user("10 + 238")])
print(res)

# pass tools in the run method: in this case, the model will prepare a tool call
res_w_tool_call=chat_generator.run([ChatMessage.from_user("10 + 238")], tools=[add_tool])
print(res_w_tool_call)

{
    'replies': [
        ChatMessage(
            _role=<ChatRole.ASSISTANT: 'assistant'>,
            _content=[TextContent(text='10 + 238 = 248.')],
            _meta={
                'model': 'gpt-4o-mini-2024-07-18',
                'index': 0,
                'finish_reason': 'stop',
                'usage': {'completion_tokens': 8, 'prompt_tokens': 11, 'total_tokens': 19}
            }
        )
    ]
}

{
    'replies': [
        ChatMessage(
            _role=<ChatRole.ASSISTANT: 'assistant'>,
            _content=[
                ToolCall(
                    tool_name='addition_tool',
                    arguments={'a': 10, 'b': 238},
                    id='call_GaVZnGszepASznKpYVe23mHf'
                )
            ],
            _meta={
                'model': 'gpt-4o-mini-2024-07-18',
                'index': 0,
                'finish_reason': 'tool_calls',
                'usage': {'completion_tokens': 19, 'prompt_tokens': 51, 'total_tokens': 70}
            }
        )
    ]
}

`ToolInvoker` 组件

此组件负责执行语言模型准备的工具调用。它期望一个消息列表（可能包括工具调用），并返回一个工具消息列表，其中包含工具调用的结果。

from haystack.components.tools import ToolInvoker

# Initialize the Tool Invoker with the weather tool
tool_invoker = ToolInvoker(tools=[add_tool])

# let's pass the messages from the previous example
print(tool_invoker.run(res_w_tool_call["replies"]))

{
    'tool_messages': [
        ChatMessage(
            _role=<ChatRole.TOOL: 'tool'>,
            _content=[
                ToolCallResult(
                    result='248',
                    origin=ToolCall(
                        tool_name='addition_tool',
                        arguments={'a': 10, 'b': 238},
                        id='call_GaVZnGszepASznKpYVe23mHf'
                    ),
                    error=False
                )
            ],
            _meta={}
        )
    ]
}

`ToolInvoker` 具有 2 个额外的 init 参数

  • `raise_on_failure`：如果为 True，则组件在发生错误时（找不到工具、工具调用错误、工具结果转换错误）会引发异常。否则，它会返回一个 `ChatMessage` 对象，其中 `error=True`，并在 `result` 中包含错误的描述。
  • `convert_result_to_json_string`：如果为 True，则工具调用结果将使用 `json.dumps` 转换为字符串。如果为 False，则使用 `str`（默认）转换结果。

让我们看看 `raise_on_failure` 如何工作。

tool_call = ToolCall(tool_name="division_tool", arguments={"c": 1, "d": 2})

tool_invoker = ToolInvoker(tools=[add_tool], raise_on_failure=True)

tool_invoker.run([ChatMessage.from_assistant(tool_calls=[tool_call])])

---------------------------------------------------------------------------

ToolNotFoundException                     Traceback (most recent call last)

<ipython-input-24-f1d4067e6631> in <cell line: 5>()
      3 tool_invoker = ToolInvoker(tools=[add_tool], raise_on_failure=True)
      4 
----> 5 tool_invoker.run([ChatMessage.from_assistant(tool_calls=[tool_call])])


/usr/local/lib/python3.10/dist-packages/haystack_experimental/components/tools/tool_invoker.py in run(self, messages)
    194                     msg = _TOOL_NOT_FOUND.format(tool_name=tool_name, available_tools=self._tools_with_names.keys())
    195                     if self.raise_on_failure:
--> 196                         raise ToolNotFoundException(msg)
    197                     tool_messages.append(ChatMessage.from_tool(tool_result=msg, origin=tool_call, error=True))
    198                     continue


ToolNotFoundException: Tool division_tool not found in the list of tools. Available tools are: dict_keys(['addition_tool']).

tool_invoker = ToolInvoker(tools=[add_tool], raise_on_failure=False)

print(tool_invoker.run([ChatMessage.from_assistant(tool_calls=[tool_call])]))

{
    'tool_messages': [
        ChatMessage(
            _role=<ChatRole.TOOL: 'tool'>,
            _content=[
                ToolCallResult(
                    result="Tool division_tool not found in the list of tools. Available tools are: 
dict_keys(['addition_tool']).",
                    origin=ToolCall(tool_name='division_tool', arguments={'c': 1, 'd': 2}, id=None),
                    error=True
                )
            ],
            _meta={}
        )
    ]
}

端到端示例

在本节中，我们将把迄今为止涵盖的所有内容整合到一些实际的、端到端的示例中。

一个简单的用例

我们将从创建一个基本的助手开始，该助手可以通过使用天气工具或依赖自身知识来回答用户问题。

# Define a dummy weather tool

import random

def dummy_weather(location: str):
    return {"temp": f"{random.randint(-10, 40)} °C",
            "humidity": f"{random.randint(0, 100)}%"}


weather_tool = Tool(
    name="weather",
    description="A tool to get the weather",
    function=dummy_weather,
    parameters={
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
    },
)

from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.tools import ToolInvoker

# Initialize the Chat Generator with the weather tool
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", tools=[weather_tool])

# Initialize the Tool Invoker with the weather tool
tool_invoker = ToolInvoker(tools=[weather_tool])

user_message = ChatMessage.from_user("What is the weather in Berlin?")

replies = chat_generator.run(messages=[user_message])["replies"]
print(f"assistant messages: {replies}")

# If the assistant message contains a tool call, run the tool invoker
if replies[0].tool_calls:
    tool_messages = tool_invoker.run(messages=replies)["tool_messages"]
    print(f"tool messages: {tool_messages}")

assistant messages: [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[ToolCall(tool_name='weather', 
arguments={'location': 'Berlin'}, id='call_YEvCEAmlvc42JGXV84NU8wtV')], _meta={'model': 'gpt-4o-mini-2024-07-18', 
'index': 0, 'finish_reason': 'tool_calls', 'usage': {'completion_tokens': 13, 'prompt_tokens': 50, 'total_tokens': 
63}})]

tool messages: [ChatMessage(_role=<ChatRole.TOOL: 'tool'>, _content=[ToolCallResult(result="{'temp': '22 °C', 
'humidity': '35%'}", origin=ToolCall(tool_name='weather', arguments={'location': 'Berlin'}, 
id='call_YEvCEAmlvc42JGXV84NU8wtV'), error=False)], _meta={})]

助手能够正确识别何时需要使用工具来回答问题并调用适当的工具。

user_message = ChatMessage.from_user("What is the capital of Australia?")

replies = chat_generator.run(messages=[user_message])["replies"]
print(f"assistant messages: {replies}")

# If the assistant message contains a tool call, run the tool invoker
if replies[0].tool_calls:
    tool_messages = tool_invoker.run(messages=replies)["tool_messages"]
    print(f"tool messages: {tool_messages}")

assistant messages: [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text='The capital 
of Australia is Canberra.')], _meta={'model': 'gpt-4o-mini-2024-07-18', 'index': 0, 'finish_reason': 'stop', 
'usage': {'completion_tokens': 8, 'prompt_tokens': 50, 'total_tokens': 58}})]

这次，助手使用其内部知识来回答问题。

使用 Pipeline

为了使用 Pipeline 实现类似的功能，我们将引入一个 Conditional Router，它根据回复是否包含工具调用来指导流程。

from haystack import Pipeline
from haystack.components.routers import ConditionalRouter
from typing import List

routes = [
    {
        "condition": "{{replies[0].tool_calls | length > 0}}",
        "output": "{{replies}}",
        "output_name": "there_are_tool_calls",
        "output_type": List[ChatMessage],
    },
    {
        "condition": "{{replies[0].tool_calls | length == 0}}",
        "output": "{{replies}}",
        "output_name": "final_replies",
        "output_type": List[ChatMessage],
    },
]

tools_pipe = Pipeline()
tools_pipe.add_component("generator", OpenAIChatGenerator(model="gpt-4o-mini", tools=[weather_tool]))
tools_pipe.add_component("router", ConditionalRouter(routes, unsafe=True))
tools_pipe.add_component("tool_invoker", ToolInvoker(tools=[weather_tool]))


tools_pipe.connect("generator.replies", "router")
tools_pipe.connect("router.there_are_tool_calls", "tool_invoker")

tools_pipe.show()

/usr/local/lib/python3.10/dist-packages/haystack/components/routers/conditional_router.py:141: UserWarning: Unsafe mode is enabled. This allows execution of arbitrary code in the Jinja template. Use this only if you trust the source of the template.
  warn(msg)

res=tools_pipe.run({"messages":[ChatMessage.from_user("What is the capital of Australia?")]})
print(res)

print("-"*50)

res=tools_pipe.run({"messages":[ChatMessage.from_user("What is the weather in Berlin?")]})
print(res)

{
    'router': {
        'final_replies': [
            ChatMessage(
                _role=<ChatRole.ASSISTANT: 'assistant'>,
                _content=[TextContent(text='The capital of Australia is Canberra.')],
                _meta={
                    'model': 'gpt-4o-mini-2024-07-18',
                    'index': 0,
                    'finish_reason': 'stop',
                    'usage': {'completion_tokens': 8, 'prompt_tokens': 50, 'total_tokens': 58}
                }
            )
        ]
    }
}

--------------------------------------------------

{
    'tool_invoker': {
        'tool_messages': [
            ChatMessage(
                _role=<ChatRole.TOOL: 'tool'>,
                _content=[
                    ToolCallResult(
                        result="{'temp': '7 °C', 'humidity': '46%'}",
                        origin=ToolCall(
                            tool_name='weather',
                            arguments={'location': 'Berlin'},
                            id='call_q3KA4O62IBLfOljuUKVPzdxy'
                        ),
                        error=False
                    )
                ],
                _meta={}
            )
        ]
    }
}

在此示例中，在工具调用的情况下，我们得到原始的工具调用结果，包装在来自工具角色的 `ChatMessage` 中。在下一个示例中，我们将看到如何进一步处理此结果。

使用聊天生成器处理工具结果

根据我们的用例和涉及的工具，我们可能希望在将工具的输出呈现给用户之前对其进行进一步处理。这可以使响应更用户友好。

在下一个示例中，我们将把工具的响应传递回聊天生成器进行最终处理。

chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", tools=[weather_tool])
tool_invoker = ToolInvoker(tools=[weather_tool])

user_message = ChatMessage.from_user("What is the weather in Berlin?")

replies = chat_generator.run(messages=[user_message])["replies"]
print(f"assistant messages: {replies}")

if replies[0].tool_calls:

    tool_messages = tool_invoker.run(messages=replies)["tool_messages"]
    print(f"tool messages: {tool_messages}")

    # we pass all the messages to the Chat Generator
    messages = [user_message] + replies + tool_messages
    final_replies = chat_generator.run(messages=messages)["replies"]
    print(f"final assistant messages: {final_replies}")

assistant messages: [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[ToolCall(tool_name='weather', 
arguments={'location': 'Berlin'}, id='call_jHX0RCDHRKX7h8V9RrNs6apy')], _meta={'model': 'gpt-4o-mini-2024-07-18', 
'index': 0, 'finish_reason': 'tool_calls', 'usage': {'completion_tokens': 13, 'prompt_tokens': 50, 'total_tokens': 
63}})]

tool messages: [ChatMessage(_role=<ChatRole.TOOL: 'tool'>, _content=[ToolCallResult(result="{'temp': '2 °C', 
'humidity': '15%'}", origin=ToolCall(tool_name='weather', arguments={'location': 'Berlin'}, 
id='call_jHX0RCDHRKX7h8V9RrNs6apy'), error=False)], _meta={})]

final assistant messages: [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text='The 
current weather in Berlin is 2 °C with a humidity level of 15%.')], _meta={'model': 'gpt-4o-mini-2024-07-18', 
'index': 0, 'finish_reason': 'stop', 'usage': {'completion_tokens': 19, 'prompt_tokens': 85, 'total_tokens': 
104}})]

助手会改进工具的输出，以创建更易于人类阅读的响应。

使用 Pipeline

Pipeline 与之前的 Pipeline 类似。

我们引入了一个自定义组件 `MessageCollector` 来临时存储消息。

from typing import Any, Dict, List
from haystack import component
from haystack.core.component.types import Variadic

@component()
class MessageCollector:
    def __init__(self):
        self._messages = []

    @component.output_types(messages=List[ChatMessage])
    def run(self, messages: Variadic[List[ChatMessage]]) -> Dict[str, Any]:

        self._messages.extend([msg for inner in messages for msg in inner])
        return {"messages": self._messages}

    def clear(self):
        self._messages = []

message_collector = MessageCollector()

routes = [
    {
        "condition": "{{replies[0].tool_calls | length > 0}}",
        "output": "{{replies}}",
        "output_name": "there_are_tool_calls",
        "output_type": List[ChatMessage],
    },
    {
        "condition": "{{replies[0].tool_calls | length == 0}}",
        "output": "{{replies}}",
        "output_name": "final_replies",
        "output_type": List[ChatMessage],
    },
]

tool_agent = Pipeline()
tool_agent.add_component("message_collector", message_collector)
tool_agent.add_component("generator", OpenAIChatGenerator(model="gpt-4o-mini", tools=[weather_tool]))
tool_agent.add_component("router", ConditionalRouter(routes, unsafe=True))
tool_agent.add_component("tool_invoker", ToolInvoker(tools=[weather_tool]))


tool_agent.connect("message_collector", "generator.messages")
tool_agent.connect("generator.replies", "router")
tool_agent.connect("router.there_are_tool_calls", "tool_invoker")
tool_agent.connect("router.there_are_tool_calls", "message_collector")
tool_agent.connect("tool_invoker.tool_messages", "message_collector")

tool_agent.show()

message_collector.clear()
res=tool_agent.run({"messages":[ChatMessage.from_user("What is the capital of Australia?")]})
print(res)

print("-"*50)

message_collector.clear()
res=tool_agent.run({"messages":[ChatMessage.from_user("What is the weather in Berlin?")]})
print(res)

print("-"*50)

message_collector.clear()
res=tool_agent.run({"messages":[ChatMessage.from_user("What is the weather in Rome and Bangkok?")]})
print(res)

{
    'router': {
        'final_replies': [
            ChatMessage(
                _role=<ChatRole.ASSISTANT: 'assistant'>,
                _content=[TextContent(text='The capital of Australia is Canberra.')],
                _meta={
                    'model': 'gpt-4o-mini-2024-07-18',
                    'index': 0,
                    'finish_reason': 'stop',
                    'usage': {'completion_tokens': 8, 'prompt_tokens': 50, 'total_tokens': 58}
                }
            )
        ]
    }
}

--------------------------------------------------

{
    'router': {
        'final_replies': [
            ChatMessage(
                _role=<ChatRole.ASSISTANT: 'assistant'>,
                _content=[TextContent(text='The current weather in Berlin is 9 °C with a humidity of 40%.')],
                _meta={
                    'model': 'gpt-4o-mini-2024-07-18',
                    'index': 0,
                    'finish_reason': 'stop',
                    'usage': {'completion_tokens': 18, 'prompt_tokens': 85, 'total_tokens': 103}
                }
            )
        ]
    }
}

--------------------------------------------------

{
    'router': {
        'final_replies': [
            ChatMessage(
                _role=<ChatRole.ASSISTANT: 'assistant'>,
                _content=[
                    TextContent(
                        text='The current weather is as follows:\n\n- **Rome**: Temperature is 31 °C with 80% 
humidity.\n- **Bangkok**: Temperature is 20 °C with 23% humidity.'
                    )
                ],
                _meta={
                    'model': 'gpt-4o-mini-2024-07-18',
                    'index': 0,
                    'finish_reason': 'stop',
                    'usage': {'completion_tokens': 43, 'prompt_tokens': 147, 'total_tokens': 190}
                }
            )
        ]
    }
}

有关涉及多个工具和人工干预的更复杂示例，请参阅本教程：使用函数调用构建聊天代理。