异步函数调用gpt

nx7onnlm  于 2023-10-24  发布在  其他
关注(0)|答案(1)|浏览(235)
`import json
import requests
import openai

OPENAI_KEY = "XXX"
openai.api_key = 'XXX'

GPT_MODEL = "gpt-3.5-turbo-0613"

def chat_completion_request(messages, functions=None, function_call=None, model=GPT_MODEL):
headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + OPENAI_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

life_url = 'XXX'

def get_movies():
    movies_array = []
    try:
        response = requests.get(life_url)
        data = response.json()
        for i in data['complex'][0]['movies']:
            movies_array.append(i['title'])
        return movies_array
    except:
        return "I am unable to parse this, please try something new."
    

async def similaries(arguments):
    similar_movie = json.loads(arguments)['similar_movie']
    array_strings = get_movies()
    try:
        message = f" En '{array_strings}' hay una pelicula que pueda parecerse a '{similar_movie}'?.Retorna un objeto JSON con las keys 'movie_name'"
        res = await openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": message}],
            temperature=0.0,
        )
        print(res)
        print(res['choices'][0]['message']['content'])
        return res['choices'][0]['message']['content']
    except:
        return "cannot find the movie, please try something new."
    

def get_movie(arguments):
    try:
        movie_name = json.loads(arguments)['movie_name']
        response = requests.get(life_url)
        data = response.json()
        for i in data['complex'][0]['movies']:
            if i['title'] == movie_name:
                return i['title']
    except:
        return "Sorry, I am unable to parse this, please try something new."

def get_director_by_movie(arguments):
    directors_array = []
    try:
        movie_name = json.loads(arguments)['movie_name']
        response = requests.get(life_url)
        data = response.json()
        for i in data['complex'][0]['movies']:
            if i['title'] == movie_name:
                for j in i['cast']:
                    if j['isDirector'] == True:
                        directors_array.append(j['name'])
        return directors_array
    except:
        return "Could not find the movie, please try something new."

print(get_director_by_movie('LA NOCHE DEL CRIMEN - 2D Subtitulada'))

# --------------------------------------------------------------------------------------------

functions = [
    {
        "name": "similaries",
        "description": "It will get the similar movies.",
        "parameters": {
            "type": "object",
            "properties": {
                "similar_movie": {
                    "type": "string",
                    "description": "This is similar name"
                }
            },
            "required": ["similar_movie"],
        },
    },
    {
        "name": "get_director_by_movie",
        "description": "It will get the director of the movie.",
        "parameters": {
            "type": "object",
            "properties": {
                "movie_name": {
                    "type": "string",
                    "description": "This is the name of the movie.",
                }
            },
            "required": ["movie_name"],
        },
    }]

user_input = input(
    "Please enter your question here: (if you want to exit then write 'exit' or 'bye'.) ")

while user_input.strip().lower() != "exit" and user_input.strip().lower() != "bye":
    # prompt
    messages = [{"role": "system", "content": "Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."}]
    messages.append({"role": "user", "content": user_input})

    # calling chat_completion_request to call ChatGPT completion endpoint
    chat_response = chat_completion_request(
        messages, functions=functions
    )

    # fetch response of ChatGPT and call the function
    assistant_message = chat_response.json()["choices"][0]["message"]

    if assistant_message['content']:
        print("Response is: ", assistant_message['content'])
    else:
        fn_name = assistant_message["function_call"]["name"]
        arguments = assistant_message["function_call"]["arguments"]
        function = locals()[fn_name]
        result = function(arguments)
        print("Response is: ", result)

user_input = input("Please enter your question here: ")`

我正在做一个函数调用测试,在我的一个函数中我对gpt进行了一个调用,这很耗时,问题是虽然函数调用工作正常,但它返回了一个promise
Please enter your question here:(if you want to exit then write 'exit' or 'bye'.)Como se llama la pelicula de barbie?Response is:<coroutine object similaries at 0x7f96c92ad690>

nnsrf1az

nnsrf1az1#

你的问题应该很简单
你的函数返回一个协程,而不是你真正想要的。

import asyncio

# ... (your previous code)

async def similaries(arguments):
    similar_movie = json.loads(arguments)['similar_movie']
    array_strings = get_movies()
    try:
        message = f" En '{array_strings}' hay una pelicula que pueda parecerse a '{similar_movie}'?.Retorna un objeto JSON con las keys 'movie_name'"
        res = await openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": message}],
            temperature=0.0,
        )
        print(res)
        print(res['choices'][0]['message']['content'])
        return res['choices'][0]['message']['content']
    except:
        return "cannot find the movie, please try something new."

# ... (your previous code)

loop = asyncio.get_event_loop()

while user_input.strip().lower() != "exit" and user_input.strip().lower() != "bye":
    # ... (your previous code)

    # calling chat_completion_request to call ChatGPT completion endpoint
    chat_response = chat_completion_request(
        messages, functions=functions
    )

    # fetch response of ChatGPT and call the function
    assistant_message = chat_response.json()["choices"][0]["message"]

    if assistant_message['content']:
        print("Response is: ", assistant_message['content'])
    else:
        fn_name = assistant_message["function_call"]["name"]
        arguments = assistant_message["function_call"]["arguments"]
        function = locals()[fn_name]

        if asyncio.iscoroutinefunction(function):
            result = loop.run_until_complete(function(arguments))
        else:
            result = function(arguments)
        
        print("Response is: ", result)

    user_input = input("Please enter your question here: ")

loop.close()

这应该有希望通过一个循环和等待函数来解决你的问题。

相关问题