Create app.py

app.py

@@ -0,0 +1,222 @@
+import os
+import time
+import base64
+from io import BytesIO
+
+import gradio as gr
+from dotenv import load_dotenv
+
+# ---- Load .env if present (not required) ----
+load_dotenv()
+
+# Lazy imports so the Space can boot even if one SDK is missing
+def _lazy_import_openai():
+    try:
+        from openai import OpenAI
+        return OpenAI
+    except Exception as e:
+        raise RuntimeError(f"OpenAI SDK not available: {e}")
+
+def _lazy_import_gemini():
+    try:
+        import google.generativeai as genai
+        return genai
+    except Exception as e:
+        raise RuntimeError(f"Google Generative AI SDK not available: {e}")
+
+APP_TITLE = "ZEN Dual-Engine AI — GPT-5 + Nano-Banana (Gemini)"
+SYSTEM_DEFAULT = (
+    "You are ZEN Assistant. Be concise, accurate, and actionable. "
+    "Cite sources when asked. Avoid sensitive data. If an image is provided, describe or analyze it clearly."
+)
+
+# ---- Core model callers ----
+def call_openai(api_key:str, model:str, messages:list, temperature:float=0.4, max_tokens:int=1024):
+    OpenAI = _lazy_import_openai()
+    client = OpenAI(api_key=api_key)
+    # OpenAI chat.completions remains widely supported across models
+    resp = client.chat.completions.create(
+        model=model,
+        messages=messages,
+        temperature=temperature,
+        max_tokens=max_tokens,
+    )
+    return resp.choices[0].message.content
+
+def _pil_to_base64(image):
+    buffer = BytesIO()
+    image.convert("RGB").save(buffer, format="JPEG", quality=90)
+    return base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+def call_gemini(api_key:str, model:str, user_text:str, system_prompt:str, image=None, temperature:float=0.4):
+    genai = _lazy_import_gemini()
+    genai.configure(api_key=api_key)
+    safety_settings = [
+        {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
+        {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
+        {"category": "HARM_CATEGORY_SEXUAL_CONTENT", "threshold": "BLOCK_NONE"},
+        {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
+    ]
+    generation_config = {
+        "temperature": temperature,
+    }
+    model_obj = genai.GenerativeModel(
+        model_name=model,
+        system_instruction=system_prompt,
+        safety_settings=safety_settings,
+        generation_config=generation_config,
+    )
+    parts = [user_text.strip() if user_text else ""]
+    if image is not None:
+        # Gemini supports direct PIL images as parts
+        parts.append(image)
+
+    resp = model_obj.generate_content(parts)
+    if hasattr(resp, "text") and resp.text:
+        return resp.text
+    # Fallback for candidates API shape
+    if getattr(resp, "candidates", None):
+        return resp.candidates[0].content.parts[0].text
+    return "(No response text returned.)"
+
+# ---- Guardrails ----
+BLOCKLIST = ["{{", "}}", "<script", "</script>"]
+
+def blocked(text:str) -> bool:
+    if not text:
+        return False
+    low = text.lower()
+    return any(tok in low for tok in BLOCKLIST)
+
+# ---- Orchestrator ----
+def infer(
+    provider,               # "OpenAI (GPT-5)" or "Google (Nano-Banana)"
+    openai_api_key,
+    google_api_key,
+    model_name,
+    system_prompt,
+    user_message,
+    image,
+    temperature,
+    max_tokens,
+    history
+):
+    history = history or []
+    t0 = time.time()
+
+    if provider.startswith("OpenAI"):
+        api_key = openai_api_key or os.getenv("OPENAI_API_KEY", "")
+        if not api_key:
+            raise gr.Error("Add your OpenAI API key in Settings.")
+        # Build OpenAI messages array (ignore image for OpenAI path to avoid SDK variance)
+        messages = [{"role": "system", "content": system_prompt.strip() or SYSTEM_DEFAULT}]
+        for h in history:
+            messages.append({"role":"user","content": h[0]})
+            messages.append({"role":"assistant","content": h[1]})
+        messages.append({"role":"user","content": user_message})
+
+        if blocked(user_message):
+            assistant = "Request blocked by safety policy. Please rephrase."
+        else:
+            assistant = call_openai(api_key, model_name.strip() or "gpt-5", messages, temperature, max_tokens)
+
+    else:
+        api_key = google_api_key or os.getenv("GOOGLE_API_KEY", "") or os.getenv("GEMINI_API_KEY", "")
+        if not api_key:
+            raise gr.Error("Add your Google (Gemini) API key in Settings.")
+        # Gemini supports multimodal messages; we pass the current prompt + optional image
+        if blocked(user_message):
+            assistant = "Request blocked by safety policy. Please rephrase."
+        else:
+            assistant = call_gemini(
+                api_key=api_key,
+                model=model_name.strip() or "gemini-2.5-nano-banana",
+                user_text=user_message,
+                system_prompt=(system_prompt.strip() or SYSTEM_DEFAULT),
+                image=image,
+                temperature=temperature
+            )
+
+    latency_ms = int((time.time() - t0) * 1000)
+    cost_est = estimate_cost(provider, model_name, user_message, assistant)
+
+    history.append((user_message, assistant))
+    return history, latency_ms, cost_est
+
+# Cheap & cheerful cost estimate stub (tokens ≈ chars/4)
+def estimate_cost(provider, model, prompt, reply):
+    prompt_toks = int(len(prompt or "") / 4)
+    reply_toks  = int(len(reply or "") / 4)
+    total = prompt_toks + reply_toks
+
+    # You can tune these easily
+    if provider.startswith("OpenAI"):
+        # Example: $3 / 1M input + $12 / 1M output → blended rough rate
+        dollars = total / 1_000_000.0 * 7.5
+    else:
+        # Example Gemini/Nano-Banana nominal placeholder
+        dollars = total / 1_000_000.0 * 5.0
+    return round(dollars, 4)
+
+# ---- UI ----
+with gr.Blocks(fill_height=True, theme=gr.themes.Soft()) as demo:
+    gr.Markdown(f"# {APP_TITLE}")
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            provider = gr.Radio(
+                ["OpenAI (GPT-5)", "Google (Nano-Banana)"],
+                value="OpenAI (GPT-5)",
+                label="Engine"
+            )
+            model_name = gr.Textbox(
+                label="Model name",
+                value="gpt-5",
+                placeholder="e.g., gpt-5  •  gemini-2.5-nano-banana"
+            )
+            system_prompt = gr.Textbox(label="System prompt", value=SYSTEM_DEFAULT, lines=3)
+
+            user_message = gr.Textbox(label="Your message", placeholder="Ask anything…", lines=4)
+            image = gr.Image(label="Optional image (Gemini path supports vision)", type="pil")
+
+            with gr.Row():
+                temperature = gr.Slider(0.0, 1.0, value=0.4, step=0.05, label="Temperature")
+                max_tokens = gr.Slider(128, 4096, value=1024, step=64, label="Max tokens")
+
+            send = gr.Button("Send", variant="primary")
+
+        with gr.Column(scale=4):
+            chat = gr.Chatbot(label="Conversation", height=420, type="messages")
+            with gr.Row():
+                latency = gr.Number(label="Latency (ms)", interactive=False)
+                cost = gr.Number(label="Est. cost (USD)", interactive=False)
+
+            with gr.Accordion("Settings • Bring Your Own Keys", open=False):
+                openai_api_key = gr.Textbox(
+                    label="OPENAI_API_KEY", type="password",
+                    placeholder="sk-... (kept in session; not saved)"
+                )
+                google_api_key = gr.Textbox(
+                    label="GOOGLE_API_KEY (Gemini)", type="password",
+                    placeholder="AIza... (kept in session; not saved)"
+                )
+                gr.Markdown(
+                    "You can also set environment variables `OPENAI_API_KEY` and `GOOGLE_API_KEY` "
+                    "in the Space Secrets for a smoother experience."
+                )
+
+    def _on_send(provider, openai_key, google_key, model, sys, msg, img, temp, maxtok, hist):
+        if not (msg and msg.strip()):
+            raise gr.Error("Type a message first.")
+        return infer(provider, openai_key, google_key, model, sys, msg, img, temp, int(maxtok), hist)
+
+    send.click(
+        _on_send,
+        inputs=[provider, openai_api_key, google_api_key, model_name, system_prompt, user_message, image, temperature, max_tokens, chat],
+        outputs=[chat, latency, cost],
+        show_progress="minimal"
+    )
+
+if __name__ == "__main__":
+    # queue() avoids race conditions under load on Spaces
+    demo.queue(max_size=64).launch()