Update app.py

app.py

@@ -1,16 +1,10 @@
-# app.py - FULLY WORKING AI Research Agent for Hugging Face Spaces
+# app.py - AI Research Agent with PDF RAG + Voice (100% HF Spaces Compatible - Nov 2025)
 import os
 import re
-import ast
-import operator
 import logging
-import requests
 import tempfile
-import time
 from pathlib import Path
-from typing import List, Dict, Any
-from datetime import datetime
-
+from typing import List, Dict
 import numpy as np
 from tqdm import tqdm
 import PyPDF2
@@ -23,149 +17,34 @@ from gtts import gTTS
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# ========================================
-# TOOLS & UTILITIES
-# ========================================
-class WebSearchTool:
-    def __init__(self, max_results: int = 5):
-        self.max_results = max_results
-        self.base_url = "https://api.duckduckgo.com/"
-
-    def search(self, query: str) -> Dict[str, Any]:
-        try:
-            params = {
-                'q': query, 'format': 'json', 'no_redirect': '1',
-                'no_html': '1', 'skip_disambig': '1'
-            }
-            response = requests.get(self.base_url, params=params, timeout=10)
-            response.raise_for_status()
-            data = response.json()
-
-            results = {
-                'abstract': data.get('Abstract', '') or data.get('Answer', ''),
-                'related': [
-                    {'text': t.get('Text', ''), 'url': t.get('FirstURL', '')}
-                    for t in data.get('RelatedTopics', [])[:self.max_results]
-                    if 'Text' in t
-                ]
-            }
-            return results
-        except Exception as e:
-            logger.error(f"Web search failed: {e}")
-            return {'abstract': '', 'related': []}
-
 # ========================================
 # DOCUMENT PROCESSING
 # ========================================
-class DocumentProcessor:
-    def load_documents(self, data_directory: str) -> List[Dict[str, Any]]:
-        documents = []
-        path = Path(data_directory)
-        for file_path in path.rglob("*.pdf"):
-            try:
-                text = ""
-                with open(file_path, 'rb') as f:
-                    reader = PyPDF2.PdfReader(f)
-                    for page in reader.pages:
-                        page_text = page.extract_text()
-                        if page_text:
-                            text += page_text + "\n"
-                if text.strip():
-                    documents.append({
-                        'doc_id': str(file_path.relative_to(path)),
-                        'content': text,
-                        'file_path': str(file_path)
-                    })
-            except Exception as e:
-                logger.error(f"Error reading {file_path}: {e}")
-        return documents
-
-class DocumentChunker:
-    def __init__(self, chunk_size=512, chunk_overlap=50):
-        self.chunk_size = chunk_size
-        self.chunk_overlap = chunk_overlap
-
-    def chunk_documents(self, documents: List[Dict]) -> List[Dict]:
-        chunks = []
-        for doc in documents:
-            text = re.sub(r'\s+', ' ', doc['content']).strip()
-            start = 0
-            while start < len(text):
-                end = start + self.chunk_size
-                chunk = text[start:end]
-                if end < len(text):
-                    last_space = chunk.rfind(' ')
-                    if last_space > self.chunk_size // 2:
-                        end = start + last_space
-                chunks.append({
-                    'chunk_id': f"{doc['doc_id']}_{start}",
-                    'content': text[start:end].strip(),
-                    'doc_id': doc['doc_id'],
-                    'source_file': doc['file_path']
-                })
-                start = end - self.chunk_overlap
-                if start >= len(text):
-                    break
-        return [c for c in chunks if len(c['content']) > 50]
-
-class EmbeddingGenerator:
-    def __init__(self, model_name='all-MiniLM-L6-v2'):
-        self.model = SentenceTransformer(model_name)
-
-    def generate(self, chunks: List[Dict]) -> np.ndarray:
-        texts = [c['content'] for c in chunks]
-        return self.model.encode(texts, batch_size=32, show_progress_bar=False, convert_to_numpy=True)
-
-    def query_embedding(self, query: str) -> np.ndarray:
-        return self.model.encode([query], convert_to_numpy=True)[0]
-
-# ========================================
-# RETRIEVER
-# ========================================
-class DocumentRetriever:
-    def __init__(self):
-        self.chunks = []
-        self.index = None
-        self.embedder = EmbeddingGenerator()
-
-    def build_index(self, chunks: List[Dict], embeddings: np.ndarray):
-        self.chunks = chunks
-        dim = embeddings.shape[1]
-        self.index = faiss.IndexFlatIP(dim)
-        normalized = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
-        self.index.add(normalized.astype('float32'))
-
-    def search(self, query: str, k: int = 8) -> List[Dict]:
-        if not self.index:
-            return []
-        q_emb = self.embedder.query_embedding(query)
-        q_norm = q_emb / np.linalg.norm(q_emb)
-        scores, indices = self.index.search(q_norm.reshape(1, -1).astype('float32'), k)
-        results = []
-        for score, idx in zip(scores[0], indices[0]):
-            if idx < len(self.chunks):
-                chunk = self.chunks[idx].copy()
-                chunk['similarity'] = float(score)
-                results.append(chunk)
-        return results
-
-# ========================================
-# AGENT TOOLS
-# ========================================
-class AgenticTools:
-    def __init__(self):
-        self.web_search = WebSearchTool()
-
-    def calculator(self, expr: str):
-        try:
-            safe_expr = re.sub(r'[^0-9+\-*/(). ]', '', expr)
-            result = eval(ast.parse(safe_expr, mode='eval').body, {"__builtins__": {}})
-            return {"success": True, "result": result}
-        except:
-            return {"success": False, "result": "Invalid math"}
-
-    def web_search(self, query: str):
-        return {"success": True, "result": self.web_search.search(query)}
+def extract_text_from_pdf(pdf_path: str) -> str:
+    text = ""
+    try:
+        with open(pdf_path, 'rb') as f:
+            reader = PyPDF2.PdfReader(f)
+            for page in reader.pages:
+                page_text = page.extract_text()
+                if page_text:
+                    text += page_text + "\n"
+    except Exception as e:
+        logger.error(f"PDF read error: {e}")
+    return text
+
+def chunk_text(text: str, chunk_size: int = 512, overlap: int = 50) -> List[str]:
+    text = re.sub(r'\s+', ' ', text).strip()
+    chunks = []
+    start = 0
+    while start < len(text):
+        end = start + chunk_size
+        chunk = text[start:end]
+        chunks.append(chunk)
+        start = end - overlap
+        if start >= len(text):
+            break
+    return [c for c in chunks if len(c.strip()) > 30]
 
 # ========================================
 # MAIN AGENT
@@ -173,18 +52,21 @@ class AgenticTools:
 class AgenticRAGAgent:
     def __init__(self):
         self.retriever = None
-        self.groq = Groq(api_key=os.getenv("GROQ_API_KEY")) if os.getenv("GROQ_API_KEY") else None
-        self.tools = AgenticTools()
+        self.chunks = []
+        self.embedder = SentenceTransformer('all-MiniLM-L6-v2')
+        
+        api_key = os.getenv("GROQ_API_KEY")
+        self.groq = Groq(api_key=api_key) if api_key else None
 
-    def clean_for_speech(self, text: str) -> str:
-        text = re.sub(r'\*\*|\*|_|`|\[.*?\]|\(.*?\)', '', text)
+    def clean_for_tts(self, text: str) -> str:
+        text = re.sub(r'[\*_`#\[\]]', '', text)
         text = re.sub(r'\s+', ' ', text).strip()
         return text
 
-    def generate_voice(self, text: str):
-        if not text or not text.strip():
+    def text_to_speech(self, text: str):
+        if not text.strip():
             return None
-        clean = self.clean_for_speech(text)
+        clean = self.clean_for_tts(text)
         try:
             tts = gTTS(text=clean, lang='en')
             tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
@@ -196,118 +78,130 @@ class AgenticRAGAgent:
 
     def upload_pdfs(self, files):
         if not files:
-            return "No files uploaded."
+            return "No files selected."
+
+        data_dir = Path("sample_data")
+        data_dir.mkdir(exist_ok=True)
+
+        all_chunks = []
+        all_texts = []
 
-        os.makedirs("sample_data", exist_ok=True)
-        saved = []
         for file in files:
-            if file.name.lower().endswith(".pdf"):
-                dest = os.path.join("sample_data", os.path.basename(file.name))
-                with open(dest, "wb") as f:
-                    f.write(file.read() if callable(getattr(file, 'read', None)) else file)
-                saved.append(dest)
-
-        if not saved:
-            return "No valid PDF files."
-
-        # Process documents
-        processor = DocumentProcessor()
-        chunker = DocumentChunker()
-        docs = processor.load_documents("sample_data")
-        chunks = chunker.chunk_documents(docs)
-        embedder = EmbeddingGenerator()
-        embeddings = embedder.generate(chunks)
-
-        self.retriever = DocumentRetriever()
-        self.retriever.build_index(chunks, embeddings)
-
-        return f"Loaded {len(saved)} PDFs → {len(chunks)} chunks ready! Ask anything."
-
-    def answer_query(self, query: str, history: list):
-        if not query.strip():
+            if not str(file.name).lower().endswith('.pdf'):
+                continue
+            dest = data_dir / Path(file.name).name
+            # Gradio File upload object handling
+            with open(dest, "wb") as f:
+                if hasattr(file, 'read'):
+                    f.write(file.read())
+                else:
+                    f.write(file)
+
+            text = extract_text_from_pdf(dest)
+            if text.strip():
+                chunks = chunk_text(text)
+                all_chunks.extend(chunks)
+                all_texts.append(f"From {dest.name}: {len(chunks)} chunks")
+
+        if not all_chunks:
+            return "No text extracted from PDFs."
+
+        # Build embeddings + FAISS index
+        embeddings = self.embedder.encode(all_chunks, batch_size=32, show_progress_bar=False)
+        dim = embeddings.shape[1]
+        embeddings = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
+        
+        index = faiss.IndexFlatIP(dim)
+        index.add(embeddings.astype('float32'))
+
+        self.chunks = all_chunks
+        self.index = index
+
+        return f"Loaded {len(files)} PDFs → {len(all_chunks)} chunks ready!\nAsk anything."
+
+    def query(self, question: str, history: List):
+        if not question.strip():
             return history, None
 
         if not history:
             history = []
 
         # Greeting
-        if query.strip().lower() in ["hi", "hello", "hey", "hola"]:
-            resp = "Hello! I'm your AI Research Agent with voice answers. Upload PDFs and ask complex questions!"
-            history.append([query, resp])
-            audio = self.generate_voice(resp)
-            return history, audio
-
-        if not self.retriever:
-            resp = "Please upload at least one PDF document first!"
-            history.append([query, resp])
+        if question.strip().lower() in ["hi", "hello", "hey", "howdy"]:
+            resp = "Hello! I'm your AI Research Agent with voice output. Upload PDFs and ask questions!"
+            history.append([question, resp])
+            return history, self.text_to_speech(resp)
+
+        if not hasattr(self, 'index') or self.index is None:
+            resp = "Please upload and process at least one PDF first!"
+            history.append([question, resp])
             return history, None
 
-        # Retrieve + Answer
-        docs = self.retriever.search(query, k=8)
-        context = "\n\n".join([d['content'][:1000] for d in docs[:5]])
+        # Retrieve
+        q_emb = self.embedder.encode([question])
+        q_emb = q_emb / np.linalg.norm(q_emb)
+        D, I = self.index.search(q_emb.astype('float32'), k=6)
+        context = "\n\n".join([self.chunks[i] for i in I[0] if i < len(self.chunks)])
+
+        prompt = f"""You are an expert assistant.
+Use only the following context to answer:
 
-        prompt = f"""You are an expert research assistant.
-Context from documents:
 {context}
 
-Question: {query}
-Provide a clear, accurate, and concise answer. Use bullet points if helpful."""
+Question: {question}
+Answer clearly and concisely:"""
 
         try:
             if not self.groq:
-                answer = "GROQ_API_KEY not set. Set it in Secrets."
+                answer = "GROQ_API_KEY missing! Add it in Space Secrets."
             else:
                 resp = self.groq.chat.completions.create(
                     model="llama-3.1-70b-versatile",
                     messages=[{"role": "user", "content": prompt}],
                     temperature=0.3,
-                    max_tokens=800
+                    max_tokens=700
                 )
                 answer = resp.choices[0].message.content.strip()
         except Exception as e:
-            answer = f"LLM Error: {str(e)}"
+            answer = f"Error: {str(e)}"
 
-        history.append([query, answer])
-        audio = self.generate_voice(answer)
+        history.append([question, answer])
+        audio = self.text_to_speech(answer)
         return history, audio
 
 # ========================================
-# GRADIO APP
+# GRADIO INTERFACE
 # ========================================
-def create_app():
+def build_app():
     agent = AgenticRAGAgent()
 
-    with gr.Blocks(title="AI Research Agent - RAG + Voice", theme=gr.themes.Soft()) as demo:
-        gr.Markdown("""
-        # 🤖 AI Research Agent (Agentic RAG + Voice)
-        Upload PDFs → Ask complex questions → Get answers with **voice**
-        """)
+    with gr.Blocks(title="AI Research Agent - RAG + Voice 🔊", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# 🤖 AI Research Agent\nUpload PDFs → Ask Questions → Get Voice Answers")
 
         with gr.Row():
-            with gr.Column(scale=3):
+            with gr.Column(scale=4):
                 chatbot = gr.Chatbot(height=600)
-                msg = gr.Textbox(placeholder="Ask about your documents...", label="Your Question")
+                msg = gr.Textbox(placeholder="What do you want to know from your documents?", label="Question")
                 with gr.Row():
-                    send = gr.Button("Send", variant="primary")
-                    clear = gr.Button("Clear Chat")
-                audio_out = gr.Audio(label="Voice Response", autoplay=True, interactive=False)
+                    send = gr.Button("Send 🚀", variant="primary")
+                    clear = gr.Button("Clear")
+                audio = gr.Audio(label="Voice Answer", autoplay=True)
 
             with gr.Column(scale=1):
-                gr.Markdown("### Upload Research PDFs")
-                file_input = gr.Files(file_types=[".pdf"], file_count="multiple")
-                status = gr.Textbox(label="Status", interactive=False, lines=4)
+                gr.Markdown("### Upload PDFs")
+                files = gr.Files(file_types=[".pdf"], file_count="multiple")
+                status = gr.Textbox(label="Status", interactive=False, lines=6)
 
-        # Events
-        def respond(message, chat_history):
-            new_hist, audio_file = agent.answer_query(message, chat_history)
-            return "", new_hist, audio_file
+        def respond(q, h):
+            new_h, aud = agent.query(q, h)
+            return "", new_h, aud
 
-        msg.submit(respond, [msg, chatbot], [msg, chatbot, audio_out])
-        send.click(respond, [msg, chatbot], [msg, chatbot, audio_out])
-        clear.click(lambda: ([], None), outputs=[chatbot, audio_out])
-        file_input.change(agent.upload_pdfs, file_input, status)
+        msg.submit(respond, [msg, chatbot], [msg, chatbot, audio])
+        send.click(respond, [msg, chatbot], [msg, chatbot, audio])
+        clear.click(lambda: ([], None), outputs=[chatbot, audio])
+        files.change(agent.upload_pdfs, files, status)
 
-        gr.Markdown("**Secret Required:** Add `GROQ_API_KEY` in Space Secrets (free at console.groq.com)")
+        gr.Markdown("**Required**: Add `GROQ_API_KEY` in Settings → Secrets (free at [console.groq.com](https://console.groq.com))")
 
     return demo
 
@@ -315,5 +209,5 @@ def create_app():
 # LAUNCH
 # ========================================
 if __name__ == "__main__":
-    app = create_app()
+    app = build_app()
     app.launch(server_name="0.0.0.0", server_port=7860)