使用 OpenAI Assistants API 调用异步函数-Python教程-PHP中文网

使用 OpenAI Assistants API 调用异步函数

碧海醫心

发布： 2025-10-19 08:57:05

原创

398人浏览过

使用 openai assistants api 调用异步函数

本文旨在解决在使用 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)。

AppMall应用商店

AI应用商店，提供即时交付、按需付费的人工智能应用服务

查看详情

接下来，我们需要在调用函数的地方使用 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 辅助函数提供了一个简洁通用的方式来处理函数调用，提高了代码的可读性和可维护性。

以上就是使用 OpenAI Assistants API 调用异步函数的详细内容，更多请关注php中文网其它相关文章！