I was messing around with the Assistants API from OpenAI with my discord bot and overall it has been relatively straightforward. However I cannot figure out how to call an asynchronous function. When running the attached code and executing the command on discord it makes a request successfully and I get a response with the function_call
in it, but the function doesn't actually run. I put a couple print()
in the function to try and figure out what might be the issue and none of them ever show in the console, but there's no errors from either python or OpenAI's API. When I use the simple add_numbers()
function it works perfectly.
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
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")
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,
}
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(system, 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 = functions_dict[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)
assistant_message_text = chat_response.json()["choices"][0]["message"]["content"]
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))
function_name
and funtion_args
both return the correct values, they return this in my console:
toggle_growlight
{'lightstate': 'off'}
Is there something special I have to do in order to call an asynchronous function?
After some further experimenting I managed to get it working so I thought I'd share what I did. I created an execute_function
async helper function that checks if the function from functions_dict
is an async coroutine function. If it is, it awaits the function call; otherwise, it calls it normally. Then I changed result = execute_function(function_name, function_args)
to result = await execute_function(function_name, function_args)
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)
# ...
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)