app.py

from openai import AsyncAssistantEventHandler
from openai import AsyncOpenAI
import gradio as gr
import asyncio
import os

# set the keys
client = AsyncOpenAI(
  api_key=os.getenv("OPENAI_API_KEY")
)

assistantID = os.getenv("OPENAI_ASSISTANT_ID")

mypassword = os.getenv("SPACE_PASSWORD")

mytitle = "<h1 align='center'>Quels sont les enjeux de l'intelligence articielle (IA) qui sont au coeur du rapport de la<br/>députée Corinne Cahen sur le budget de l'État luxembourgeois pour l'année 2025 ?</h1>"

mydescription="""
<h3 align='center'>Explorez les 103 recommendations au sujet de l'intelligence artificielle formulées dans le rapport sur le budget 2025 à l'attention du gouvernement.</h3>
<table width=100%>
  <tr>
    <th width=50% bgcolor="Moccasin">Posez vos questions en français ou dans une autre langue :</th>
    <th width=50% bgcolor="Khaki">Réponse du chatbot parlementaire (File-Search Assistant OpenAI) :</th>
  </tr>
</table>
"""


myarticle ="""
<h3>Contexte :</h3>
<p>Cet espace de démonstration sur HuggingFace, qui utilise l'API du <a href="https://platform.openai.com/docs/models">modèle OpenAI</a> 
 gpt-4o-mini-2024-07-18, a été réalisé par <a href="https://github.com/mbarnig">Marco Barnig</a>.</p>
<p>Les réponses des exemples sont enregistrées dans des fichiers locaux.</p>
"""

myinput = [
    gr.Radio(["Recherche", "Exemples"], label = "Les exemples fonctionnent sans mot de passe.", value="Exemples"),
    gr.Textbox(lines=1, label="Entrez le mot de passe secret !", scale=1),
    gr.Textbox(lines=3, label="Que voulez-vous savoir ?", scale=7),
    gr.Radio(["Wien ass de Rapporteur vum Staatsbudget 2025 ?",
              "Qui sont les collaboratrices et collaborateurs du rapporteur de budget ?", 
              "Combien de recommandations sont émises dans le rapport ?", 
              "Veuillez énumérer tous les domaines impactés par l'intelligence artificielle.", 
              "What is the most ambitious recommendation on AI ?",
              "Quelle est la recommandation sur l'IA la plus facile à mettre en oeuvre ?",
              "Qui étaient les membres de la COFI en 2024 ?"
             ], 
             value="Qui sont les collaboratrices et collaborateurs du rapporteur de budget ?",
             label="Exemples")
]

class EventHandler(AsyncAssistantEventHandler):
    def __init__(self) -> None:
        super().__init__()
        self.response_text = ""

    async def on_text_created(self, text) -> None:
        self.response_text += str(text)

    async def on_text_delta(self, delta, snapshot):
        self.response_text += str(delta.value)

    async def on_text_done(self, text):
        pass

    async def on_tool_call_created(self, tool_call):
        self.response_text += f"\n[Tool Call]: {str(tool_call.type)}\n"

    async def on_tool_call_delta(self, delta, snapshot):
        if snapshot.id != getattr(self, "current_tool_call", None):
            self.current_tool_call = snapshot.id
            self.response_text += f"\n[Tool Call Delta]: {str(delta.type)}\n"

        if delta.type == 'code_interpreter':
            if delta.code_interpreter.input:
                self.response_text += str(delta.code_interpreter.input)
            if delta.code_interpreter.outputs:
                self.response_text += "\n\n[Output]:\n"
                for output in delta.code_interpreter.outputs:
                    if output.type == "logs":
                        self.response_text += f"\n{str(output.logs)}"

    async def on_tool_call_done(self, text):
        pass

# Initialize session variables
session_data = {"assistant_id": assistantID, "thread_id": None}

async def initialize_thread():
    # Create a Thread
    thread = await client.beta.threads.create()
    # Store thread ID in session_data for later use
    session_data["thread_id"] = thread.id

async def generate_response(user_input):
    assistant_id = session_data["assistant_id"]
    thread_id = session_data["thread_id"]

    # Add a Message to the Thread
    oai_message = await client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_input
    )

    # Create and Stream a Run
    event_handler = EventHandler()

    async with client.beta.threads.runs.stream(
        thread_id=thread_id,
        assistant_id=assistant_id,
        instructions="Please assist the user with their query.",
        event_handler=event_handler,
    ) as stream:
        # Yield incremental updates
        async for _ in stream:
            await asyncio.sleep(0.1)  # Small delay to mimic streaming
            yield event_handler.response_text

# Gradio interface function (generator)
async def gradio_chat_interface(mode, password, user_input, example):
    if mode == "Exemples":
        filename = example[-6:-2] + ".md"
        file = open("examples/" + filename, "r")
        output = file.read()
        yield output
    else:
        # check the password
        if password == "":
            yield "Veuillez entrer le mot de passe pour faire des recherches !"
        elif password != mypassword:
            yield "Veuillez entrer le mot de passe correct pour faire des recherches !"
        elif user_input == "":
            yield "Le champ de recherche est vide ! Veuillez entrer votre question."
        else:

            # Create a new event loop if none exists (or if we are in a new thread)
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)

            # Initialize the thread if not already done
            if session_data["thread_id"] is None:
                await initialize_thread()

            # Generate and yield responses
            async for response in generate_response(user_input):
                yield response

# Set up Gradio interface with streaming
interface = gr.Interface(
    fn=gradio_chat_interface,
    inputs=myinput,
    outputs="markdown",
    title=mytitle,
    description=mydescription,
    article=myarticle,
    live=False,
    flagging_mode="never"
)

# Launch the Gradio app
interface.launch()