Create app.py

app.py

@@ -0,0 +1,220 @@
+# app.py
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+
+import streamlit as st
+from langchain_community.document_loaders import PyPDFLoader
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_community.vectorstores import FAISS
+
+# Google (Gemini) bindings
+import google.generativeai as genai
+from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI
+
+# Load local .env if present (Spaces uses Secrets; .env optional)
+load_dotenv()
+
+# -------------------------
+# Config / paths
+# -------------------------
+# On Hugging Face Spaces, use writable path /workspace or repo root.
+WORKDIR = Path("/workspace") if Path("/workspace").exists() else Path(".")
+DATA_FILE = WORKDIR / "350_QA_dataset.pdf"  # copy dataset into the repo, or mount/upload
+VECTOR_DIR = WORKDIR / "vectorstore"
+
+# Read the Google API key from environment (Spaces secrets show up as env vars)
+GOOGLE_API_KEY = os.environ.get("GOOGLE_API")  # You will set this in Space Secrets
+
+# -------------------------
+# Utility: pick available models dynamically
+# -------------------------
+def pick_models():
+    """
+    Query genai.list_models() and pick:
+      - an embedding model (supports 'embedText')
+      - a generation/chat model (supports 'generateContent' or 'generateText')
+    Returns (embedding_model_name, chat_model_name)
+    """
+    genai.configure(api_key=GOOGLE_API_KEY)
+    models = genai.list_models()
+
+    embed_model = None
+    chat_model = None
+    for m in models:
+        caps = getattr(m, "supported_generation_methods", None) or getattr(m, "capabilities", None) or []
+        # normalize to list of strings
+        caps = list(caps)
+        if "embedText" in caps and embed_model is None:
+            embed_model = m.name
+        if ("generateContent" in caps or "generateText" in caps) and chat_model is None:
+            chat_model = m.name
+        if embed_model and chat_model:
+            break
+
+    # Fallback defaults (if unavailable in your account, the list_models() result will guide you)
+    if embed_model is None:
+        embed_model = "models/text-embedding-004"
+    if chat_model is None:
+        # do not hardcode gemini; fallback to a safe chat model name if available in your account
+        chat_model = "models/chat-bison-001"  # may still 404 if not available; the UI will show available list
+    return embed_model, chat_model
+
+
+# -------------------------
+# Build vector store
+# -------------------------
+def build_vectorstore(embedding_model):
+    if not DATA_FILE.exists():
+        raise FileNotFoundError(f"{DATA_FILE} not found in repo. Upload the PDF to the repo root or /workspace.")
+
+    st.info("Loading PDF and creating chunks (this runs once)...")
+    loader = PyPDFLoader(str(DATA_FILE))
+    docs = loader.load()
+
+    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=150)
+    chunks = splitter.split_documents(docs)
+
+    st.info(f"Using embedding model: {embedding_model}")
+    embeddings = GoogleGenerativeAIEmbeddings(model=embedding_model, google_api_key=GOOGLE_API_KEY)
+
+    vectorstore = FAISS.from_documents(chunks, embeddings)
+    VECTOR_DIR.mkdir(parents=True, exist_ok=True)
+    vectorstore.save_local(str(VECTOR_DIR))
+    st.success("Vector store built and saved.")
+
+
+# -------------------------
+# Load vector store
+# -------------------------
+def load_vectorstore(embedding_model):
+    if not VECTOR_DIR.exists():
+        st.warning("Vectorstore not found. Build it first (button above).")
+        return None
+    embeddings = GoogleGenerativeAIEmbeddings(model=embedding_model, google_api_key=GOOGLE_API_KEY)
+    return FAISS.load_local(str(VECTOR_DIR), embeddings, allow_dangerous_deserialization=True)
+
+
+# -------------------------
+# Generate answer (RAG)
+# -------------------------
+SYSTEM_PROMPT = """
+You are an EV Service Expert Assistant for a customer support team of an electric vehicle manufacturer.
+Your primary knowledge source is an internal 350-entry complaint and resolution knowledge base extracted from "350_QA_dataset.pdf".
+
+You will receive retrieved chunks and the user's query. Use ONLY the retrieved context.
+
+Respond using this structure:
+1. Issue summary
+2. Likely cause / explanation
+3. Recommended solution / actions
+4. When to visit the service center
+
+If no matching context exists, say:
+"This specific issue is not covered in my internal EV complaint database. Based on general patterns, here are some safe next steps..."
+"""
+
+def answer_query(chat_model_name, query, vectorstore, history=None, k=5):
+    # similarity search
+    docs = vectorstore.similarity_search(query, k=k) if vectorstore else []
+    context = "\n\n---\n\n".join(d.page_content for d in docs) if docs else "[No matching context found]"
+
+    # build prompt
+    history_text = ""
+    if history:
+        for m in history:
+            role = "User" if m["role"] == "user" else "Assistant"
+            history_text += f"{role}: {m['content']}\n"
+    if not history_text:
+        history_text = "[No previous messages]"
+
+    prompt = [
+        SYSTEM_PROMPT,
+        "\n\nConversation history:\n",
+        history_text,
+        "\n\nRetrieved context:\n",
+        context,
+        "\n\nCurrent user question:\n",
+        query,
+    ]
+
+    model = genai.GenerativeModel(chat_model_name)
+    resp = model.generate_content(prompt)
+    return resp.text, docs
+
+
+# -------------------------
+# Streamlit UI
+# -------------------------
+st.set_page_config(page_title="EV Service Assistant (Spaces)", layout="centered")
+st.title("⚡ EV Service Expert (Hugging Face Space)")
+
+if not GOOGLE_API_KEY:
+    st.error("Missing Google API key. Go to your Space settings → Secrets and add a secret named GOOGLE_API with your API key.")
+    st.stop()
+
+with st.expander("Available models (fetched from Google)"):
+    try:
+        embed_model_name, chat_model_name = pick_models()
+        st.write("Embedding model selected:", embed_model_name)
+        st.write("Chat model selected:", chat_model_name)
+    except Exception as e:
+        st.error(f"Could not list models: {e}")
+        st.stop()
+
+col1, col2 = st.columns([1, 2])
+with col1:
+    if not VECTOR_DIR.exists():
+        if st.button("Build vector store from PDF"):
+            try:
+                build_vectorstore(embed_model_name)
+            except Exception as e:
+                st.error(f"Failed to build vectorstore: {e}")
+    else:
+        st.success("Vectorstore already exists.")
+        if st.button("Rebuild vectorstore (force)"):
+            try:
+                if VECTOR_DIR.exists():
+                    import shutil
+                    shutil.rmtree(VECTOR_DIR)
+                build_vectorstore(embed_model_name)
+            except Exception as e:
+                st.error(f"Rebuild failed: {e}")
+
+with col2:
+    st.markdown("**System prompt (fixed):**")
+    st.code(SYSTEM_PROMPT, language="text")
+
+# chat UI
+if "messages" not in st.session_state:
+    st.session_state.messages = []
+
+query = st.text_input("Describe the EV issue you want help with:")
+if st.button("Submit") and query.strip():
+    vectorstore = load_vectorstore(embed_model_name)
+    if vectorstore is None:
+        st.error("Vectorstore not available. Build it first.")
+    else:
+        with st.spinner("Searching knowledge base and generating answer..."):
+            try:
+                answer, docs = answer_query(chat_model_name, query, vectorstore, history=st.session_state.messages, k=5)
+            except Exception as e:
+                st.error(f"Generation failed: {e}")
+                answer, docs = "[Error generating answer]", []
+
+        st.session_state.messages.append({"role": "user", "content": query})
+        st.session_state.messages.append({"role": "assistant", "content": answer, "sources": docs})
+
+# render messages
+for msg in st.session_state.messages:
+    if msg["role"] == "user":
+        st.markdown(f"**User:** {msg['content']}")
+    else:
+        st.markdown(f"**Assistant:** {msg['content']}")
+        if msg.get("sources"):
+            with st.expander("Retrieved context"):
+                for i, d in enumerate(msg["sources"], 1):
+                    page = d.metadata.get("page", "n/a")
+                    st.markdown(f"**Chunk {i} — page {page}**")
+                    st.write(d.page_content)
+                    st.markdown("---")