Create app.py

app.py

@@ -0,0 +1,143 @@
+import os
+import asyncio
+import time
+from dotenv import load_dotenv
+import chainlit as cl
+import google.generativeai as genai
+
+# Load environment variables from .env (if any)
+load_dotenv()
+
+# Model selection (change if needed)
+MODEL_NAME = "gemini-2.0-flash"
+
+# Default system prompt
+SYSTEM_PROMPT = "You are a helpful, friendly AI assistant. Provide clear and concise responses."
+
+@cl.on_chat_start
+async def start():
+    """
+    Initialize the chat session with Google Generative (Gemini) client and chat object.
+    Stores client, model, chat, and message history in the user session.
+    """
+    try:
+        api_key = os.environ.get("GOOGLE_API_KEY")
+        if not api_key:
+            raise ValueError("GOOGLE_API_KEY environment variable is not set")
+
+        # Configure the google generative client
+        genai.configure(api_key=api_key)
+
+        # Create model object
+        model = genai.GenerativeModel(MODEL_NAME)
+
+        # Build initial message history with the system prompt
+        message_history = [{"role": "system", "content": SYSTEM_PROMPT}]
+
+        # Start a persistent chat object for this session (Gemini chat API)
+        # (We start it here once and reuse for the session)
+        chat = model.start_chat(history=message_history)
+
+        # Save into Chainlit user session
+        cl.user_session.set("genai_model", model)
+        cl.user_session.set("genai_chat", chat)
+        cl.user_session.set("message_history", message_history)
+
+        await cl.Message(
+            content="Hello! I'm your Gemini-powered assistant. How can I help you today?"
+        ).send()
+
+    except ValueError as e:
+        await cl.Message(
+            content=f"⚠️ Configuration Error: {str(e)}\nPlease make sure GOOGLE_API_KEY is set in the environment variables."
+        ).send()
+    except Exception as e:
+        await cl.Message(content=f"⚠️ Error initializing Gemini client: {e}").send()
+
+
+@cl.on_stop
+async def on_stop():
+    """Cleanup when the chat session ends."""
+    try:
+        cl.user_session.clear()
+    except Exception:
+        pass
+
+
+async def _call_chat_send_message(chat, user_text):
+    """
+    Blocking wrapper to call chat.send_message in a thread.
+    Returns the iterable (list) of response parts produced by the Gemini client.
+    """
+    return await asyncio.to_thread(chat.send_message, user_text)
+
+
+async def handle_error(error: Exception) -> str:
+    """Helper function to format error messages"""
+    if "session" in str(error).lower():
+        return "⚠️ Session error occurred. Please refresh the page and try again."
+    return f"⚠️ An error occurred: {str(error)}"
+
+
+@cl.on_message
+async def main(user_message: cl.Message):
+    """
+    Process user messages and generate AI responses using Gemini.
+    Uses a background thread to run the blocking Gemini client call, then updates
+    the Chainlit message progressively as parts arrive.
+    """
+    # Prepare a response message slot (so we can update progressively)
+    response_message = cl.Message(content="")
+    await response_message.send()
+
+    try:
+        # Retrieve session data
+        chat = cl.user_session.get("genai_chat")
+        message_history = cl.user_session.get("message_history")
+
+        if chat is None or message_history is None:
+            raise ValueError("Session not initialized properly. Please refresh the page.")
+
+        # Add user to history (keep history for your own bookkeeping if desired)
+        user_text = user_message.content or ""
+        message_history.append({"role": "user", "content": user_text})
+        cl.user_session.set("message_history", message_history)
+
+        # Call the blocking Gemini chat.send_message in a background thread
+        # This returns an iterable of response parts (depending on client)
+        response_iterable = await _call_chat_send_message(chat, user_text)
+
+        # Iterate through returned parts and update the message progressively
+        full_response = ""
+        try:
+            for part in response_iterable:
+                # Each part typically has `.text` attribute (defensive)
+                text_piece = getattr(part, "text", None)
+                if text_piece is None:
+                    # fallback to generic string representation
+                    text_piece = str(part)
+                # accumulate and update
+                full_response += text_piece
+                response_message.content = full_response
+                await response_message.update()
+                # small sleep not strictly necessary; keeps UI responsive
+                await asyncio.sleep(0.01)
+        except TypeError:
+            # If response_iterable is not directly iterable, coerce to string
+            full_response = str(response_iterable)
+            response_message.content = full_response
+            await response_message.update()
+
+        # Append assistant response to message history (if you want to keep it)
+        message_history.append({"role": "assistant", "content": full_response})
+        cl.user_session.set("message_history", message_history)
+
+    except Exception as e:
+        error_message = await handle_error(e)
+        # Update the response_message with the error text
+        try:
+            response_message.content = error_message
+            await response_message.update()
+        except Exception:
+            # If updating fails, send a new message
+            await cl.Message(content=error_message).send()