函数调用并不是什么新事物。2023年7月，OpenAI为其GPT模型引入了函数调用，这一特性现在也被竞争对手效仿。最近，谷歌的Gemini API也开始支持这项功能，而Anthropic也正在将这项功能整合到Claude中。对大型语言模型（LLMs）来说，函数调用正变得越来越重要，增强了它们的功能。学习这项技术越来越有用！
考虑到这一点，我打算写一篇全面的教程文章，涵盖超出基础知识介绍的函数调用（关于这个主题的基础教程已经有很多了）。重点将放在实际实现上，构建一个完全自主的AI代理程序，并将其与Streamlit集成，以实现类似ChatGPT的界面效果。虽然本教程中使用OpenAI作为示例，但本教程也可以轻松地适应其他支持函数调用的大型语言模型（如LLM），例如Gemini模型。
函数调用是用来干嘛的？函数调用使开发人员能够描述函数（也可以将其视为工具），并让模型智能输出调用这些函数所需参数的 JSON 对象。简单来说，它能做到：
- 开发者描述可以执行的动作或操作。
-
模型智能选择并输出相应的 JSON 对象来调用这些操作。
- 自主决策能力：模型可以智能地选择工具来应对问题。
- 可靠的响应解析：响应采用JSON格式，而不是通常的对话形式。乍一看可能不太显眼，但正是这一点使得LLM能够通过API等方式连接到外部系统。
这为我们打开了无数的可能性：
- 自主性更强的AI助手：机器人可以与内部系统交互，执行如处理客户订单和退货等任务，而不仅仅是回答问题。
- 个人研究助手：如果你想规划旅行，助手可以搜索网络、抓取内容、比较选项，并将结果汇总到Excel表格中。
- 物联网语音控制：模型可以控制设备或根据检测到的意图来提出建议，比如调整空调的温度。
(稍微岔開一點話題：要實現這些潛力，我們需要一個系統化的辦法來設計和測試我們的提示。我也寫過一篇文章談這個!)
像数据科学家那样设计提示：使用DSPy优化和测试提示towardsdatascience.com直接进入正题，让我们来探讨一下这个函数调用的概念吧！
函数调用的结构是怎样的参考Gemini的函数调用文档，函数调用具有以下结构，如下所示，在OpenAI中情况也一样。
来自: Gemini的函数调用文档页面，的图片
- 用户向应用发出指令
- 应用将用户提供的指令以及功能描述传递给模型，功能描述说明了模型可以使用的工具
- 根据功能描述，模型建议使用哪个工具及其相关请求参数。请注意，模型只会提出建议的工具和参数，而不实际执行这些功能调用
- 和 5. 应用会根据响应调用相应的API
6. API的响应再次被输入模型，以便生成易于人类理解的回复
8. 应用程序把最后的回复给用户，然后重复从1开始。
这可能听起来有些复杂，但我们會通过具体例子来详细解释这个概念。
建筑在进入代码部分之前，先来简单介绍一下示例应用的结构
解决办法在这里我们为来酒店的游客创建了一个助手。这个助手可以使用以下工具来访问其他应用。
get_items,purchase_item: 通过 API 访问存储在数据库中的产品目录，分别用于获取商品列表和完成购买rag_pipeline_func: 通过 Retrieval Augmented Generation (RAG) 连接到文档存储等功能，从无结构文本（如酒店宣传册）中提取信息
- 大型语言模型（LLM） : 通过OpenRouter访问的GPT-4 Turbo。这个流程可以通过修改少量代码来适应其他支持函数调用的LLM，如Gemini。
- 嵌入模型 : all-MiniLM-L6-v2
- 向量数据库 : Haystack的InMemoryDocumentStore
- LLM框架 : Haystack，因其易用性、出色文档和管道构建透明度而选择。本教程实际上是他们关于这一主题的精彩教程的扩展
好了，咱们开始吧！
示例应用程序 准备阶段:前往Github克隆我的代码。以下内容可以在function_calling_demo笔记本中找到哦。
请也创建并激活一个虚拟环境（virtual environment），然后执行 pip install -r requirements.txt 来安装所需的软件包，并确保其正确安装。
我们首先连接到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
# 在运行这段代码之前，请将您的API密钥设置为环境变量
load_dotenv()
OPENROUTER_API_KEY = os.getenv('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”。
chat_generator.run(messages=[ChatMessage.from_user("返回这个文本：测试")])回复应该像这样：
{'replies': [ChatMessage(content="'内容'", role=<ChatRole.ASSISTANT: 'assistant'>, name=None, meta={'model': 'openai/gpt-4-turbo-preview', 'index': 0, 'finish_reason': 'stop', 'usage': {}})]}在这里，我们建立应用程序与两个数据源之间的连接：文档库用于非结构化文本，以及应用数据库通过API连接。
使用管道来标注文档
我们提供了样本文本在 documents 中，让模型进行检索增强型生成（RAG）。这些文本会被转换成向量，并存储在内存文档库中。
从haystack导入Pipeline, Document 类
从haystack.document_stores.in_memory导入InMemoryDocumentStore
从haystack.components.writers导入DocumentWriter
从haystack.components.embedders导入SentenceTransformersDocumentEmbedder
# 示例文档内容
documents = [
Document(content="咖啡店从早上9点开门到下午5点关门。"),
Document(content="健身房从早上6点开门到晚上10点关门。")
]
# 创建文档存储库
document_store = InMemoryDocumentStore()
# 创建一个管道来将文本转换为嵌入，并将这些嵌入存储在文档存储库中
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
{'文档作者': {'文档数': 2}}启动API服务器实例
一个使用Flask创建的API服务器在db_api.py文件中。请在终端中运行python db_api.py来启动该服务器。
This would be shown in the terminal, if successfully executed
请注意，在 db_api.py 中添加了一些初始的数据，請检查一下。
数据库里的示例数据
第二步：定义函数在这里我们准备模型实际需要调用的函数，在完成函数调用之后（如在函数调用结构中所述的步骤4到5）
RAG模块
即 rag_pipeline_func 这个函数。这个函数的目的是让模型通过搜索存储在文档库中的文本来提供答案。我们首先将RAG检索定义为一个Haystack流水线流程。
从haystack.components.embedders导入SentenceTransformersTextEmbedder作为SentenceTransformersTextEmbedder
从haystack.components.retrievers.in_memory导入InMemoryEmbeddingRetriever作为InMemoryEmbeddingRetriever
从haystack.components.builders导入PromptBuilder作为PromptBuilder
从haystack.components.generators导入OpenAIGenerator作为OpenAIGenerator
template = """
根据给定的上下文回答问题。
上下文:
{% for document in documents %}
{{ document.content }}
{% endfor %}
问题: {{ question }}
回答:
"""
rag_pipe = Pipeline()
rag_pipe.add_component('embedder', SentenceTransformersTextEmbedder(模型='sentence-transformers/all-MiniLM-L6-v2'))
rag_pipe.add_component('retriever', InMemoryEmbeddingRetriever(文档存储=文档存储))
rag_pipe.add_component('prompt_builder', PromptBuilder(template=template))
# 注意：我们使用的是OpenAIGenerator，而不是OpenAIChatGenerator，因为后者只接受List[str]类型的输入，无法接受prompt_builder的输出字符串
rag_pipe.add_component('llm', OpenAIGenerator(api_key=Secret.from_env_var('OPENROUTER_API_KEY'), api_base_url='https://openrouter.ai/api/v1', 模型='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 = 「咖啡店什么时候开门？」
rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})这应该产生如下的输出。请注意，模型提供的回复是根据我们之前提供的样本文档里的内容。
{'llm': {'回复内容': ['咖啡店在早上9点开门。'],
'元数据': [{'模型': 'openai/gpt-4-turbo-preview',
'索引': 0,
'完成原因': 'stop',
'使用统计': {'完成的token数': 9,
'提示的token数': 60,
'总的token数': 69,
'费用': 0.00087}]}}我们可以把 rag_pipe 变成一个函数，只输出 replies，不提供其他细节。
def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
return {"reply": result["llm"]["replies"][0]}API请求
我们定义了这两个 get_items 和 purchase_item 函数，用于与数据库交互。
# Flask的默认本地URL，如有需要可以修改
db_base_url = 'http://127.0.0.1:5000'
# 通过requests从数据库中获取数据
import requests
import json
# get_categories 函数是提示中提供的，但不会被用作工具
def get_categories():
response = requests.get(f'{db_base_url}/category')
data = response.json()
return data
# 获取分类数据
def get_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
# 购买商品
def purchase_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()列出工具
现在我们定义好了这些函数，我们需要让模型认识这些函数，并通过描述来指导它们如何使用。
因为我们在使用OpenAI，tools 的格式如下所示，遵循了根据OpenAI的规范要求的格式规范。
工具列表 = [
{
"type": "function",
"function": {
"name": "get_items",
"description": "检索数据库中的项目列表",
"parameters": {
"type": "对象",
"properties": {
"ids": {
"type": "字符串",
"description": "要获取的项目ID的逗号分隔列表",
},
"categories": {
"type": "字符串",
"description": "要获取的项目分类的逗号分隔列表",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "purchase_item",
"description": "购买某个项目",
"parameters": {
"type": "对象",
"properties": {
"id": {
"type": "字符串",
"description": "给定的产品ID，这里不接受产品名称。请先在数据库中查找产品ID。",
},
"quantity": {
"type": "整数",
"description": "购买的项目数量",
},
},
"required": [],
},
}
},
{
"type": "function",
"function": {
"name": "rag_pipeline_func",
"description": "从酒店宣传册中提取信息",
"parameters": {
"type": "对象",
"properties": {
"query": {
"type": "字符串",
"description": "搜索查询，根据用户的对话内容推断。应是一个问题或陈述。",
}
},
"required": ["query"],
},
},
}
]我们现在可以测试函数调用功能所需的条件了。下面我们要做几件事：
- 给模型提供初始提示，以提供一些上下文
- 一个由用户生成的消息样本
- 我们把工具列表传给聊天生成器中的
tools
# 1. 初始提示
context = f"""你是来酒店旅游的游客的助手。
你可以访问一个数据库（包括{get_categories()}这些类别），游客可以在这里购买商品，你也可以查看酒店的宣传册。
如果游客的问题在数据库中找不到答案，你可以看看宣传册。
如果游客的问题在宣传册中找不到答案，你可以建议游客向酒店工作人员咨询。
"""
messages = [
ChatMessage.from_system(context),
# 2. 用户示例消息
ChatMessage.from_user("我可以买咖啡吗？"),
]
# 3. 传递工具列表并调用聊天生成器
response = chat_generator.run(messages=messages, generation_kwargs={"tools": tools})
response ---------- 回复 ----------
{'回复列表': [ChatMessage(content='[{"index": 0, "id": "call_AkTWoiJzx5uJSgKW0WAI1yBB", "function": {"arguments": {"categories":"Food and beverages"}, "name": "get_items"}, "type": "function"}]', role=<ChatRole.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("显示函数名和参数如下：")
print("函数名是：", function_name)
print("函数参数是：", function_args) ---------- 响应内容 ----------
函数名称: get_items
函数参数列表: {‘categories’: ‘食品与饮料’}当遇到另一个问题时，系统会选择一个更相关的工具来使用。
# 另一个问题
messages.append(ChatMessage.from_user("咖啡店在哪里？"))
# 调用聊天生成器，并传递工具列表
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)
print("函数参数:", function_args) ---------- 响应 ----------
函数名: rag_pipeline_func
函数参数: {'query': "咖啡店在哪儿？"}注意一下，这里并没有真正调用任何函数，接下来我们要做这个。
调用函数
然后我们可以把这些参数传递给选定的函数。
## 查找并调用带有给定参数的相应函数
## 可用函数字典
available_functions = {"get_items": get_items, "purchase_item": purchase_item, "rag_pipeline_func": rag_pipeline_func}
函数调用变量 = available_functions[函数名称]
函数响应 = 函数调用变量(**函数参数)
print("打印:", 函数响应) ---------- 回复 ----------
功能响应: {'回复': '提供的上下文仅说明了咖啡馆的营业时间，但未指定其物理位置。因此，无法根据现有信息确定咖啡馆的位置。'}rag_pipeline_func的响应结果可以作为上下文传递给聊天，通过将其追加到messages，以便模型给出最终答案。
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
chat_generator.运行(messages=messages)
response_msg = response["回复"][0]
输出(response_msg.content) 酒店里的咖啡店位置，我建议直接问酒店的工作人员。他们能准确地告诉你在哪里。我们现在已经结束了一轮聊天！
步骤 4: 转换为互动聊天功能上面的代码展示了如何进行函数调用，但我们希望更进一步，让它变成一个互动式的聊天。
在这里我展示两种方法来做这件事，从更原始的 input() 函数在笔记本中直接打印对话，到通过 Streamlit 渲染对话，提供类似 ChatGPT 的聊天界面。
input()
代码是从Haystack的教程中复制的，这使我们能够快速地测试模型的效果。注意：此应用程序旨在演示函数调用的概念，并非旨在非常健壮或可靠，例如，支持同时订购多个项目，避免不正确的信息输出等。
import json
from haystack.dataclasses import ChatMessage, ChatRole
response = None
messages = [
ChatMessage.from_system(context)
]
while True:
# 如果 OpenAI 的响应是一个工具请求
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:
## 解析函数调用的详情
function_name = function_call["function"]["name"]
function_args = json.loads(function_call["function"]["arguments"])
## 找到并调用对应的函数
function_to_call = available_functions[function_name]
function_response = function_to_call(**function_args)
## 通过 `ChatMessage.from_function` 添加函数响应到 messages 列表中
messages.append(ChatMessage.from_function(content=json.dumps(function_response), name=function_name))
# 常规对话
else:
# 如果上一条消息不是系统发出的
if not messages[-1].is_from(ChatRole.SYSTEM):
messages.append(response["replies"][0])
user_input = input("输入您的消息 👇 INFO: 输入 'exit' 或 'quit' 结束对话\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})在 IDE 中开展交互式对话
虽然能用，但我们可能更希望有一个更好看的。
Streamlit的界面
Streamlit 可以将数据脚本转化为可共享的网络应用，为我们的应用程序提供了一个简洁的用户界面。如上所示的代码被改编为位于我的仓库中的 streamlit 文件夹中的 Streamlit 应用。
你可以这样运行它。
- 如果你还没有这么做，使用
python db_api.py启动 API 服务器 - 设置环境变量
OPENROUTER_API_KEY，例如export OPENROUTER_API_KEY = ‘@REPLACE WITH YOUR API KEY’（假设你在使用 Linux 或 git bash 终端） - 在终端中进入
streamlit文件夹，使用cd streamlit - 使用
streamlit run app.py运行 Streamlit。一个新的浏览器标签页应该会自动打开，显示应用程序
差不多就是这样了！希望你也喜欢这篇文章。
Streamlit用户界面
如无特别注明，所有图片均为作者拍摄