教程：使用函数调用构建聊天代理

在 Colab 中打开下载

_{最后更新：2025 年 8 月 25 日}

级别：高级
完成时间：20 分钟
使用的组件：InMemoryDocumentStore，SentenceTransformersDocumentEmbedder，SentenceTransformersTextEmbedder，InMemoryEmbeddingRetriever，ChatPromptBuilder，OpenAIChatGenerator，ToolInvoker
先决条件：您必须拥有一个OpenAI API 密钥，并熟悉创建管道。
目标：完成本教程后，您将学会如何使用 OpenAI 的函数调用功能，构建演示代理行为的聊天应用程序。

概述

📚 有用的资源

OpenAI 的函数调用将大型语言模型连接到外部工具。通过向 OpenAI API 调用提供包含函数及其规范的 tools 列表，您可以轻松构建可以通过调用外部 API 来回答问题或从文本中提取结构化信息的聊天助手。

在本教程中，您将学习如何将 Haystack 管道转换为函数调用工具，以及如何通过 OpenAIChatGenerator 使用 OpenAI 的 Chat Completion API 实现应用程序，以获得代理行为。

设置开发环境

使用 pip 安装 Haystack 和 sentence-transformers

%%bash

pip install haystack-ai
pip install "sentence-transformers>=4.1.0"

将您的 OpenAI API 密钥保存为环境变量

import os
from getpass import getpass

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

了解 OpenAIChatGenerator

OpenAIChatGenerator 是一个支持 OpenAI 的函数调用功能的组件，通过 Chat Completion API 实现。与 OpenAIGenerator 不同，与 OpenAIChatGenerator 通信的方式是通过 ChatMessage 列表。在Generators vs Chat Generators 中了解更多它们之间的区别。

要开始使用 OpenAIChatGenerator，请创建一个具有“SYSTEM”角色的 ChatMessage 对象，使用 ChatMessage.from_system()，并创建另一个具有“USER”角色的 ChatMessage 对象，使用 ChatMessage.from_user()。然后，将此消息列表传递给 OpenAIChatGenerator 并运行。

from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator

messages = [
    ChatMessage.from_system("Always respond in German even if some input data is in other languages."),
    ChatMessage.from_user("What's Natural Language Processing? Be brief."),
]

chat_generator = OpenAIChatGenerator(model="gpt-4o-mini")
chat_generator.run(messages=messages)

基本流式传输

OpenAIChatGenerator 支持流式传输，提供一个 streaming_callback 函数，然后再次运行 chat_generator 以查看差异。

from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk

chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", streaming_callback=print_streaming_chunk)
response = chat_generator.run(messages=messages)

从 Haystack Pipeline 创建函数调用工具

要使用 OpenAI 的函数调用，您需要将 tools 引入您的 OpenAIChatGenerator。

在本例中，您将使用 Haystack RAG 管道作为工具之一。因此，您需要将文档索引到文档存储中，然后在文档存储之上构建一个 RAG 管道。

使用管道索引文档

创建一个管道，将小型示例数据集及其嵌入存储到 InMemoryDocumentStore 中。您将使用 SentenceTransformersDocumentEmbedder 为您的文档生成嵌入，并使用 DocumentWriter 将它们写入文档存储。

将这些组件添加到管道后，连接它们并运行管道。

如果您想了解在将文件索引到文档存储之前进行预处理，请遵循预处理不同文件类型教程。

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder

documents = [
    Document(content="My name is Jean and I live in Paris."),
    Document(content="My name is Mark and I live in Berlin."),
    Document(content="My name is Giorgio and I live in Rome."),
    Document(content="My name is Marta and I live in Madrid."),
    Document(content="My name is Harry and I live in London."),
]

document_store = InMemoryDocumentStore()

indexing_pipeline = Pipeline()
indexing_pipeline.add_component(
    instance=SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"), name="doc_embedder"
)
indexing_pipeline.add_component(instance=DocumentWriter(document_store=document_store), name="doc_writer")

indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")

indexing_pipeline.run({"doc_embedder": {"documents": documents}})

构建 RAG 管道

使用 SentenceTransformersTextEmbedder、InMemoryEmbeddingRetriever、ChatPromptBuilder 和 OpenAIChatGenerator 构建一个基本的检索增强生成管道。

有关使用 Haystack 创建 RAG 管道的分步指南，请遵循创建您的第一个带检索增强的 QA 管道教程。

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator

template = [
    ChatMessage.from_system(
        """
Answer the questions based on the given context.

Context:
{% for document in documents %}
    {{ document.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
    )
]
rag_pipe = Pipeline()
rag_pipe.add_component("embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
rag_pipe.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store))
rag_pipe.add_component("prompt_builder", ChatPromptBuilder(template=template))
rag_pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))

rag_pipe.connect("embedder.embedding", "retriever.query_embedding")
rag_pipe.connect("retriever", "prompt_builder.documents")
rag_pipe.connect("prompt_builder.prompt", "llm.messages")

运行管道

在将其用作函数调用工具之前，用查询测试此管道，看看它是否按预期工作。

query = "Where does Mark live?"
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})

将 Haystack 管道转换为工具

将 rag_pipe.run 调用包装在一个名为 rag_pipeline_func 的函数中。此函数应接受 query 作为输入，并返回您之前构建的 RAG 管道中 LLM 生成的响应。接下来，按照工具初始化中的步骤初始化一个 Tool；定义 parameters，并为工具设置 name 和 description。

Haystack 中的 Tool 抽象层支持与多个 ChatGenerators 进行无缝的函数调用/工具调用。从 Haystack 2.9 开始，这包括对 AnthropicChatGenerator、OllamaChatGenerator 和 OpenAIChatGenerator 的支持，未来还会有更多集成。

from haystack.tools import Tool


def rag_pipeline_func(query: str):
    result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
    return {"reply": result["llm"]["replies"][0].text}


parameters = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "The query to use in the search. Infer this from the user's message. It should be a question or a statement",
        }
    },
    "required": ["query"],
}

rag_pipeline_tool = Tool(
    name="rag_pipeline_tool",
    description="Get information about where people live",
    parameters=parameters,
    function=rag_pipeline_func,
)

从函数创建工具

除了 rag_pipeline_tool 之外，创建一个名为 get_weather_tool 的新工具，用于获取城市的天气信息。

首先，创建一个函数来模拟对外部天气服务的 API 调用。与上一个工具传递 JSON 作为参数不同，使用create_tool_from_function。此函数需要使用 Annotated 类型来描述工具参数，以获取额外详细信息。但是，基于此信息，create_tool_from_function 可以自动推断参数并生成 JSON 模式，因此您无需单独定义 parameters。

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

WEATHER_INFO = {
    "Berlin": {"weather": "mostly sunny", "temperature": 7, "unit": "celsius"},
    "Paris": {"weather": "mostly cloudy", "temperature": 8, "unit": "celsius"},
    "Rome": {"weather": "sunny", "temperature": 14, "unit": "celsius"},
    "Madrid": {"weather": "sunny", "temperature": 10, "unit": "celsius"},
    "London": {"weather": "cloudy", "temperature": 9, "unit": "celsius"},
}


def get_weather(
    city: Annotated[str, "the city for which to get the weather"] = "Berlin",
    unit: Annotated[Literal["Celsius", "Fahrenheit"], "the unit for the temperature"] = "Celsius",
):
    """A simple function to get the current weather for a location."""
    if city in WEATHER_INFO:
        return WEATHER_INFO[city]
    else:
        return {"weather": "sunny", "temperature": 21.8, "unit": "fahrenheit"}


weather_tool = create_tool_from_function(get_weather)

使用工具运行 OpenAIChatGenerator

要使用工具调用功能，您需要将工具列表作为 tools 传递给 OpenAIChatGenerator。

使用系统消息指示模型使用提供的工具，然后提供需要工具调用的查询作为用户消息

from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk

user_messages = [
    ChatMessage.from_system(
        "Use the tool that you're provided with. Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
    ),
    ChatMessage.from_user("Can you tell me where Mark lives?"),
]

chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", streaming_callback=print_streaming_chunk)
response = chat_generator.run(messages=user_messages, tools=[rag_pipeline_tool, weather_tool])

作为响应，您将收到一个 ChatMessage，其中包含关于工具名称和参数的信息，这些信息以 ToolCall 对象的形式提供。

{'replies': [
    ChatMessage(
        _role=<ChatRole.ASSISTANT: 'assistant'>,
        _content=[TextContent(text=''), ToolCall(tool_name='rag_pipeline_tool', arguments={'query': 'Where does Mark live?'}, id='call_xHEPMFrkHEKsi7tFB8FItkFC')],
        _name=None,
        meta={'model': 'gpt-4o-mini-2024-07-18', 'index': 0, 'finish_reason': 'tool_calls', 'usage': {}}
        )
    ]
}

接下来，使用 ToolInvoker 处理包含工具调用的 ChatMessage 对象，调用相应的工具，并将结果作为 ChatMessage 列表返回。

from haystack.components.tools import ToolInvoker

tool_invoker = ToolInvoker(tools=[rag_pipeline_tool, weather_tool])

if response["replies"][0].tool_calls:
    tool_result_messages = tool_invoker.run(messages=response["replies"])["tool_messages"]
    print(f"tool result messages: {tool_result_messages}")

最后一步是使用工具调用结果再次运行 OpenAIChatGenerator，并获得最终答案。

# Pass all the messages to the ChatGenerator with the correct order
messages = user_messages + response["replies"] + tool_result_messages
final_replies = chat_generator.run(messages=messages, tools=[rag_pipeline_tool, weather_tool])["replies"]

构建聊天代理

如上所述，OpenAI Chat Completions API 不会调用工具；相反，模型会生成 JSON，您可以在代码中使用它来调用工具。这就是为什么为了构建一个端到端的聊天代理，您需要为每个消息检查 OpenAI 响应是否为 tool_calls。如果是，您需要使用提供的参数调用相应的工具，并将工具响应发送回 OpenAI。否则，将用户消息和工具响应都追加到 messages 列表中，以进行常规对话。让我们构建一个可以处理所有情况的应用程序。

为了给您的应用程序构建一个漂亮的 UI，您可以使用 Gradio，它附带一个聊天界面。安装 gradio，运行下面的代码单元，并使用输入框与可以访问您上面创建的两个工具的聊天应用程序进行交互。

您可以尝试的示例查询

“*瑞典的首都是哪里？*”：一个没有工具调用的基本查询
“*你能告诉我 Giorgio 住在哪儿吗？*”：一个带有一个工具调用的基本查询
“*柏林的天气怎么样？*”，“*那里是晴天吗？*”：查看消息是否被记录并发送
“*Jean 住的地方天气怎么样？*”：强制进行两次工具调用
“*今天天气怎么样？*”：强制 OpenAI 询问更多澄清

请记住，OpenAI 模型有时会产生幻觉答案或工具，可能无法按预期工作。

%%bash

pip install -U gradio

import gradio as gr
import json

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

tool_invoker = ToolInvoker(tools=[rag_pipeline_tool, weather_tool])
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", tools=[rag_pipeline_tool, weather_tool])
response = None
messages = [
    ChatMessage.from_system(
        "Use the tool that you're provided with. Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
    )
]


def chatbot_with_tc(message, history):
    global messages
    messages.append(ChatMessage.from_user(message))
    response = chat_generator.run(messages=messages)

    while True:
        # if OpenAI response is a function call
        if response and response["replies"][0].tool_calls:
            tool_result_messages = tool_invoker.run(messages=response["replies"])["tool_messages"]
            messages = messages + response["replies"] + tool_result_messages
            response = chat_generator.run(messages=messages)

        # Regular Conversation
        else:
            messages.append(response["replies"][0])
            break
    return response["replies"][0].text


demo = gr.ChatInterface(
    fn=chatbot_with_tc,
    type="messages",
    examples=[
        "Can you tell me where Giorgio lives?",
        "What's the weather like in Madrid?",
        "Who lives in London?",
        "What's the weather like where Mark lives?",
    ],
    title="Ask me about weather or where people live!",
    theme=gr.themes.Ocean(),
)

## Uncomment the line below to launch the chat app with UI
# demo.launch()

下一步

🎉 恭喜！您已经学会了如何使用 OpenAI 函数调用和 Haystack Pipelines 构建演示代理行为的聊天应用程序。

如果您喜欢本教程，还有更多关于 Haystack 的内容可以学习。

要随时了解最新的 Haystack 发展，您可以订阅我们的时事通讯或加入 Haystack discord 社区。

感谢阅读！

创建视觉+文本 RAG 流水线