函数调用有什么用？

函数调用使开发人员能够描述函数（也称为工具，可以将其视为模型要采取的操作，例如执行计算或下订单），并且让模型智能地选择输出包含调用这些函数的参数的 JSON 对象。简而言之，它允许：

• 自主决策：模型可以智能地选择工具来回答问题。

• 可靠的解析：响应采用 JSON 格式，而不是更典型的类似对话的响应。乍一看似乎没什么，但这就是 LLM 能够连接到外部系统的原因，比如通过具有结构化输入的 API。

它开启了无数的可能性：

• 自主人工智能助手：除了提供查询答案外，机器人还可以与内部系统交互，完成客户订单和退货等任务

• 个人研究助理：假设您正在计划旅行，助理可以搜索网络、抓取内容、比较选项并在 Excel 中汇总结果。

• 物联网语音命令：模型可以控制设备或根据检测到的意图建议操作，例如调整空调温度。

函数调用的结构

借用Gemini 的函数调用文档，函数调用具有以下结构，在 OpenAI 中的工作方式相同

1. 用户向应用程序发出提示
2. 应用程序传递用户提供的提示和函数声明，函数声明是模型可以使用的工具的描述
3. 根据函数声明，模型建议使用的工具以及相关的请求参数。请注意，模型仅输出建议的工具和参数，而不实际调用函数
4. & 5. 应用程序根据响应调用相关API

架构

在深入研究代码之前，先简单介绍一下演示应用程序的架构

解决方案

在这里，我们为参观酒店的游客建立了一个助手。助手可以访问以下工具，从而允许助手访问外部应用程序。

• get_items, purchase_item: 通过API连接数据库中的商品目录，分别用于检索商品列表和购买

• rag_pipeline_func：使用检索增强生成（RAG）连接到文档存储，从非结构化文本（例如酒店手册）中获取信息

技术栈

• 嵌入模型：全MiniLM-L6-v2

• 矢量数据库：Haystack 的 InMemoryDocumentStore

• LLM：GPT-4 Turbo 通过 OpenRouter 访问。借助 OpenRouter，访问不同的 LLM API。Gemini 表示，如果其他 LLM 支持函数调用，只需稍微更改代码即可将该流程改编为使用其他 LLM

• LLM 框架：Haystack的易用性、丰富的文档和管道构建的透明度。本教程实际上是他们针对同一主题的精彩教程的扩展

Agent应用

准备

前往Github克隆我们的代码。以下内容可以在：

function_calling_demoNotebook中找到。

初始化

我们首先连接到 OpenRouter。或者，如果您有 OpenAI API 密钥，则使用原始版本OpenAIChatGenerator而不覆盖api_base_url也可以。

import os
from dotenv import load_dotenv
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.utils import Secret
from haystack.dataclasses import ChatMessage
from haystack.components.generators.utils import print_streaming_chunk

# Set your API key as environment variable before executing this
load_dotenv()
OPENROUTER_API_KEY = os.environ.get('OPENROUTER_API_KEY')

chat_generator = OpenAIChatGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
  api_base_url="https://openrouter.ai/api/v1",
  model="openai/gpt-4-turbo-preview",
        streaming_callback=print_streaming_chunk)

chat_generator.run(messages=[ChatMessage.from_user("Return this text: 'test'")])

---------- The response should look like this ----------{'replies': [ChatMessage(content="'test'", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'stop', 'usage': {}})]}

步骤1：建立数据存储

在这里，我们在应用程序和两个数据源之间建立连接：非结构化文本的文档存储和通过 API 的应用程序数据库

使用管道索引文档

documents我们为模型提供示例文本以执行检索增强生成（RAG）。文本被转换为嵌入并存储在内存文档存储中。

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

# Sample documents
documents = [
    Document(content="Coffee shop opens at 9am and closes at 5pm."),
    Document(content="Gym room opens at 6am and closes at 10pm.")
]

# Create the document store
document_store = InMemoryDocumentStore()

# Create a pipeline to turn the texts into embeddings and store them in the document store
indexing_pipeline = Pipeline()
indexing_pipeline.add_component(
    "doc_embedder", SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
)
indexing_pipeline.add_component("doc_writer", DocumentWriter(document_store=document_store))

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

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

它应该输出这个，对应于documents我们创建的示例：

{ 'doc_writer' :  { 'documents_writing' :  2 } }

启动 API 服务器

下面创建一个用Flask制作的API服务器db_api.py来连接SQLite。请通过python db_api.py在终端中运行来启动它

另请注意，一些初始数据已添加到db_api.py

第 2 步：定义函数

在这里，我们为模型准备实际的函数，以便在函数调用之后调用（步骤 4-5，如函数调用的结构中所述）

RAG函数

即rag_pipeline_func.这是为了让模型通过搜索文档存储中存储的文本来提供答案。我们首先将 RAG 检索定义为 Haystack 管道：

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

template = """
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", PromptBuilder(template=template))
# Note to llm: We are using OpenAIGenerator, not the OpenAIChatGenerator, because the latter only accepts List[str] as input and cannot accept prompt_builder's str output
rag_pipe.add_component("llm", OpenAIGenerator(api_key=Secret.from_env_var("OPENROUTER_API_KEY"),
  api_base_url="https://openrouter.ai/api/v1",
  model="openai/gpt-4-turbo-preview"))

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

测试功能是否有效：

query = “When does the coffee shop open?”
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})

这应该产生以下输出。请注意replies，模型给出的内容来自我们之前提供的示例文档

{'llm':{'replies':['The coffee shop opens at 9am.'],
  'meta':[{'model': 'openai/gpt-4-turbo-preview',
    'index':0,
    'finish_reason': 'stop',
    'usage':{'completion_tokens':9,
     'prompt_tokens':60,
     'total_tokens':69,
     'total_cost':0.00087}}]}}

然后我们可以将 变成rag_pipe一个函数，它提供了replies唯一的而不添加其他细节

defrag_pipeline_func ( query: str ): 
result = rag_pipe.run({ "embedder" : { "text" : query}, "prompt_builder" : { "question" : query}}) 

return { "reply" : result[ "llm " ][ "replies" ][ 0 ]}

API调用

# Flask's default local URL, change it if necessary
db_base_url = 'http://127.0.0.1:5000'

# Use requests to get the data from the database
import requests
import json

# get_categories is supplied as part of the prompt, it is not used as a tool
defget_categories():
    response = requests.get(f'{db_base_url}/category')
    data = response.json()
    return data

defget_items(ids=None,categories=None):
    params = {
        'id': ids,
        'category': categories,
    }
    response = requests.get(f'{db_base_url}/item', params=params)
    data = response.json()
    return data

defpurchase_item(id,quantity):

    headers = {
    'Content-type':'application/json', 
    'Accept':'application/json'
    }

    data = {
        'id': id,
        'quantity': quantity,
    }
    response = requests.post(f'{db_base_url}/item/purchase', json=data, headers=headers)
    return response.json()

定义工具列表

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_items",
            "description": "Get a list of items from the database",
            "parameters": {
                "type": "object",
                "properties": {
                    "ids": {
                        "type": "string",
                        "description": "Comma separated list of item ids to fetch",
                    },
                    "categories": {
                        "type": "string",
                        "description": "Comma separated list of item categories to fetch",
                    },
                },
                "required": [],
            },
        }
    },
    {
        "type": "function",
        "function": {
            "name": "purchase_item",
            "description": "Purchase a particular item",
            "parameters": {
                "type": "object",
                "properties": {
                    "id": {
                        "type": "string",
                        "description": "The given product ID, product name is not accepted here. Please obtain the product ID from the database first.",
                    },
                    "quantity": {
                        "type": "integer",
                        "description": "Number of items to purchase",
                    },
                },
                "required": [],
            },
        }
    },
    {
        "type": "function",
        "function": {
            "name": "rag_pipeline_func",
            "description": "Get information from hotel brochure",
            "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"],
            },
        },
    }
]

第 3 步：将所有内容放在一起

1. 向模型提供初始提示，为其提供一些背景信息

2. 提供用户生成的示例消息

3. 最重要的是，我们将工具列表传递给聊天生成器tools

# 1. Initial prompt
context = f"""You are an assistant to tourists visiting a hotel.
You have access to a database of items (which includes {get_categories()}) that tourists can buy, you also have access to the hotel's brochure.
If the tourist's question cannot be answered from the database, you can refer to the brochure.
If the tourist's question cannot be answered from the brochure, you can ask the tourist to ask the hotel staff.
"""
messages = [
    ChatMessage.from_system(context),
    # 2. Sample message from user
    ChatMessage.from_user("Can I buy a coffee?"),
    ]

# 3. Passing the tools list and invoke the chat generator
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
response

---------- Response ----------
{'replies': [ChatMessage(content='[{"index": 0, "id": "call_AkTWoiJzx5uJSgKW0WAI1yBB", "function": {"arguments": "{\\"categories\\":\\"Food and beverages\\"}", "name": "get_items"}, "type": "function"}]', role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'tool_calls', 'usage': {}})]}

现在让我们检查响应。请注意函数调用如何返回模型选择的函数以及调用所选函数的参数。

function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)

---------- Response ----------
Function Name: get_items
Function Arguments: {‘categories’: ‘Food and beverages’}

当提出另一个问题时，模型将使用另一个更相关的工具。

# Another question
messages.append(ChatMessage.from_user("Where's the coffee shop?"))

# Invoke the chat generator, and passing the tools list
response = chat_generator.run(messages=messages, generation_kwargs= {"tools": tools})
function_call = json.loads(response["replies"][0].content)[0]
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
print("Function Name:", function_name)
print("Function Arguments:", function_args)

---------- Response ----------
Function Name: rag_pipeline_func
Function Arguments: {'query': "Where's the coffee shop?"}

再次注意，这里没有调用实际的函数，这就是我们接下来要做的。

调用函数

然后我们可以将参数输入到所选函数中：

## Find the correspoding function and call it with the given arguments
available_functions = {"get_items": get_items, "purchase_item": purchase_item,"rag_pipeline_func": rag_pipeline_func}
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
print("Function Response:", function_response)

---------- Response ----------
Function Response: {'reply': 'The provided context does not specify a physical location for the coffee shop, only its operating hours. Therefore, I cannot determine where the coffee shop is located based on the given information.'}

然后，可以将来自的响应rag_pipeline_func作为上下文传递到聊天，方法是将其附加到 下messages，以便模型提供最终答案。

messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name)) 
response = chat_generator.run(messages=messages) 
response_msg = response[ "replies" ][ 0 ] 

print (response_msg.content)

---------- Response ----------
For the location of the coffee shop within the hotel, I recommend asking the hotel staff directly. They will be able to guide you to it accurately.
“对于酒店内咖啡店的位置，我建议直接询问酒店工作人员。他们将能够准确地指导您。”

我们现在已经完成了聊天周期！

第四步：变成互动聊天

上面的代码展示了如何完成函数调用，但我们想更进一步，将其变成交互式聊天。在这里，我展示了两种方法来实现这一点，从input()将对话打印到笔记本本身的更原始方法，到通过Streamlit渲染它以为其提供类似 ChatGPT 的 UI。

input() loop

代码是从Haystack的教程中复制过来的，这可以让我们快速测试模型。注意：创建此应用程序是为了演示函数调用的思想，并不意味着完全健壮，例如同时支持多个项目的顺序、没有幻觉等。

import json
from haystack.dataclasses import ChatMessage, ChatRole

response = None
messages = [
    ChatMessage.from_system(context)
]

whileTrue:
    # if OpenAI response is a tool call
    if response and response["replies"][0].meta["finish_reason"] == "tool_calls":
        function_calls = json.loads(response["replies"][0].content)

        for function_call in function_calls:
            ## Parse function calling information
            function_name = function_call["function"]["name"]
            function_args = json.loads(function_call["function"]["arguments"])

            ## Find the correspoding function and call it with the given arguments
            function_to_call = available_functions[function_name]
            function_response = function_to_call(**function_args)

            ## Append function response to the messages list using `ChatMessage.from_function`
            messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))

    # Regular Conversation
    else:
        # Append assistant messages to the messages list
        ifnot messages[-1].is_from(ChatRole.SYSTEM):
            messages.append(response["replies"][0])

        user_input = input("ENTER YOUR MESSAGE ? INFO: Type 'exit' or 'quit' to stop\n")
        if user_input.lower() == "exit"or user_input.lower() == "quit":
            break
        else:
            messages.append(ChatMessage.from_user(user_input))

    response = chat_generator.run(messages=messages, generation_kwargs={"tools": tools})

虽然它有效，但我们可能想要一些看起来更好的东西。

Streamlit界面

Streamlit 将数据脚本转换为可共享的 Web 应用程序，这为我们的应用程序提供了简洁的 UI。上面显示的代码已改编成我们的存储库streamlit文件夹下的 Streamlit 应用程序。

可以通过以下方式运行它：

1. 如果您还没有这样做，请使用以下命令启动 API 服务器python db_api.py

2. 将 OPENROUTER_API_KEY 设置为环境变量，例如export OPENROUTER_API_KEY = ‘@REPLACE WITH YOUR API KEY’假设您在 Linux 上/使用 git bash 执行

3. 在命令行中使用以下命令导航到streamlit文件夹cd streamlit

4. 使用 运行 Streamlit streamlit run app.py。运行应用程序的浏览器中应该会自动创建一个新选项卡