app.py

from os import getenv as os_getenv, path as os_path
from time import sleep
from json import loads as json_loads
import gradio as gr
from openai import OpenAI
client = OpenAI()

assistant_id = os_getenv("OPENAI_ASSISTANT_ID")
vector_id = os_getenv("OPENAI_VECTOR_ID") 

ANS_SHORT = "Court"
ANS_REGULAR = "Normale"
ANS_LONG = "Détaillé"

def chat(user_message, history, state, long_or_short):
    if (state is None) or (not state['user']):
        gr.Warning("Vous devez d'abord vous authentifier")
        yield "Vous devez d'abord vous authentifier"
    else:
        thread = state['thread']
        if thread is None:
            thread = client.beta.threads.create(
                tool_resources={
                    "file_search": {
                        "vector_store_ids": [vector_id]
                    }
                },
            )
            state['thread'] = thread

        client.beta.threads.messages.create(
            thread_id=thread.id,
            role="user",
            content=user_message,
        )

        details = ""
        if state["user"] != "anonymous":
            details += f"User's name is {state['user']}."

        if   long_or_short == ANS_SHORT:
            details = "Please make a complete but short answer"
        elif long_or_short == ANS_LONG:
            details = "Please make complete and detailed answer. Long is better than short. Use tables and lists if you need it"

        with client.beta.threads.runs.stream(
            thread_id=thread.id,
            assistant_id=assistant_id,
            additional_instructions=details,
        ) as stream:
            total = ''
            for delta in stream.text_deltas:
                total += delta
                yield total

def chat_nostream(user_message, history, state):
    if (state is None) or (not state['user']):
        gr.Warning("Vous devez d'abord vous authentifier")
        yield "Vous devez d'abord vous authentifier"
    else:
        thread = state['thread']
        if thread is None:
            thread = client.beta.threads.create(
                tool_resources={
                    "file_search": {
                        "vector_store_ids": [vector_id]
                    }
                }
            )
            state['thread'] = thread

        # Add the user's message to the thread
        client.beta.threads.messages.create(
            thread_id=thread.id,
            role="user",
            content=user_message,
        )

        # Run the Assistant
        run = client.beta.threads.runs.create(thread_id=thread.id,
                                            assistant_id=assistant_id)

        while True:
            run_status = client.beta.threads.runs.retrieve(thread_id=thread.id,
                                                            run_id=run.id)
            print(f"Run status: {run_status.status}")
            if run_status.status == 'completed':
                break
            sleep(5)

        messages = client.beta.threads.messages.list(thread_id=thread.id)
        response = messages.data[0].content[0].text.value

        yield response

def new_state():
    return gr.State({
        "user": None,
        "thread": None,
    })

def auth(token, state):
    tokens=os_getenv("APP_TOKENS", None)
    if tokens is None:
        state["user"] = "anonymous"
    else:
        tokens=json_loads(tokens)
        state["user"] = tokens.get(token, None)
    return "", state

AUTH_JS = """function auth_js(token, state) {
    if (!!document.location.hash) {
        token = document.location.hash
        document.location.hash=""
    }        
    return [token, state]
}
"""

theme = gr.Theme.from_hub("freddyaboulton/[email protected]")
theme.set(
    color_accent_soft="#6272a4",
    button_primary_text_color="*button_secondary_text_color",
    button_primary_background_fill="*button_secondary_background_fill")

with gr.Blocks(
    title="Je suis Le Petit Nicolas",
    fill_height=True, 
    # theme=gr.themes.Base()
    theme=theme,
    css="footer {visibility: hidden}"
    ) as demo:
        state = new_state()

        gr.HTML("""
            <h1>Je suis Le Petit Nicolas</h1>
            <p>Parlez avec Nicolas.</p>
        """)
        with gr.Row(variant="compact"):
            gr.HTML("Tout au long de la réponse")
            long_or_short = gr.Radio(
                choices = [ANS_SHORT, ANS_REGULAR, ANS_LONG],
                value = ANS_REGULAR,
                show_label=False,
                container=False,
                label = "Tout au long de la réponse",
                scale=3)

        gr.ChatInterface(
            chat,
            chatbot=gr.Chatbot(show_label=False, render=False, bubble_full_width=False),
            additional_inputs=[state, long_or_short],
            examples=[
                ["Comment tes amis s'appellent-ils? comment sont?"],
                ["est-ce que tu aimes Noël?"],
            ],
        )

        token = gr.Textbox(visible=False)
        demo.load(auth,
            [token,state],
            [token,state],
            js=AUTH_JS)

demo.launch(
    height=700,
    show_api=False,
    allowed_paths=["."])