Create app.py

app.py

@@ -0,0 +1,190 @@
+import gradio as gr
+import fitz
+import numpy as np
+import requests
+import faiss
+import re
+import json
+import pandas as pd
+from docx import Document
+from pptx import Presentation
+from sentence_transformers import SentenceTransformer
+from concurrent.futures import ThreadPoolExecutor
+
+# Configuration
+GROQ_API_KEY = "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"
+MODEL_NAME = "all-MiniLM-L6-v2"
+CHUNK_SIZE = 512
+MAX_TOKENS = 4096
+MODEL = SentenceTransformer(MODEL_NAME)
+WORKERS = 8
+
+class DocumentProcessor:
+    def __init__(self):
+        self.index = faiss.IndexFlatIP(MODEL.get_sentence_embedding_dimension())
+        self.chunks = []
+        self.processor_pool = ThreadPoolExecutor(max_workers=WORKERS)
+
+    def extract_text_from_pptx(self, file_path):
+        try:
+            prs = Presentation(file_path)
+            return " ".join([shape.text for slide in prs.slides for shape in slide.shapes if hasattr(shape, "text")])
+        except Exception as e:
+            print(f"PPTX Error: {str(e)}")
+            return ""
+
+    def extract_text_from_xls_csv(self, file_path):
+        try:
+            if file_path.endswith(('.xls', '.xlsx')):
+                df = pd.read_excel(file_path)
+            else:
+                df = pd.read_csv(file_path)
+            return " ".join(df.astype(str).values.flatten())
+        except Exception as e:
+            print(f"Spreadsheet Error: {str(e)}")
+            return ""
+
+    def extract_text_from_pdf(self, file_path):
+        try:
+            doc = fitz.open(file_path)
+            return " ".join(page.get_text("text", flags=fitz.TEXT_PRESERVE_WHITESPACE) for page in doc)
+        except Exception as e:
+            print(f"PDF Error: {str(e)}")
+            return ""
+
+    def process_file(self, file):
+        try:
+            file_path = file.name
+            if file_path.endswith('.pdf'):
+                text = self.extract_text_from_pdf(file_path)
+            elif file_path.endswith('.docx'):
+                text = " ".join(p.text for p in Document(file_path).paragraphs)
+            elif file_path.endswith('.txt'):
+                with open(file_path, 'r', encoding='utf-8') as f:
+                    text = f.read()
+            elif file_path.endswith('.pptx'):
+                text = self.extract_text_from_pptx(file_path)
+            elif file_path.endswith(('.xls', '.xlsx', '.csv')):
+                text = self.extract_text_from_xls_csv(file_path)
+            else:
+                return ""
+            return re.sub(r'\s+', ' ', text).strip()
+        except Exception as e:
+            print(f"Processing Error: {str(e)}")
+            return ""
+
+    def semantic_chunking(self, text):
+        words = re.findall(r'\S+\s*', text)
+        chunks = [''.join(words[i:i+CHUNK_SIZE//2]) for i in range(0, len(words), CHUNK_SIZE//2)]
+        return chunks[:1000]
+
+    def process_documents(self, files):
+        self.chunks = []
+        if not files:
+            return "No files uploaded!"
+
+        texts = list(self.processor_pool.map(self.process_file, files))
+        with ThreadPoolExecutor(max_workers=WORKERS) as executor:
+            chunk_lists = list(executor.map(self.semantic_chunking, texts))
+
+        all_chunks = [chunk for chunk_list in chunk_lists for chunk in chunk_list]
+        if not all_chunks:
+            return "Error: No chunks generated from documents"
+
+        try:
+            embeddings = MODEL.encode(
+                all_chunks,
+                batch_size=512,
+                convert_to_tensor=True,
+                show_progress_bar=False
+            ).cpu().numpy().astype('float32')
+
+            self.index.reset()
+            self.index.add(embeddings)
+            self.chunks = all_chunks
+            return f"Successfully Processed {len(all_chunks)} chunks from {len(files)} files"
+        except Exception as e:
+            print(f"Embedding Error: {str(e)}")
+            return f"Error: {str(e)}"
+
+    def query(self, question):
+        if not self.chunks:
+            return "Please process documents first", False
+
+        try:
+            question_embedding = MODEL.encode([question], convert_to_tensor=True).cpu().numpy().astype('float32')
+            _, indices = self.index.search(question_embedding, 3)
+            context = "\n".join([self.chunks[i] for i in indices[0] if i < len(self.chunks)])
+
+            response = requests.post(
+                "https://api.groq.com/openai/v1/chat/completions",
+                headers={"Authorization": f"Bearer {GROQ_API_KEY}"},
+                json={
+                    "messages": [{
+                        "role": "user",
+                        "content": f"Answer concisely: {question}\nContext: {context}"
+                    }],
+                    "model": "mixtral-8x7b-32768",
+                    "temperature": 0.3,
+                    "max_tokens": MAX_TOKENS,
+                    "stream": True
+                },
+                timeout=20
+            )
+
+            if response.status_code != 200:
+                return f"API Error: {response.text}", False
+
+            full_answer = []
+            for chunk in response.iter_lines():
+                if chunk:
+                    try:
+                        decoded = chunk.decode('utf-8').strip()
+                        if decoded.startswith('data:'):
+                            data = json.loads(decoded[5:])
+                            if content := data.get('choices', [{}])[0].get('delta', {}).get('content', ''):
+                                full_answer.append(content)
+                    except:
+                        continue
+
+            return ''.join(full_answer), True
+        except Exception as e:
+            print(f"Query Error: {str(e)}")
+            return f"Error: {str(e)}", False
+
+processor = DocumentProcessor()
+
+def ask_question(question, chat_history):
+    if not question.strip():
+        return chat_history
+    answer, success = processor.query(question)
+    return chat_history + [(question, answer if success else f"Error: {answer}")]
+
+with gr.Blocks(title="RAG System", css=".footer {display: none !important}") as app:
+    gr.Markdown("## Multi-Format-Reader")
+    with gr.Row():
+        files = gr.File(file_count="multiple",
+                      file_types=[".pdf", ".docx", ".txt", ".pptx", ".xls", ".xlsx", ".csv"],
+                      label="Upload Documents")
+        process_btn = gr.Button("Process", variant="primary")
+    status = gr.Textbox(label="Processing Status", interactive=False)
+    chatbot = gr.Chatbot(height=500, label="Chat History")
+    with gr.Row():
+        question = gr.Textbox(label="Your Query", placeholder="Enter your question...", max_lines=3)
+        ask_btn = gr.Button("Ask", variant="primary")
+    clear_btn = gr.Button("Clear Chat")
+
+    process_btn.click(
+        processor.process_documents,
+        files,
+        status
+    )
+    ask_btn.click(
+        ask_question,
+        [question, chatbot],
+        chatbot
+    ).then(lambda: "", None, question)
+    clear_btn.click(lambda: [], None, chatbot)
+
+app.launch()
+