本文旨在解决在使用 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 asyncioasync 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)。
接下来，我们需要在调用函数的地方使用 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 asyncioimport osimport jsonimport requestsimport picklefrom discord.ext import commandsfrom smartplug import SmartPlug # 假设 smartplug 库已安装# 假设 functions.json 包含了函数定义with open("functions.json", 'r') as file: functions = json.load(file)def add_numbers(num1, num2): return num1 + num2async 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() returnfunctions_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 eclass 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 辅助函数提供了一个简洁通用的方式来处理函数调用，提高了代码的可读性和可维护性。
以上就是使用 OpenAI Assistants API 调用异步函数的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1377499.html
微信扫一扫 
支付宝扫一扫 
