Create app.py

app.py

@@ -0,0 +1,481 @@
+import os
+import math
+import requests
+from typing import List, Dict, Any, Tuple
+
+import gradio as gr
+from openai import OpenAI
+
+# -------------------- CONFIG --------------------
+
+CHAT_MODEL = "gpt-5.1"  # Change here if your OpenAI model ID differs
+EMBED_MODEL = "text-embedding-3-large"
+
+DEFAULT_SYSTEM_PROMPT = """You are a Retrieval-Augmented Generation (RAG) assistant.
+
+Rules:
+- Answer ONLY using the provided knowledge base context and system instructions.
+- If the answer is not clearly supported by the context, say "I don’t know based on the current knowledge base."
+- Do not invent sources, statistics, or facts that are not present in the context.
+- When applicable, cite which source you used (e.g., "According to the uploaded PDF" or "Based on zenai.world").
+- Be clear, concise, and structured.
+"""
+
+PRESET_CONFIGS = {
+    "None (manual setup)": {
+        "system": DEFAULT_SYSTEM_PROMPT,
+        "urls": "",
+        "text": "",
+    },
+    "ZEN Sites Deep QA (zenai.world + AI Arena)": {
+        "system": DEFAULT_SYSTEM_PROMPT
+        + "\n\nYou specialize in answering questions about ZEN AI’s mission, programs, and AI Arena.",
+        "urls": "https://zenai.world\nhttps://us.zenai.biz",
+        "text": "ZEN AI builds the first global AI × Web3 literacy movement with youth, homeschool, and professional tracks.",
+    },
+    "Policy Explainer (external PDFs / links)": {
+        "system": DEFAULT_SYSTEM_PROMPT
+        + "\n\nYou act as a neutral policy explainer. Summarize clearly, highlight key risks and opportunities.",
+        "urls": "",
+        "text": "This preset is for uploading AI policy PDFs, legal texts, and reports.",
+    },
+    "Research Notebook / Personal RAG Sandbox": {
+        "system": DEFAULT_SYSTEM_PROMPT
+        + "\n\nYou help the user explore, connect, and synthesize insights from their personal notes and documents.",
+        "urls": "",
+        "text": "Use this as a sandbox for notebooks, transcripts, and long-form notes.",
+    },
+}
+
+
+# -------------------- HELPER FUNCTIONS --------------------
+
+def chunk_text(text: str, max_chars: int = 2000, overlap: int = 200) -> List[str]:
+    """Simple character-based chunking with overlap."""
+    text = (text or "").strip()
+    if not text:
+        return []
+    chunks = []
+    start = 0
+    length = len(text)
+    while start < length:
+        end = min(start + max_chars, length)
+        chunk = text[start:end]
+        chunks.append(chunk)
+        if end >= length:
+            break
+        start = max(0, end - overlap)
+    return chunks
+
+
+def cosine_similarity(a: List[float], b: List[float]) -> float:
+    """Compute cosine similarity between two vectors."""
+    if not a or not b:
+        return 0.0
+    dot = sum(x * y for x, y in zip(a, b))
+    norm_a = math.sqrt(sum(x * x for x in a))
+    norm_b = math.sqrt(sum(y * y for y in b))
+    if norm_a == 0 or norm_b == 0:
+        return 0.0
+    return dot / (norm_a * norm_b)
+
+
+def fetch_url_text(url: str) -> str:
+    """Fetch text from a URL in a very lightweight way."""
+    try:
+        resp = requests.get(url, timeout=10)
+        resp.raise_for_status()
+        # crude HTML stripping: keep text only
+        text = resp.text
+        # Remove basic tags
+        for tag in ["<script", "<style"]:
+            if tag in text:
+                # Truncate at first occurrence of script/style to avoid junk
+                text = text.split(tag)[0]
+        # Replace angle brackets
+        text = text.replace("<", " ").replace(">", " ")
+        return text
+    except Exception as e:
+        return f"[Error fetching {url}: {e}]"
+
+
+def read_file_text(path: str) -> str:
+    """Read text from simple text-based files; skip others safely."""
+    if not path:
+        return ""
+    path_lower = path.lower()
+    try:
+        if any(path_lower.endswith(ext) for ext in [".txt", ".md", ".csv", ".json"]):
+            with open(path, "r", encoding="utf-8", errors="ignore") as f:
+                return f.read()
+        # If you want to support PDFs or DOCX, you can add optional parsing here,
+        # but we avoid extra dependencies to keep the app robust.
+        return f"[Unsupported file type for RAG content: {os.path.basename(path)}]"
+    except Exception as e:
+        return f"[Error reading file {os.path.basename(path)}: {e}]"
+
+
+def build_embeddings(
+    api_key: str,
+    docs: List[Dict[str, Any]],
+) -> Tuple[List[Dict[str, Any]], str]:
+    """Embed all document chunks and return them as KB docs with embeddings."""
+    if not docs:
+        return [], "⚠️ No documents to index."
+
+    client = OpenAI(api_key=api_key)
+    kb_chunks = []
+    total_chunks = 0
+
+    for d in docs:
+        source = d.get("source", "unknown")
+        text = d.get("text", "")
+        chunks = chunk_text(text, max_chars=2000, overlap=200)
+        for idx, ch in enumerate(chunks):
+            try:
+                emb_resp = client.embeddings.create(
+                    model=EMBED_MODEL,
+                    input=ch,
+                )
+                emb = emb_resp.data[0].embedding
+                kb_chunks.append(
+                    {
+                        "id": f"{source}_{idx}",
+                        "source": source,
+                        "text": ch,
+                        "embedding": emb,
+                    }
+                )
+                total_chunks += 1
+            except Exception as e:
+                # Keep going even if one embedding fails
+                kb_chunks.append(
+                    {
+                        "id": f"{source}_{idx}_error",
+                        "source": source,
+                        "text": f"[Error embedding chunk: {e}]",
+                        "embedding": [],
+                    }
+                )
+
+    status = f"✅ Knowledge base built with {len(docs)} documents and {total_chunks} chunks."
+    return kb_chunks, status
+
+
+def retrieve_context(
+    api_key: str,
+    kb: List[Dict[str, Any]],
+    query: str,
+    top_k: int = 5,
+    similarity_threshold: float = 0.25,
+) -> Tuple[str, str]:
+    """Retrieve top-k relevant chunks from KB for the query."""
+    if not kb:
+        return "", "ℹ️ No knowledge base yet. The model will answer from instructions only."
+
+    client = OpenAI(api_key=api_key)
+    try:
+        q_emb_resp = client.embeddings.create(
+            model=EMBED_MODEL,
+            input=query,
+        )
+        q_emb = q_emb_resp.data[0].embedding
+    except Exception as e:
+        return "", f"⚠️ Error creating query embedding: {e}"
+
+    scored = []
+    for d in kb:
+        emb = d.get("embedding") or []
+        if not emb:
+            continue
+        sim = cosine_similarity(q_emb, emb)
+        scored.append((sim, d))
+
+    if not scored:
+        return "", "⚠️ No valid embeddings in KB; cannot retrieve context."
+
+    scored.sort(key=lambda x: x[0], reverse=True)
+    top = [d for (sim, d) in scored[:top_k] if sim >= similarity_threshold]
+
+    if not top:
+        return "", "ℹ️ No chunks passed the similarity threshold; answering from instructions only."
+
+    context_parts = []
+    for idx, d in enumerate(top, start=1):
+        src = d.get("source", "unknown")
+        txt = d.get("text", "")
+        context_parts.append(
+            f"[Chunk {idx} | Source: {src}]\n{txt}\n"
+        )
+
+    context = "\n\n---\n\n".join(context_parts)
+    debug = f"📚 Retrieved {len(top)} chunks from KB (top_k={top_k}, threshold={similarity_threshold})."
+    return context, debug
+
+
+# -------------------- GRADIO CALLBACKS --------------------
+
+def save_api_key(api_key: str):
+    api_key = (api_key or "").strip()
+    if not api_key:
+        return "❌ No API key provided.", ""
+    masked = f"{api_key[:4]}...{api_key[-4:]}" if len(api_key) >= 8 else "******"
+    status = f"✅ API key saved for this session: `{masked}`"
+    return status, api_key
+
+
+def apply_preset(preset_name: str):
+    cfg = PRESET_CONFIGS.get(preset_name) or PRESET_CONFIGS["None (manual setup)"]
+    return cfg["system"], cfg["urls"], cfg["text"]
+
+
+def build_knowledge_base(
+    api_key: str,
+    urls_text: str,
+    raw_text: str,
+    file_paths: List[str] | None,
+):
+    api_key = (api_key or "").strip()
+    if not api_key:
+        return "❌ Please save your OpenAI API key first.", []
+
+    docs = []
+
+    # URLs
+    urls = [u.strip() for u in (urls_text or "").splitlines() if u.strip()]
+    for u in urls:
+        txt = fetch_url_text(u)
+        docs.append({"source": u, "text": txt})
+
+    # Raw text
+    if raw_text and raw_text.strip():
+        docs.append({"source": "Raw Text Block", "text": raw_text})
+
+    # Files
+    if file_paths is not None:
+        if isinstance(file_paths, str):
+            file_paths = [file_paths]
+        for p in file_paths:
+            if not p:
+                continue
+            txt = read_file_text(p)
+            src_name = os.path.basename(p)
+            docs.append({"source": f"File: {src_name}", "text": txt})
+
+    if not docs:
+        return "⚠️ No knowledge sources provided (URLs, text, or files).", []
+
+    kb, status = build_embeddings(api_key, docs)
+    return status, kb
+
+
+def chat_with_rag(
+    user_message: str,
+    api_key: str,
+    kb: List[Dict[str, Any]],
+    system_prompt: str,
+    history: List[Dict[str, str]],
+):
+    user_message = (user_message or "").strip()
+    api_key = (api_key or "").strip()
+    system_prompt = (system_prompt or "").strip()
+
+    if not user_message:
+        return history, history, "❌ Please enter a question."
+
+    if not api_key:
+        return history, history, "❌ Please save your OpenAI API key first."
+
+    if not system_prompt:
+        system_prompt = DEFAULT_SYSTEM_PROMPT
+
+    # Retrieve context from KB
+    context, debug_retrieval = retrieve_context(api_key, kb, user_message)
+
+    client = OpenAI(api_key=api_key)
+
+    # Assemble messages for OpenAI
+    messages = []
+    combined_system = (
+        DEFAULT_SYSTEM_PROMPT.strip()
+        + "\n\n---\n\nUser System Instructions:\n"
+        + system_prompt.strip()
+    )
+    messages.append({"role": "system", "content": combined_system})
+
+    if context:
+        context_block = (
+            "You have access to the following knowledge base context.\n"
+            "You MUST base your answer ONLY on this context and the system instructions.\n"
+            "If the answer is not supported by the context, say you don’t know.\n\n"
+            f"{context}"
+        )
+        messages.append({"role": "system", "content": context_block})
+
+    # Add truncated history for conversational continuity
+    recent_history = history[-10:] if history else []
+    for msg in recent_history:
+        if msg.get("role") in ("user", "assistant"):
+            messages.append(msg)
+
+    # Current user message
+    messages.append({"role": "user", "content": user_message})
+
+    try:
+        resp = client.chat.completions.create(
+            model=CHAT_MODEL,
+            messages=messages,
+            temperature=0.3,
+            max_tokens=900,
+        )
+        answer = resp.choices[0].message.content
+    except Exception as e:
+        answer = f"⚠️ OpenAI API error: {e}"
+
+    # Update history for display and next turn
+    new_history = history + [
+        {"role": "user", "content": user_message},
+        {"role": "assistant", "content": answer},
+    ]
+
+    return new_history, new_history, debug_retrieval
+
+
+def clear_chat():
+    return [], [], ""
+
+
+# -------------------- UI LAYOUT --------------------
+
+with gr.Blocks(title="RAG Chatbot — GPT-5.1 + URLs / Files / Text") as demo:
+    gr.Markdown(
+        """
+# 🔍 RAG Chatbot — GPT-5.1 + URLs / Files / Text
+
+1. Enter your **OpenAI API key** and click **Save**.  
+2. Add knowledge via **URLs**, **uploaded files**, and/or **raw text**.  
+3. Click **Build / Refresh Knowledge Base**.  
+4. Ask questions — the bot will answer **only** from your knowledge and system instructions.
+"""
+    )
+
+    api_key_state = gr.State("")
+    kb_state = gr.State([])
+    chat_state = gr.State([])
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("### 🔑 API & System")
+
+            api_key_box = gr.Textbox(
+                label="OpenAI API Key",
+                placeholder="sk-...",
+                type="password",
+            )
+            save_api_btn = gr.Button("Save API Key", variant="primary")
+            save_status = gr.Markdown("API key not set.")
+
+            preset_dropdown = gr.Dropdown(
+                label="Presets",
+                choices=list(PRESET_CONFIGS.keys()),
+                value="None (manual setup)",
+            )
+
+            system_box = gr.Textbox(
+                label="System Instructions",
+                lines=8,
+                value=DEFAULT_SYSTEM_PROMPT,
+            )
+
+            gr.Markdown("### 📚 Knowledge Sources")
+
+            urls_box = gr.Textbox(
+                label="Knowledge URLs (one per line)",
+                lines=4,
+                placeholder="https://example.com/docs\nhttps://zenai.world",
+            )
+
+            raw_text_box = gr.Textbox(
+                label="Additional Knowledge Text",
+                lines=6,
+                placeholder="Paste any notes, docs, or reference text here...",
+            )
+
+            files_input = gr.File(
+                label="Upload Knowledge Files (.txt, .md, .csv, .json)",
+                file_count="multiple",
+                type="filepath",
+            )
+
+            build_kb_btn = gr.Button(
+                "Build / Refresh Knowledge Base",
+                variant="secondary",
+            )
+            kb_status_md = gr.Markdown("ℹ️ No knowledge base built yet.")
+
+        with gr.Column(scale=2):
+            gr.Markdown("### 💬 RAG Chat")
+
+            chatbot = gr.Chatbot(
+                label="RAG Chatbot (GPT-5.1)",
+                type="messages",
+                height=450,
+            )
+
+            user_input = gr.Textbox(
+                label="Ask a question",
+                lines=3,
+                placeholder="Ask about the content of your URLs, files, or pasted text...",
+            )
+
+            with gr.Row():
+                send_btn = gr.Button("Send", variant="primary")
+                clear_btn = gr.Button("Clear Chat")
+
+            debug_md = gr.Markdown(
+                "ℹ️ Retrieval debug info will appear here after each answer."
+            )
+
+    # Wiring: save API key
+    save_api_btn.click(
+        fn=save_api_key,
+        inputs=[api_key_box],
+        outputs=[save_status, api_key_state],
+    )
+
+    # Wiring: presets
+    preset_dropdown.change(
+        fn=apply_preset,
+        inputs=[preset_dropdown],
+        outputs=[system_box, urls_box, raw_text_box],
+    )
+
+    # Wiring: build knowledge base
+    build_kb_btn.click(
+        fn=build_knowledge_base,
+        inputs=[api_key_state, urls_box, raw_text_box, files_input],
+        outputs=[kb_status_md, kb_state],
+    )
+
+    # Wiring: chat send
+    send_btn.click(
+        fn=chat_with_rag,
+        inputs=[user_input, api_key_state, kb_state, system_box, chat_state],
+        outputs=[chatbot, chat_state, debug_md],
+    )
+
+    user_input.submit(
+        fn=chat_with_rag,
+        inputs=[user_input, api_key_state, kb_state, system_box, chat_state],
+        outputs=[chatbot, chat_state, debug_md],
+    )
+
+    # Wiring: clear chat
+    clear_btn.click(
+        fn=clear_chat,
+        inputs=[],
+        outputs=[chatbot, chat_state, debug_md],
+    )
+
+if __name__ == "__main__":
+    demo.launch()