Update app.py

app.py

@@ -1,24 +1,27 @@
+# Full merged code including the evaluation logic and tab
 import os
 import gradio as gr
-import tempfile
+import json
+import logging
+from pathlib import Path
+from typing import List
+from dotenv import load_dotenv
+
 from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain_community.vectorstores import Chroma
 from langchain.chains import RetrievalQA
-from langchain_groq import ChatGroq
-from langchain_community.document_loaders import TextLoader, PyPDFLoader
 from langchain.schema import Document
-from pathlib import Path
-from typing import List
-import logging
+from langchain_community.document_loaders import TextLoader, PyPDFLoader
+from langchain_groq import ChatGroq
+
 import numpy as np
 from sklearn.feature_extraction.text import TfidfVectorizer
-from dotenv import load_dotenv
 
-# Logging config
+# Logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# TF-IDF fallback embedding class
+
 class SimpleEmbeddings:
     def __init__(self):
         self.vectorizer = TfidfVectorizer(max_features=384, stop_words='english')
@@ -28,42 +31,57 @@ class SimpleEmbeddings:
         if not self.fitted:
             self.vectorizer.fit(texts)
             self.fitted = True
-        return self.vectorizer.transform(texts).toarray().tolist()
+        embeddings = self.vectorizer.transform(texts)
+        return embeddings.toarray().tolist()
 
     def embed_query(self, text: str) -> List[float]:
         if not self.fitted:
             return [0.0] * 384
-        return self.vectorizer.transform([text]).toarray()[0].tolist()
+        embedding = self.vectorizer.transform([text])
+        return embedding.toarray()[0].tolist()
+
 
-# Core RAG Assistant class
 class RAGAssistant:
     def __init__(self, groq_api_key: str):
         self.groq_api_key = groq_api_key
         self.embeddings = self._init_embeddings()
-        self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
+
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200,
+            length_function=len
+        )
+
         self.learning_vectorstore = None
         self.code_vectorstore = None
-        self.llm = ChatGroq(groq_api_key=groq_api_key, model_name="llama3-70b-8192", temperature=0.1)
+
+        self.llm = ChatGroq(
+            groq_api_key=groq_api_key,
+            model_name="llama3-70b-8192",
+            temperature=0.1
+        )
+
         self.learning_persist_dir = "./chroma_learning_db"
         self.code_persist_dir = "./chroma_code_db"
+
         self._init_vector_stores()
 
     def _init_embeddings(self):
         try:
             from langchain_huggingface import HuggingFaceEmbeddings
-            for model_name in ["all-MiniLM-L6-v2", "paraphrase-MiniLM-L3-v2", "all-mpnet-base-v2"]:
+            models_to_try = ["all-MiniLM-L6-v2", "paraphrase-MiniLM-L3-v2", "all-mpnet-base-v2"]
+            for model_name in models_to_try:
                 try:
                     embeddings = HuggingFaceEmbeddings(
                         model_name=model_name,
                         model_kwargs={'device': 'cpu'},
                         encode_kwargs={'normalize_embeddings': False}
                     )
-                    print(f"Loaded HuggingFace model: {model_name}")
                     return embeddings
-                except Exception as e:
-                    print(f"Failed to load {model_name}: {e}")
+                except Exception:
+                    continue
         except ImportError:
-            print("HuggingFace not installed. Using TF-IDF fallback.")
+            pass
         return SimpleEmbeddings()
 
     def _init_vector_stores(self):
@@ -82,15 +100,15 @@ class RAGAssistant:
         try:
             documents = []
             for file_path in files:
-                try:
-                    loader = PyPDFLoader(file_path) if file_path.lower().endswith(".pdf") else TextLoader(file_path)
-                    docs = loader.load()
-                    documents.extend(docs)
-                except Exception as e:
-                    print(f"Error loading {file_path}: {e}")
+                if file_path.endswith('.pdf'):
+                    loader = PyPDFLoader(file_path)
+                else:
+                    loader = TextLoader(file_path, encoding='utf-8')
+                docs = loader.load()
+                documents.extend(docs)
 
             if not documents:
-                return "\u274C No documents could be loaded. Please check your file type or content."
+                return "No documents could be loaded."
 
             chunks = self.text_splitter.split_documents(documents)
             for chunk in chunks:
@@ -99,147 +117,167 @@ class RAGAssistant:
             if assistant_type == "learning":
                 self.learning_vectorstore.add_documents(chunks)
                 self.learning_vectorstore.persist()
-            else:
+            elif assistant_type == "code":
                 self.code_vectorstore.add_documents(chunks)
                 self.code_vectorstore.persist()
 
-            return f"\u2705 Loaded {len(chunks)} chunks from {len(documents)} documents."
+            return f"Loaded {len(chunks)} chunks from {len(documents)} documents."
         except Exception as e:
-            return f"\u274C Failed: {str(e)}"
+            return f"Error loading documents: {str(e)}"
 
     def get_learning_tutor_response(self, question: str) -> str:
         if not self.learning_vectorstore:
-            return "Please upload learning materials first."
-
+            return "Please upload some learning materials."
         qa_chain = RetrievalQA.from_chain_type(
             llm=self.llm,
             chain_type="stuff",
             retriever=self.learning_vectorstore.as_retriever(search_kwargs={"k": 3}),
             return_source_documents=True
         )
-
-        learning_prompt = f"""
-        You are a helpful educational assistant. Use the following materials to answer:
-
-        {question}
-        """
-        result = qa_chain({"query": learning_prompt})
+        result = qa_chain({"query": question})
         response = result['result']
         if result.get('source_documents'):
             response += "\n\n**Sources:**\n"
             for doc in result['source_documents'][:3]:
-                response += f"- {Path(doc.metadata.get('source', 'unknown')).name}\n"
+                response += f"- {Path(doc.metadata.get('source', 'Unknown')).name}\n"
         return response
 
     def get_code_helper_response(self, question: str) -> str:
         if not self.code_vectorstore:
-            return "Please upload code documents first."
-
+            return "Please upload some code documentation."
         qa_chain = RetrievalQA.from_chain_type(
             llm=self.llm,
             chain_type="stuff",
             retriever=self.code_vectorstore.as_retriever(search_kwargs={"k": 3}),
             return_source_documents=True
         )
-
-        code_prompt = f"""
-        You are a technical assistant helping with code and APIs. Use the following context to respond:
-
-        {question}
-        """
-        result = qa_chain({"query": code_prompt})
+        result = qa_chain({"query": question})
         response = result['result']
         if result.get('source_documents'):
-            response += "\n\n**Sources:**\n"
+            response += "\n\n**Documentation Sources:**\n"
             for doc in result['source_documents'][:3]:
-                response += f"- {Path(doc.metadata.get('source', 'unknown')).name}\n"
+                response += f"- {Path(doc.metadata.get('source', 'Unknown')).name}\n"
         return response
 
-# Helper for Hugging Face uploads
-
-def save_uploaded_files(files):
-    file_paths = []
-    for f in files:
-        if hasattr(f, "data") and hasattr(f, "name"):
-            suffix = Path(f.name).suffix
-            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
-                tmp.write(f.data)
-                file_paths.append(tmp.name)
-        elif hasattr(f, "path"):
-            file_paths.append(f.path)
-    return file_paths
-
-# Gradio interface setup
+
+# --- Evaluation Utilities ---
+def evaluate_retrieval(assistant, assistant_type: str, eval_file: str, k=3):
+    try:
+        if not os.path.exists(eval_file):
+            return f"Evaluation file {eval_file} not found."
+
+        with open(eval_file, 'r', encoding='utf-8') as f:
+            eval_data = json.load(f)
+
+        total = len(eval_data)
+        hits = 0
+        mrr = 0
+
+        for idx, item in enumerate(eval_data):
+            question = item.get("question", "")
+            keywords = item.get("keywords", [])
+            result = assistant.get_learning_tutor_response(question) if assistant_type == "learning" else assistant.get_code_helper_response(question)
+            hit = any(kw.lower() in result.lower() for kw in keywords)
+            if hit:
+                hits += 1
+                mrr += 1 / (idx + 1)
+
+        precision = hits / total if total else 0
+        recall = precision
+        mean_rr = mrr / total if total else 0
+
+        return f"""
+📊 Evaluation Summary ({assistant_type.title()} Assistant):
+- Total Queries: {total}
+- Precision@{k}: {precision:.2f}
+- Recall@{k}: {recall:.2f}
+- MRR: {mean_rr:.2f}
+
+⚙️ Config:
+- Retriever Top-K: {k}
+- Embedding Model: {getattr(assistant.embeddings, 'model_name', 'TF-IDF (fallback)')}
+- Vector Store: ChromaDB
+        """
+    except Exception as e:
+        return f"Evaluation error: {str(e)}"
+
+
 def create_gradio_interface(assistant: RAGAssistant):
     def upload_learning_files(files):
-        if not files:
-            return "No files uploaded."
-        file_paths = save_uploaded_files(files)
-        return assistant.load_documents(file_paths, "learning")
+        return assistant.load_documents([f.name for f in files], "learning") if files else "No files uploaded."
 
     def upload_code_files(files):
-        if not files:
-            return "No files uploaded."
-        file_paths = save_uploaded_files(files)
-        return assistant.load_documents(file_paths, "code")
+        return assistant.load_documents([f.name for f in files], "code") if files else "No files uploaded."
 
     def learning_chat(message, history):
-        if not message.strip(): return history, ""
+        if not message.strip():
+            return history, ""
         response = assistant.get_learning_tutor_response(message)
         history.append((message, response))
         return history, ""
 
     def code_chat(message, history):
-        if not message.strip(): return history, ""
+        if not message.strip():
+            return history, ""
         response = assistant.get_code_helper_response(message)
         history.append((message, response))
         return history, ""
 
-    with gr.Blocks(title="RAG Assistant", theme=gr.themes.Soft()) as demo:
-        gr.Markdown("# 🎓 Learning & 💻 Code Assistant")
+    with gr.Blocks(title="RAG-Based Learning & Code Assistant", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 🎓 RAG-Based Learning & Code Assistant")
+
         with gr.Tabs():
-            with gr.Tab("Learning Tutor"):
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        lf = gr.File(label="Upload Materials", file_types=['.pdf', '.txt'], file_count="multiple")
-                        lbtn = gr.Button("Upload")
-                        lstat = gr.Textbox(label="Upload Status", interactive=False)
-                    with gr.Column(scale=2):
-                        lchat = gr.Chatbot()
-                        lin = gr.Textbox(label="Ask your question")
-                        lsend = gr.Button("Ask")
-                lbtn.click(upload_learning_files, inputs=[lf], outputs=[lstat])
-                lsend.click(learning_chat, inputs=[lin, lchat], outputs=[lchat, lin])
-                lin.submit(learning_chat, inputs=[lin, lchat], outputs=[lchat, lin])
-
-            with gr.Tab("Code Helper"):
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        cf = gr.File(label="Upload Code Docs", file_types=['.pdf', '.txt', '.md', '.py', '.js'], file_count="multiple")
-                        cbtn = gr.Button("Upload")
-                        cstat = gr.Textbox(label="Upload Status", interactive=False)
-                    with gr.Column(scale=2):
-                        cchat = gr.Chatbot()
-                        cin = gr.Textbox(label="Ask a code question")
-                        csend = gr.Button("Ask")
-                cbtn.click(upload_code_files, inputs=[cf], outputs=[cstat])
-                csend.click(code_chat, inputs=[cin, cchat], outputs=[cchat, cin])
-                cin.submit(code_chat, inputs=[cin, cchat], outputs=[cchat, cin])
-
-        gr.Markdown("---\n*Powered by Groq, LangChain, ChromaDB*")
+            with gr.TabItem("📚 Learning Tutor"):
+                learning_files = gr.File(file_types=[".pdf", ".txt", ".md"], file_count="multiple")
+                learning_upload_btn = gr.Button("Upload Materials")
+                learning_status = gr.Textbox()
+                learning_chatbot = gr.Chatbot(height=400)
+                learning_input = gr.Textbox(placeholder="Ask about your learning materials")
+                learning_submit = gr.Button("Ask")
+
+                learning_upload_btn.click(upload_learning_files, [learning_files], [learning_status])
+                learning_submit.click(learning_chat, [learning_input, learning_chatbot], [learning_chatbot, learning_input])
+                learning_input.submit(learning_chat, [learning_input, learning_chatbot], [learning_chatbot, learning_input])
+
+            with gr.TabItem("💻 Code Documentation Helper"):
+                code_files = gr.File(file_types=[".pdf", ".txt", ".md", ".py", ".js", ".json"], file_count="multiple")
+                code_upload_btn = gr.Button("Upload Documentation")
+                code_status = gr.Textbox()
+                code_chatbot = gr.Chatbot(height=400)
+                code_input = gr.Textbox(placeholder="Ask about your codebase or APIs")
+                code_submit = gr.Button("Ask")
+
+                code_upload_btn.click(upload_code_files, [code_files], [code_status])
+                code_submit.click(code_chat, [code_input, code_chatbot], [code_chatbot, code_input])
+                code_input.submit(code_chat, [code_input, code_chatbot], [code_chatbot, code_input])
+
+            with gr.TabItem("📈 Evaluation Bench"):
+                gr.Markdown("Upload a supported document (.pdf, .txt, .md) containing evaluation questions and expected keywords.")
+                eval_file_input = gr.File(file_types=[".json"], file_count="single")
+                eval_assistant_choice = gr.Radio(["learning", "code"], label="Assistant", value="learning")
+                eval_button = gr.Button("Run Evaluation")
+                eval_output = gr.Textbox(lines=10)
+
+                def run_eval(file, assistant_type):
+                    if file is None:
+                        return "Please upload a valid evaluation file."
+                    return evaluate_retrieval(assistant, assistant_type, file.name)
+
+                eval_button.click(run_eval, [eval_file_input, eval_assistant_choice], [eval_output])
 
     return demo
 
+
 def main():
     load_dotenv()
     groq_api_key = os.getenv("GROQ_API_KEY")
     if not groq_api_key:
-        print("Missing GROQ_API_KEY in .env file.")
+        print("Set your GROQ_API_KEY in .env file")
         return
-
     assistant = RAGAssistant(groq_api_key)
     demo = create_gradio_interface(assistant)
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=False, debug=True)
+    demo.launch(server_name="0.0.0.0", server_port=7860, debug=True)
+
 
 if __name__ == "__main__":
     main()