本文旨在解决在使用 OpenAI Assistants API 时，如何正确调用异步函数的问题。通过一个实际案例，我们将探讨如何检测函数是否为异步协程，并使用 asyncio.iscoroutinefunction 和 await 关键字来确保异步函数能够被正确执行。同时，提供了一个 execute_function 辅助函数，简化了异步和同步函数的调用流程。
在使用 OpenAI Assistants API 时，我们经常需要让 GPT 模型调用我们自定义的函数来完成特定任务。这些函数可以是同步的，也可以是异步的。然而，直接调用异步函数可能会遇到问题，导致函数没有被正确执行。本文将介绍如何解决这个问题，并提供一个通用的解决方案。
问题背景
通常，在使用 Assistants API 调用函数时，我们会从 API 的响应中提取函数名和参数，然后直接调用相应的函数。例如：
if "function_call" in assistant_message:
function_name = assistant_message["function_call"]["name"]
function_args = json.loads(assistant_message["function_call"]["arguments"])
result = functions_dict[function_name](**function_args)如果 functions_dict[function_name] 对应的是一个异步函数，直接调用 functions_dict[function_name](**function_args) 并不会执行该异步函数，而是返回一个 coroutine 对象。我们需要使用 await 关键字来执行这个 coroutine 对象。
解决方案：使用 asyncio.iscoroutinefunction 检测并 await 异步函数
为了解决这个问题，我们可以创建一个辅助函数 execute_function，该函数会检查目标函数是否为异步协程函数。如果是，则使用 await 关键字执行；否则，直接调用。
import asyncio
async def execute_function(function_name, function_args):
function_to_call = functions_dict[function_name]
if asyncio.iscoroutinefunction(function_to_call):
return await function_to_call(**function_args)
else:
return function_to_call(**function_args)在这个函数中，asyncio.iscoroutinefunction(function_to_call) 用于检查 function_to_call 是否为一个异步协程函数。如果是，则使用 await 关键字执行 function_to_call(**function_args)；否则，直接执行 function_to_call(**function_args)。
大小仅1兆左右 ，足够轻便的商城系统； 易部署，上传空间即可用，安全，稳定； 容易操作，登陆后台就可设置装饰网站； 并且使用异步技术处理网站数据，表现更具美感。 前台呈现页面，兼容主流浏览器，DIV+CSS页面设计； 如果您有一定的网页设计基础，还可以进行简易的样式修改，二次开发， 发布新样式，调整网站结构，只需修改css目录中的css.css文件即可。 商城网站完全独立，网站源码随时可供您下载
接下来，我们需要在调用函数的地方使用 await execute_function：
if "function_call" in assistant_message:
function_name = assistant_message["function_call"]["name"]
function_args = json.loads(assistant_message["function_call"]["arguments"])
result = await execute_function(function_name, function_args)请注意，因为我们使用了 await 关键字，所以包含这段代码的函数也必须是一个异步函数。
完整示例
下面是一个完整的示例，展示了如何将 execute_function 集成到你的代码中：
import asyncio
import os
import json
import requests
import pickle
from discord.ext import commands
from smartplug import SmartPlug # 假设 smartplug 库已安装
# 假设 functions.json 包含了函数定义
with open("functions.json", 'r') as file:
functions = json.load(file)
def add_numbers(num1, num2):
return num1 + num2
async def toggle_growlight(lightstate):
print("test")
plug = SmartPlug("xx.xx.xx.xx") # 替换为你的智能插座IP
await plug.update()
if lightstate == "on":
print("on")
await plug.turn_on()
return
if lightstate == "off":
print("off")
await plug.turn_off()
return
functions_dict = {
"add_numbers": add_numbers,
"toggle_growlight": toggle_growlight,
}
async def execute_function(function_name, function_args):
function_to_call = functions_dict[function_name]
if asyncio.iscoroutinefunction(function_to_call):
return await function_to_call(**function_args)
else:
return function_to_call(**function_args)
def chat_completion_request(messages, functions=None, function_call=None, model="gpt-4-1106-preview"):
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer " + os.environ.get('OPENAI_API_KEY')
}
json_data = {"model": model, "messages": messages}
if functions is not None:
json_data.update({"functions": functions})
if function_call is not None:
json_data.update({"function_call": function_call})
try:
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=json_data,
)
return response
except Exception as e:
print("Unable to generate ChatCompletion response")
print(f"Exception: {e}")
return e
class QueryCog(commands.Cog):
def __init__(self, bot):
self.bot = bot
@commands.slash_command(pass_context=True, description="Query GPT-4")
async def query(self, ctx, *, query):
await ctx.response.defer()
if not os.path.exists(f"gptcontext/{ctx.author.id}.pickle"):
with open(f"gptcontext/{ctx.author.id}.pickle", "wb") as write_file:
pickle.dump([], write_file) # 初始化为空列表
with open(f"gptcontext/{ctx.author.id}.pickle", "rb") as rf:
chathistory = pickle.load(rf)
chathistory.append({
"role": "user",
"content": f"{query}"
})
chat_response = chat_completion_request(
chathistory, functions=functions
)
assistant_message = chat_response.json()["choices"][0]["message"]
chathistory.append(assistant_message)
if "function_call" in assistant_message:
function_name = assistant_message["function_call"]["name"]
function_args = json.loads(assistant_message["function_call"]["arguments"])
result = await execute_function(function_name, function_args)
chathistory.append({
"role": "function",
"name": function_name,
"content": str(result)
})
chat_response = chat_completion_request(
chathistory, functions=functions
)
assistant_message = chat_response.json()["choices"][0]["message"]
chathistory.append(assistant_message)
if "content" in chat_response.json()["choices"][0]["message"]:
assistant_message_text = chat_response.json()["choices"][0]["message"]["content"]
else:
assistant_message_text = "Function executed successfully, but no further content was provided."
await ctx.respond(f"{assistant_message_text}")
with open(f"gptcontext/{ctx.author.id}.pickle", "wb") as write_file:
pickle.dump(chathistory, write_file)
def setup(bot):
bot.add_cog(QueryCog(bot))注意事项：
- 确保你的代码运行在 asyncio 事件循环中。
- SmartPlug 库需要正确安装和配置。
- 替换示例代码中的 xx.xx.xx.xx 为你的智能插座的实际 IP 地址。
- functions.json 文件应该包含你的函数定义，格式符合 OpenAI Assistants API 的要求。
- 添加了对 chat_response.json()["choices"][0]["message"] 中 content 缺失情况的处理，避免程序崩溃。
- 初始化 gptcontext/{ctx.author.id}.pickle 为空列表，避免首次运行出错。
总结
通过使用 asyncio.iscoroutinefunction 检测函数是否为异步协程，并使用 await 关键字来执行异步函数，我们可以确保 OpenAI Assistants API 能够正确调用我们的自定义函数，无论是同步的还是异步的。execute_function 辅助函数提供了一个简洁通用的方式来处理函数调用，提高了代码的可读性和可维护性。
