Update app.py

app.py

@@ -1,445 +1,212 @@
-import os
-import gradio as gr
-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
-import numpy as np
-from sklearn.feature_extraction.text import TfidfVectorizer
-from sklearn.metrics.pairwise import cosine_similarity
-import pickle
-from dotenv import load_dotenv
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-class SimpleEmbeddings:
-    """Simple TF-IDF based embeddings as fallback"""
-    
-    def __init__(self):
-        self.vectorizer = TfidfVectorizer(max_features=384, stop_words='english')
-        self.fitted = False
-    
-    def embed_documents(self, texts: List[str]) -> List[List[float]]:
-        """Embed a list of documents"""
-        if not self.fitted:
-            self.vectorizer.fit(texts)
-            self.fitted = True
-        
-        embeddings = self.vectorizer.transform(texts)
-        return embeddings.toarray().tolist()
-    
-    def embed_query(self, text: str) -> List[float]:
-        """Embed a single query"""
-        if not self.fitted:
-            # If not fitted, return zero vector
-            return [0.0] * 384
-        
-        embedding = self.vectorizer.transform([text])
-        return embedding.toarray()[0].tolist()
-
-class RAGAssistant:
-    def __init__(self, groq_api_key: str):
-        """Initialize the RAG Assistant with Groq API key"""
-        self.groq_api_key = groq_api_key
-        
-        # Initialize embeddings with fallback
-        self.embeddings = self._init_embeddings()
-        
-        self.text_splitter = RecursiveCharacterTextSplitter(
-            chunk_size=1000,
-            chunk_overlap=200,
-            length_function=len
-        )
-        
-        # Initialize separate vector stores for each assistant
-        self.learning_vectorstore = None
-        self.code_vectorstore = None
-        
-        # Initialize LLM
-        self.llm = ChatGroq(
-            groq_api_key=groq_api_key,
-            model_name="llama3-70b-8192",
-            temperature=0.1
-        )
-        
-        # Create persistent directories
-        self.learning_persist_dir = "./chroma_learning_db"
-        self.code_persist_dir = "./chroma_code_db"
-        
-        # Initialize vector stores
-        self._init_vector_stores()
-    
-    def _init_embeddings(self):
-        """Initialize embeddings with multiple fallback options"""
-        # Try HuggingFace first
-        try:
-            from langchain_huggingface import HuggingFaceEmbeddings
-            print("Trying HuggingFace embeddings...")
-            
-            # Try different model names
-            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"Successfully loaded HuggingFace model: {model_name}")
-                    return embeddings
-                except Exception as e:
-                    print(f"Failed to load {model_name}: {e}")
-                    continue
-            
-        except ImportError:
-            print("HuggingFace embeddings not available")
-        
-        # Fallback to simple TF-IDF embeddings
-        print("Using TF-IDF embeddings as fallback...")
-        return SimpleEmbeddings()
-    
-    def _init_vector_stores(self):
-        """Initialize ChromaDB vector stores"""
-        try:
-            # Learning Tutor vector store
-            self.learning_vectorstore = Chroma(
-                persist_directory=self.learning_persist_dir,
-                embedding_function=self.embeddings,
-                collection_name="learning_materials"
-            )
-            
-            # Code Documentation vector store
-            self.code_vectorstore = Chroma(
-                persist_directory=self.code_persist_dir,
-                embedding_function=self.embeddings,
-                collection_name="code_documentation"
-            )
-            
-        except Exception as e:
-            logger.error(f"Error initializing vector stores: {str(e)}")
-            raise
-    
-    def load_documents(self, files: List[str], assistant_type: str) -> str:
-        """Load documents into the appropriate vector store"""
-        try:
-            documents = []
-            
-            for file_path in files:
-                try:
-                    if file_path.endswith('.pdf'):
-                        loader = PyPDFLoader(file_path)
-                    else:
-                        loader = TextLoader(file_path, encoding='utf-8')
-                    
-                    docs = loader.load()
-                    documents.extend(docs)
-                except Exception as e:
-                    print(f"Error loading {file_path}: {e}")
-                    continue
-            
-            if not documents:
-                return "No documents could be loaded. Please check your files."
-            
-            # Split documents into chunks
-            chunks = self.text_splitter.split_documents(documents)
-            
-            # Add metadata to distinguish document types
-            for chunk in chunks:
-                chunk.metadata['assistant_type'] = assistant_type
-            
-            # Add to appropriate vector store
-            if assistant_type == "learning":
-                self.learning_vectorstore.add_documents(chunks)
-                self.learning_vectorstore.persist()
-            elif assistant_type == "code":
-                self.code_vectorstore.add_documents(chunks)
-                self.code_vectorstore.persist()
-            
-            return f"Successfully loaded {len(chunks)} chunks from {len(documents)} documents into {assistant_type} assistant."
-            
-        except Exception as e:
-            logger.error(f"Error loading documents: {str(e)}")
-            return f"Error loading documents: {str(e)}"
-    
-    def get_learning_tutor_response(self, question: str) -> str:
-        """Get response from Learning Tutor"""
-        try:
-            if not self.learning_vectorstore:
-                return "Please upload some learning materials first."
-            
-            # Create retrieval QA chain
-            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
-            )
-            
-            # Custom prompt for learning tutor
-            learning_prompt = f"""
-            You are an AI learning assistant that helps students understand academic concepts. 
-            Based on the provided course materials, answer the student's question clearly and educationally.
-            
-            Guidelines:
-            - Provide clear, educational explanations
-            - Use examples when helpful
-            - Reference specific sources when possible
-            - Adapt to the student's level of understanding
-            - Offer additional practice questions or related concepts when relevant
-            - Maintain an encouraging, supportive tone
-            
-            Student's question: {question}
-            
-            Please provide a helpful, educational response:
-            """
-            
-            result = qa_chain({"query": learning_prompt})
-            
-            # Format response with sources
-            response = result['result']
-            
-            if result.get('source_documents'):
-                response += "\n\n**Sources:**\n"
-                for i, doc in enumerate(result['source_documents'][:3]):
-                    source = doc.metadata.get('source', 'Unknown')
-                    response += f"- {Path(source).name}\n"
-            
-            return response
-            
-        except Exception as e:
-            logger.error(f"Error in learning tutor: {str(e)}")
-            return f"Error generating response: {str(e)}"
-    
-    def get_code_helper_response(self, question: str) -> str:
-        """Get response from Code Documentation Helper"""
-        try:
-            if not self.code_vectorstore:
-                return "Please upload some code documentation first."
-            
-            # Create retrieval QA chain
-            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
-            )
-            
-            # Custom prompt for code helper
-            code_prompt = f"""
-            You are a technical assistant that helps developers understand codebases and APIs.
-            Based on the provided documentation and code examples, answer the developer's question.
-            
-            Guidelines:
-            - Provide practical, actionable guidance
-            - Include relevant code snippets with explanations
-            - Reference specific documentation sections when possible
-            - Highlight important considerations (security, performance, errors)
-            - Suggest related APIs or patterns that might be useful
-            - Use clear, technical language appropriate for developers
-            
-            Developer's question: {question}
-            
-            Please provide a helpful technical response:
-            """
-            
-            result = qa_chain({"query": code_prompt})
-            
-            # Format response with sources
-            response = result['result']
-            
-            if result.get('source_documents'):
-                response += "\n\n**Documentation Sources:**\n"
-                for i, doc in enumerate(result['source_documents'][:3]):
-                    source = doc.metadata.get('source', 'Unknown')
-                    response += f"- {Path(source).name}\n"
-            
-            return response
-            
-        except Exception as e:
-            logger.error(f"Error in code helper: {str(e)}")
-            return f"Error generating response: {str(e)}"
-
-def create_gradio_interface(assistant: RAGAssistant):
-    """Create Gradio interface for the RAG Assistant"""
-    
-    def upload_learning_files(files):
-        if not files:
-            return "No files uploaded."
-        
-        file_paths = [f.name for f in files]
-        return assistant.load_documents(file_paths, "learning")
-    
-    def upload_code_files(files):
-        if not files:
-            return "No files uploaded."
-        
-        file_paths = [f.name for f in files]
-        return assistant.load_documents(file_paths, "code")
-    
-    def learning_chat(message, 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, ""
-        
-        response = assistant.get_code_helper_response(message)
-        history.append((message, response))
-        return history, ""
-    
-    # Create Gradio interface
-    with gr.Blocks(title="RAG-Based Learning & Code Assistant", theme=gr.themes.Soft()) as demo:
-        gr.Markdown("# 🎓 RAG-Based Learning & Code Assistant")
-        gr.Markdown("Upload your documents and ask questions to get intelligent responses!")
-        
-        with gr.Tabs():
-            # Learning Tutor Tab
-            with gr.TabItem("📚 Learning Tutor"):
-                gr.Markdown("### Personalized Learning Assistant")
-                gr.Markdown("Upload textbooks, lecture notes, and study materials to get personalized learning assistance.")
-                
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        learning_files = gr.File(
-                            label="Upload Learning Materials (PDF, TXT)",
-                            file_count="multiple",
-                            file_types=[".pdf", ".txt", ".md"]
-                        )
-                        learning_upload_btn = gr.Button("Upload Materials", variant="primary")
-                        learning_status = gr.Textbox(label="Upload Status", interactive=False)
-                    
-                    with gr.Column(scale=2):
-                        learning_chatbot = gr.Chatbot(
-                            label="Learning Tutor Chat",
-                            height=400
-                        )
-                        learning_input = gr.Textbox(
-                            label="Ask a question about your course materials",
-                            placeholder="e.g., Can you explain the concept of machine learning?"
-                        )
-                        learning_submit = gr.Button("Ask Question", variant="primary")
-                
-                learning_upload_btn.click(
-                    upload_learning_files,
-                    inputs=[learning_files],
-                    outputs=[learning_status]
-                )
-                
-                learning_submit.click(
-                    learning_chat,
-                    inputs=[learning_input, learning_chatbot],
-                    outputs=[learning_chatbot, learning_input]
-                )
-                
-                learning_input.submit(
-                    learning_chat,
-                    inputs=[learning_input, learning_chatbot],
-                    outputs=[learning_chatbot, learning_input]
-                )
-            
-            # Code Documentation Helper Tab
-            with gr.TabItem("💻 Code Documentation Helper"):
-                gr.Markdown("### Developer Documentation Assistant")
-                gr.Markdown("Upload API documentation, code examples, and technical guides to get development assistance.")
-                
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        code_files = gr.File(
-                            label="Upload Code Documentation (PDF, TXT, MD)",
-                            file_count="multiple",
-                            file_types=[".pdf", ".txt", ".md", ".py", ".js", ".json"]
-                        )
-                        code_upload_btn = gr.Button("Upload Documentation", variant="primary")
-                        code_status = gr.Textbox(label="Upload Status", interactive=False)
-                    
-                    with gr.Column(scale=2):
-                        code_chatbot = gr.Chatbot(
-                            label="Code Helper Chat",
-                            height=400
-                        )
-                        code_input = gr.Textbox(
-                            label="Ask about APIs, code examples, or troubleshooting",
-                            placeholder="e.g., How do I implement authentication in this API?"
-                        )
-                        code_submit = gr.Button("Ask Question", variant="primary")
-                
-                code_upload_btn.click(
-                    upload_code_files,
-                    inputs=[code_files],
-                    outputs=[code_status]
-                )
-                
-                code_submit.click(
-                    code_chat,
-                    inputs=[code_input, code_chatbot],
-                    outputs=[code_chatbot, code_input]
-                )
-                
-                code_input.submit(
-                    code_chat,
-                    inputs=[code_input, code_chatbot],
-                    outputs=[code_chatbot, code_input]
-                )
-        
-        # Footer
-        gr.Markdown("---")
-        gr.Markdown("*Powered by LangChain, ChromaDB, and Groq API*")
-    
-    return demo
-
-def main():
-    """Main function to run the application"""
-    # Get Groq API key from environment variable
-    load_dotenv()
-    groq_api_key = os.getenv("GROQ_API_KEY")
-    
-    if not groq_api_key:
-        print("Please set your GROQ_API_KEY environment variable")
-        print("You can get a free API key from: https://console.groq.com/")
-        return
-    
-    try:
-        # Initialize RAG Assistant
-        print("Initializing RAG Assistant...")
-        assistant = RAGAssistant(groq_api_key)
-        
-        # Create and launch Gradio interface
-        demo = create_gradio_interface(assistant)
-        
-        print("Starting RAG-Based Learning & Code Assistant...")
-        print("Access the application at: http://localhost:7860")
-        
-        demo.launch(
-            server_name="0.0.0.0",
-            server_port=7860,
-            share=False,
-            debug=True
-        )
-        
-    except Exception as e:
-        logger.error(f"Error starting application: {str(e)}")
-        print(f"Error: {str(e)}")
-
-if __name__ == "__main__":
+import os
+import gradio as gr
+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
+import numpy as np
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+import pickle
+from dotenv import load_dotenv
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class SimpleEmbeddings:
+    def __init__(self):
+        self.vectorizer = TfidfVectorizer(max_features=384, stop_words='english')
+        self.fitted = False
+
+    def embed_documents(self, texts: List[str]) -> List[List[float]]:
+        if not self.fitted:
+            self.vectorizer.fit(texts)
+            self.fitted = True
+        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
+        embedding = self.vectorizer.transform([text])
+        return embedding.toarray()[0].tolist()
+
+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, 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.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
+            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}
+                    )
+                    return embeddings
+                except:
+                    continue
+        except ImportError:
+            pass
+        return SimpleEmbeddings()
+
+    def _init_vector_stores(self):
+        try:
+            self.learning_vectorstore = Chroma(
+                persist_directory=self.learning_persist_dir,
+                embedding_function=self.embeddings,
+                collection_name="learning_materials"
+            )
+            self.code_vectorstore = Chroma(
+                persist_directory=self.code_persist_dir,
+                embedding_function=self.embeddings,
+                collection_name="code_documentation"
+            )
+        except Exception as e:
+            logger.error(f"Error initializing vector stores: {str(e)}")
+            raise
+
+    def load_documents(self, files: List[str], assistant_type: str) -> str:
+        try:
+            documents = []
+            for file_path in files:
+                try:
+                    loader = PyPDFLoader(file_path) if file_path.endswith('.pdf') else TextLoader(file_path, encoding='utf-8')
+                    docs = loader.load()
+                    documents.extend(docs)
+                except Exception as e:
+                    print(f"Error loading {file_path}: {e}")
+                    continue
+            if not documents:
+                return "No documents could be loaded. Please check your files."
+            chunks = self.text_splitter.split_documents(documents)
+            for chunk in chunks:
+                chunk.metadata['assistant_type'] = assistant_type
+            if assistant_type == "learning":
+                self.learning_vectorstore.add_documents(chunks)
+                self.learning_vectorstore.persist()
+            elif assistant_type == "code":
+                self.code_vectorstore.add_documents(chunks)
+                self.code_vectorstore.persist()
+            return f"Successfully loaded {len(chunks)} chunks from {len(documents)} documents into {assistant_type} assistant."
+        except Exception as e:
+            logger.error(f"Error loading documents: {str(e)}")
+            return f"Error loading documents: {str(e)}"
+
+    def get_learning_tutor_response(self, question: str) -> str:
+        try:
+            if not self.learning_vectorstore:
+                return "Please upload some learning materials first."
+            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
+            )
+            prompt = f"""
+            You are an AI learning assistant. Answer the following student question based on uploaded course materials.
+            Question: {question}
+            """
+            result = qa_chain({"query": prompt})
+            response = result['result']
+            if result.get('source_documents'):
+                response += "\n\n**Sources:**\n"
+                for i, doc in enumerate(result['source_documents'][:3]):
+                    source = doc.metadata.get('source', 'Unknown')
+                    response += f"- {Path(source).name}\n"
+            return response
+        except Exception as e:
+            logger.error(f"Error in learning tutor: {str(e)}")
+            return f"Error generating response: {str(e)}"
+
+    def get_code_helper_response(self, question: str) -> str:
+        try:
+            if not self.code_vectorstore:
+                return "Please upload some code documentation first."
+            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
+            )
+            prompt = f"""
+            You are a code assistant. Answer the following developer question based on uploaded technical documentation.
+            Question: {question}
+            """
+            result = qa_chain({"query": prompt})
+            response = result['result']
+            if result.get('source_documents'):
+                response += "\n\n**Documentation Sources:**\n"
+                for i, doc in enumerate(result['source_documents'][:3]):
+                    source = doc.metadata.get('source', 'Unknown')
+                    response += f"- {Path(source).name}\n"
+            return response
+        except Exception as e:
+            logger.error(f"Error in code helper: {str(e)}")
+            return f"Error generating response: {str(e)}"
+
+def evaluate_retrieval(query: str, ground_truth_docs: List[str], retriever, k: int = 5):
+    try:
+        retrieved_docs = retriever.get_relevant_documents(query)
+        top_k = [doc.page_content for doc in retrieved_docs[:k]]
+        hits = sum([1 for doc in top_k if any(gt.lower() in doc.lower() for gt in ground_truth_docs)])
+        precision = hits / k
+        recall = hits / len(ground_truth_docs) if ground_truth_docs else 0.0
+        print("\n🔍 Query:", query)
+        print("📄 Top-K Retrieved Documents:")
+        for i, doc in enumerate(top_k, 1):
+            print(f"{i}. {doc[:200]}...")
+        print(f"\n📊 Evaluation Results:")
+        print(f"✅ Precision@{k}: {precision:.2f}")
+        print(f"🔁 Recall@{k}: {recall:.2f}")
+        return {
+            f"Precision@{k}": precision,
+            f"Recall@{k}": recall,
+            "Hits": hits,
+            "Retrieved": top_k
+        }
+    except Exception as e:
+        logger.error(f"❌ Error during evaluation: {str(e)}")
+        return {
+            f"Precision@{k}": 0.0,
+            f"Recall@{k}": 0.0,
+            "Hits": 0,
+            "Retrieved": []
+        }
+
+def main():
+    load_dotenv()
+    groq_api_key = os.getenv("GROQ_API_KEY")
+    if not groq_api_key:
+        print("Please set your GROQ_API_KEY environment variable")
+        return
+    assistant = RAGAssistant(groq_api_key)
+
+    # Example Evaluation
+    query = "What is supervised learning?"
+    ground_truth_docs = ["Supervised learning is a type of machine learning where the model learns from labeled data."]
+    evaluate_retrieval(
+        query=query,
+        ground_truth_docs=ground_truth_docs,
+        retriever=assistant.learning_vectorstore.as_retriever(search_kwargs={"k": 5})
+    )
+
+if __name__ == "__main__":
     main()