Spaces:
Runtime error
Runtime error
File size: 2,816 Bytes
99ef172 c6dfe2a 99ef172 c6dfe2a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 |
import gradio as gr
import asyncio, httpx
import async_timeout
from loguru import logger
from typing import Optional, List
from pydantic import BaseModel
import os
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv("OPENAI_API_KEY")
class Message(BaseModel):
role: str
content: str
async def make_completion(messages:List[Message], nb_retries:int=3, delay:int=30) -> Optional[str]:
"""
Sends a request to the ChatGPT API to retrieve a response based on a list of previous messages.
"""
header = {
"Content-Type": "application/json",
"Authorization": f"Bearer {API_KEY}"
}
try:
async with async_timeout.timeout(delay=delay):
async with httpx.AsyncClient(headers=header) as aio_client:
counter = 0
keep_loop = True
while keep_loop:
logger.debug(f"Chat/Completions Nb Retries : {counter}")
try:
resp = await aio_client.post(
url = "https://api.openai.com/v1/chat/completions",
json = {
"model": "gpt-3.5-turbo",
"messages": messages
}
)
logger.debug(f"Status Code : {resp.status_code}")
if resp.status_code == 200:
return resp.json()["choices"][0]["message"]["content"]
else:
logger.warning(resp.content)
keep_loop = False
except Exception as e:
logger.error(e)
counter = counter + 1
keep_loop = counter < nb_retries
except asyncio.TimeoutError as e:
logger.error(f"Timeout {delay} seconds !")
return None
async def predict(input, history):
"""
Predict the response of the chatbot and complete a running list of chat history.
"""
history.append({"role": "user", "content": input})
response = await make_completion(history)
history.append({"role": "assistant", "content": response})
messages = [(history[i]["content"], history[i+1]["content"]) for i in range(0, len(history)-1, 2)]
return messages, history
"""
Gradio Blocks low-level API that allows to create custom web applications (here our chat app)
"""
with gr.Blocks() as demo:
logger.info("Starting Demo...")
chatbot = gr.Chatbot(label="WebGPT")
state = gr.State([])
with gr.Row():
txt = gr.Textbox(show_label=False, placeholder="Enter text and press enter").style(container=False)
txt.submit(predict, [txt, state], [chatbot, state])
demo.launch(server_port=8080) |