Create app.py

app.py

@@ -0,0 +1,612 @@
+# -*- coding: utf-8 -*-
+
+"""app
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1pwwcBb5Zlw1DA3u5K8W8mjrwBTBWXc1L
+"""
+
+import gradio as gr
+import numpy as np
+from transformers import pipeline
+import os
+import time
+import groq
+import uuid  # For generating unique filenames
+
+# Updated imports to address LangChain deprecation warnings:
+from langchain_groq import ChatGroq
+from langchain.schema import HumanMessage
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_community.vectorstores import Chroma
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain.docstore.document import Document
+
+# Importing chardet (make sure to add chardet to your requirements.txt)
+import chardet
+
+import fitz  # PyMuPDF for PDFs
+import docx  # python-docx for Word files
+import gtts  # Google Text-to-Speech library
+from pptx import Presentation  # python-pptx for PowerPoint files
+import re
+
+# FastAPI imports
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.responses import JSONResponse, FileResponse
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from typing import Optional
+import io
+import soundfile as sf
+import librosa
+
+# Enhanced Whisper model for speech-to-text with better configuration
+try:
+    transcriber = pipeline(
+        "automatic-speech-recognition",
+        model="openai/whisper-small.en",  # Upgraded from base to small for better accuracy
+        device=-1,  # Use CPU (-1) or GPU (0)
+        chunk_length_s=30,
+        stride_length_s=5,
+        batch_size=8
+    )
+except Exception as e:
+    print(f"Warning: Could not load enhanced Whisper model: {e}")
+    # Fallback to basic model
+    transcriber = pipeline("automatic-speech-recognition", model="openai/whisper-base.en")
+
+# Set API Key (Ensure it's stored securely in an environment variable)
+groq.api_key = os.getenv("GROQ_API_KEY")
+
+# Initialize Chat Model
+chat_model = ChatGroq(model_name="llama-3.3-70b-versatile", api_key=groq.api_key)
+
+# Initialize Embeddings and chromaDB
+os.makedirs("chroma_db", exist_ok=True)
+embedding_model = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
+vectorstore = Chroma(
+    embedding_function=embedding_model,
+    persist_directory="chroma_db"
+)
+
+# Short-term memory for the LLM
+chat_memory = []
+
+# Audio processing parameters
+AUDIO_SAMPLE_RATE = 16000  # Whisper works best with 16kHz
+
+# Prompt for quiz generation with added remark
+quiz_prompt = """
+You are an AI assistant specialized in education and assessment creation. Given an uploaded document or text, generate a quiz with a mix of multiple-choice questions (MCQs) and fill-in-the-blank questions. The quiz should be directly based on the key concepts, facts, and details from the provided material.
+
+Generate 20 Questions.
+
+Remove all unnecessary formatting generated by the LLM, including <think> tags, asterisks, markdown formatting, and any bold or italic text, as well as **, ###, ##, and # tags.
+
+For each question:
+- Provide 4 answer choices (for MCQs), with only one correct answer.
+- Ensure fill-in-the-blank questions focus on key terms, phrases, or concepts from the document.
+- Include an answer key for all questions.
+- Ensure questions vary in difficulty and encourage comprehension rather than memorization.
+- Additionally, implement an instant feedback mechanism:
+    - When a user selects an answer, indicate whether it is correct or incorrect.
+    - If incorrect, provide a brief explanation from the document to guide learning.
+    - Ensure responses are concise and educational to enhance understanding.
+
+Output Example:
+1. Fill in the blank: The LLM Agent framework has a central decision-making unit called the _______________________.
+
+Answer: Agent Core
+
+Feedback: The Agent Core is the central component of the LLM Agent framework, responsible for managing goals, tool instructions, planning modules, memory integration, and agent persona.
+
+2. What is the main limitation of LLM-based applications?
+a) Limited token capacity
+b) Lack of domain expertise
+c) Prone to hallucination
+d) All of the above
+
+Answer: d) All of the above
+
+Feedback: LLM-based applications have several limitations, including limited token capacity, lack of domain expertise, and being prone to hallucination, among others.
+
+3. Given the following info, what is the value of P(jam|Rain)?
+P(no Rain) = 0.8;
+P(no Jam) = 0.2;
+P(Rain|Jam) = 0.1
+
+a) 0.016
+b) 0.025
+c) 0.1
+d) 0.4
+
+Answer: d) 0.4
+
+Feedback: This question tests understanding of Bayes' Theorem by requiring the calculation of conditional probability using the given values.
+"""
+
+# Function to clean AI response by removing unwanted formatting
+def clean_response(response):
+    """Removes <think> tags, asterisks, and markdown formatting."""
+    cleaned_text = re.sub(r"<think>.*?</think>", "", response, flags=re.DOTALL)
+    cleaned_text = re.sub(r"(\*\*|\*|\[|\])", "", cleaned_text)
+    cleaned_text = re.sub(r"^##+\s*", "", cleaned_text, flags=re.MULTILINE)
+    cleaned_text = re.sub(r"\\", "", cleaned_text)
+    cleaned_text = re.sub(r"---", "", cleaned_text)
+    return cleaned_text.strip()
+
+# Function to generate quiz based on content
+def generate_quiz(content):
+    prompt = f"{quiz_prompt}\n\nDocument content:\n{content}"
+    response = chat_model([HumanMessage(content=prompt)])
+    cleaned_response = clean_response(response.content)
+    return cleaned_response
+
+# Function to retrieve relevant documents from vectorstore based on user query
+def retrieve_documents(query):
+    results = vectorstore.similarity_search(query, k=3)
+    return [doc.page_content for doc in results]
+
+# Function to convert tuple format to message format
+def convert_to_message_format(chat_history):
+    """Convert from [(user, bot)] format to [{"role": "user", "content": user}, {"role": "assistant", "content": bot}] format"""
+    message_format = []
+    for user_msg, bot_msg in chat_history:
+        message_format.append({"role": "user", "content": user_msg})
+        message_format.append({"role": "assistant", "content": bot_msg})
+    return message_format
+
+# Function to convert message format to tuple format for processing
+def convert_to_tuple_format(chat_history):
+    """Convert from message format back to tuple format for processing"""
+    tuple_format = []
+    for i in range(0, len(chat_history), 2):
+        if i+1 < len(chat_history):
+            user_msg = chat_history[i]["content"]
+            bot_msg = chat_history[i+1]["content"]
+            tuple_format.append((user_msg, bot_msg))
+    return tuple_format
+
+# Function to handle chatbot interactions with short-term memory
+def chat_with_groq(user_input, chat_history):
+    try:
+        # Convert message format to tuple format for processing
+        tuple_history = convert_to_tuple_format(chat_history)
+        
+        # Retrieve relevant documents for additional context
+        relevant_docs = retrieve_documents(user_input)
+        context = "\n".join(relevant_docs) if relevant_docs else "No relevant documents found."
+
+        # Construct proper prompting with conversation history
+        system_prompt = "You are a helpful AI assistant. Answer questions accurately and concisely."
+        conversation_history = "\n".join(chat_memory[-10:])  # Keep the last 10 exchanges
+        prompt = f"{system_prompt}\n\nConversation History:\n{conversation_history}\n\nUser Input: {user_input}\n\nContext:\n{context}"
+
+        # Call the chat model
+        response = chat_model([HumanMessage(content=prompt)])
+
+        # Clean response to remove any unwanted formatting
+        cleaned_response_text = clean_response(response.content)
+
+        # Append conversation history
+        chat_memory.append(f"User: {user_input}")
+        chat_memory.append(f"AI: {cleaned_response_text}")
+
+        # Update chat history - add new messages in the correct format
+        chat_history.append({"role": "user", "content": user_input})
+        chat_history.append({"role": "assistant", "content": cleaned_response_text})
+
+        # Convert response to speech
+        audio_file = speech_playback(cleaned_response_text)
+
+        return chat_history, "", audio_file
+    except Exception as e:
+        error_msg = f"Error: {str(e)}"
+        chat_history.append({"role": "user", "content": user_input})
+        chat_history.append({"role": "assistant", "content": error_msg})
+        return chat_history, "", None
+
+# Function to play response as speech using gTTS
+def speech_playback(text):
+    try:
+        # Generate a unique filename for each audio file
+        unique_id = str(uuid.uuid4())
+        audio_file = f"output_audio_{unique_id}.mp3"
+
+        # Convert text to speech
+        tts = gtts.gTTS(text, lang='en')
+        tts.save(audio_file)
+
+        # Return the path to the audio file
+        return audio_file
+    except Exception as e:
+        print(f"Error in speech_playback: {e}")
+        return None
+
+# Function to detect encoding safely
+def detect_encoding(file_path):
+    try:
+        with open(file_path, "rb") as f:
+            raw_data = f.read(4096)
+            detected = chardet.detect(raw_data)
+            encoding = detected["encoding"]
+        return encoding if encoding else "utf-8"
+    except Exception:
+        return "utf-8"
+
+# Function to extract text from PDF
+def extract_text_from_pdf(pdf_path):
+    try:
+        doc = fitz.open(pdf_path)
+        text = "\n".join([page.get_text("text") for page in doc])
+        return text if text.strip() else "No extractable text found."
+    except Exception as e:
+        return f"Error extracting text from PDF: {str(e)}"
+
+# Function to extract text from Word files (.docx)
+def extract_text_from_docx(docx_path):
+    try:
+        doc = docx.Document(docx_path)
+        text = "\n".join([para.text for para in doc.paragraphs])
+        return text if text.strip() else "No extractable text found."
+    except Exception as e:
+        return f"Error extracting text from Word document: {str(e)}"
+
+# Function to extract text from PowerPoint files (.pptx)
+def extract_text_from_pptx(pptx_path):
+    try:
+        presentation = Presentation(pptx_path)
+        text = ""
+        for slide in presentation.slides:
+            for shape in slide.shapes:
+                if hasattr(shape, "text"):
+                    text += shape.text + "\n"
+        return text if text.strip() else "No extractable text found."
+    except Exception as e:
+        return f"Error extracting text from PowerPoint: {str(e)}"
+
+# Function to process documents safely
+def process_document(file):
+    try:
+        file_extension = os.path.splitext(file.name)[-1].lower()
+        if file_extension in [".png", ".jpg", ".jpeg"]:
+            return "Error: Images cannot be processed for text extraction."
+        if file_extension == ".pdf":
+            content = extract_text_from_pdf(file.name)
+        elif file_extension == ".docx":
+            content = extract_text_from_docx(file.name)
+        elif file_extension == ".pptx":
+            content = extract_text_from_pptx(file.name)
+        else:
+            encoding = detect_encoding(file.name)
+            with open(file.name, "r", encoding=encoding, errors="replace") as f:
+                content = f.read()
+        text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
+        documents = [Document(page_content=chunk) for chunk in text_splitter.split_text(content)]
+        vectorstore.add_documents(documents)
+
+        quiz = generate_quiz(content)
+        return f"Document processed successfully (File Type: {file_extension}). Quiz generated:\n{quiz}"
+    except Exception as e:
+        return f"Error processing document: {str(e)}"
+
+# Enhanced function to handle speech-to-text conversion with audio preprocessing
+def preprocess_audio(audio_data, sample_rate):
+    """
+    Enhanced audio preprocessing for better STT accuracy
+    """
+    try:
+        # Convert to mono if stereo
+        if audio_data.ndim > 1:
+            audio_data = np.mean(audio_data, axis=1)
+        
+        # Convert to float32
+        audio_data = audio_data.astype(np.float32)
+        
+        # Normalize audio
+        max_val = np.max(np.abs(audio_data))
+        if max_val > 0:
+            audio_data = audio_data / max_val
+        
+        # Resample to 16kHz if needed (Whisper works best with 16kHz)
+        if sample_rate != AUDIO_SAMPLE_RATE:
+            audio_data = librosa.resample(audio_data, orig_sr=sample_rate, target_sr=AUDIO_SAMPLE_RATE)
+            sample_rate = AUDIO_SAMPLE_RATE
+        
+        # Apply noise reduction (simple high-pass filter)
+        import scipy.signal as sp
+        nyquist = sample_rate / 2
+        cutoff = 80  # High-pass filter cutoff frequency in Hz
+        b, a = sp.butter(2, cutoff/nyquist, btype='high')
+        audio_data = sp.filtfilt(b, a, audio_data)
+        
+        return audio_data, sample_rate
+    
+    except Exception as e:
+        print(f"Audio preprocessing error: {e}")
+        # Return original audio if preprocessing fails
+        return audio_data, sample_rate
+
+def transcribe_audio(audio):
+    """
+    Enhanced speech-to-text transcription with better error handling and preprocessing
+    """
+    try:
+        if audio is None:
+            return "No audio input detected."
+        
+        sample_rate, audio_data = audio
+        
+        # Preprocess audio
+        audio_data, sample_rate = preprocess_audio(audio_data, sample_rate)
+        
+        # Ensure audio is not too short
+        if len(audio_data) / sample_rate < 0.5:  # Less than 0.5 seconds
+            return "Audio too short. Please record at least 1 second of audio."
+        
+        # Ensure audio is not too long (to prevent timeouts)
+        max_duration = 30  # seconds
+        if len(audio_data) / sample_rate > max_duration:
+            # Truncate audio
+            max_samples = max_duration * sample_rate
+            audio_data = audio_data[:max_samples]
+        
+        # Use Whisper with better configuration
+        result = transcriber({
+            "sampling_rate": sample_rate, 
+            "raw": audio_data
+        })
+        
+        transcription = result["text"].strip()
+        
+        if not transcription:
+            return "No speech detected. Please try again with clearer audio."
+        
+        return transcription
+        
+    except Exception as e:
+        error_msg = f"Transcription error: {str(e)}"
+        print(error_msg)
+        return f"Sorry, I couldn't process the audio. Please try again. Error: {str(e)}"
+
+# FastAPI Application
+app = FastAPI(title="Tutor AI API", description="Enhanced Speech-to-Text Tutor AI API")
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# FastAPI Routes
+@app.get("/")
+async def root():
+    return {"message": "Tutor AI API is running", "version": "1.0"}
+
+@app.post("/api/transcribe")
+async def api_transcribe_audio(file: UploadFile = File(...)):
+    """
+    Enhanced API endpoint for speech-to-text transcription
+    """
+    try:
+        # Check if file is audio
+        if not file.content_type.startswith('audio/'):
+            raise HTTPException(status_code=400, detail="File must be an audio file")
+        
+        # Read audio file
+        contents = await file.read()
+        
+        # Convert to numpy array using soundfile
+        audio_io = io.BytesIO(contents)
+        audio_data, sample_rate = sf.read(audio_io)
+        
+        # Transcribe
+        transcription = transcribe_audio((sample_rate, audio_data))
+        
+        return JSONResponse({
+            "success": True,
+            "transcription": transcription,
+            "audio_duration": len(audio_data) / sample_rate if audio_data is not None else 0
+        })
+        
+    except Exception as e:
+        return JSONResponse({
+            "success": False,
+            "error": str(e)
+        }, status_code=500)
+
+@app.post("/api/chat")
+async def api_chat(message: str = Form(...)):
+    """
+    API endpoint for chat interactions
+    """
+    try:
+        # Simple chat response without memory for API
+        prompt = f"You are a helpful AI tutor. Answer the following question accurately and concisely: {message}"
+        response = chat_model([HumanMessage(content=prompt)])
+        cleaned_response = clean_response(response.content)
+        
+        return JSONResponse({
+            "success": True,
+            "response": cleaned_response
+        })
+        
+    except Exception as e:
+        return JSONResponse({
+            "success": False,
+            "error": str(e)
+        }, status_code=500)
+
+@app.post("/api/process-document")
+async def api_process_document(file: UploadFile = File(...)):
+    """
+    API endpoint for document processing
+    """
+    try:
+        # Save uploaded file temporarily
+        file_extension = os.path.splitext(file.filename)[-1].lower()
+        temp_path = f"temp_{uuid.uuid4()}{file_extension}"
+        
+        with open(temp_path, "wb") as f:
+            f.write(await file.read())
+        
+        # Process document based on type
+        if file_extension == ".pdf":
+            content = extract_text_from_pdf(temp_path)
+        elif file_extension == ".docx":
+            content = extract_text_from_docx(temp_path)
+        elif file_extension == ".pptx":
+            content = extract_text_from_pptx(temp_path)
+        else:
+            # Try text file
+            encoding = detect_encoding(temp_path)
+            with open(temp_path, "r", encoding=encoding, errors="replace") as f:
+                content = f.read()
+        
+        # Clean up temp file
+        os.remove(temp_path)
+        
+        # Generate quiz
+        quiz = generate_quiz(content)
+        
+        return JSONResponse({
+            "success": True,
+            "content_preview": content[:500] + "..." if len(content) > 500 else content,
+            "quiz": quiz
+        })
+        
+    except Exception as e:
+        return JSONResponse({
+            "success": False,
+            "error": str(e)
+        }, status_code=500)
+
+@app.get("/api/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "healthy", "timestamp": time.time()}
+
+# Clear chat history function
+def clear_chat_history():
+    chat_memory.clear()
+    return [], None
+
+def tutor_ai_chatbot():
+    """Main Gradio interface for the Tutor AI Chatbot."""
+    with gr.Blocks() as gradio_app:
+        gr.Markdown("# 📚 AI Tutor - We.(POC)")
+        gr.Markdown("An interactive Personal AI Tutor chatbot to help with your learning needs.")
+
+        # Chatbot Tab
+        with gr.Tab("AI Chatbot"):
+            with gr.Row():
+                with gr.Column(scale=3):
+                    chatbot = gr.Chatbot(height=500, type="messages")
+                    
+                with gr.Column(scale=1):
+                    audio_playback = gr.Audio(label="Audio Response", type="filepath")
+            
+            # Move the input controls here to span full width
+            with gr.Row():
+                msg = gr.Textbox(
+                    label="Ask a question", 
+                    placeholder="Type your question here...",
+                    container=False  # Removes the default container styling
+                )
+                submit = gr.Button("Send")
+            
+            with gr.Row():
+                with gr.Column(scale=1):
+                    audio_input = gr.Audio(type="numpy", label="Record or Upload Audio")
+            
+            # Voice recording tips - ONLY in AI Chatbot tab
+            with gr.Accordion("🎤 Voice Recording Tips", open=False):
+                gr.Markdown("""
+                **For better speech recognition accuracy:**
+                - 🎙️ Speak clearly and at a moderate pace
+                - 🔇 Record in a quiet environment
+                - 📍 Keep the microphone close to your mouth (10-15 cm)
+                - 🎧 Use a good quality microphone if possible
+                - 📝 Review the transcribed text before sending
+                - 🔄 If transcription is poor, try recording again or type manually
+                """)
+            
+            # Clear chat history button
+            clear_btn = gr.Button("Clear Chat")
+
+            # Handle chat interaction
+            submit.click(
+                chat_with_groq,
+                inputs=[msg, chatbot],
+                outputs=[chatbot, msg, audio_playback]
+            )
+
+            # Clear chat history function
+            clear_btn.click(
+                lambda: [],  # Return empty list in message format
+                inputs=None, 
+                outputs=[chatbot]
+            )
+
+            # Also allow Enter key to submit
+            msg.submit(
+                chat_with_groq,
+                inputs=[msg, chatbot],
+                outputs=[chatbot, msg, audio_playback]
+            )
+
+            # Add some examples of questions students might ask
+            with gr.Accordion("Example Questions", open=False):
+                gr.Examples(
+                    examples=[
+                        "Can you explain the concept of RLHF AI?",
+                        "What are AI transformers?",
+                        "What is MoE AI?",
+                        "What's gate networks AI?",
+                        "I am making a switch, please generating baking recipe?"
+                    ],
+                    inputs=msg
+                )
+
+            # Connect audio input to transcription
+            audio_input.change(fn=transcribe_audio, inputs=audio_input, outputs=msg)
+
+        # Upload Notes & Generate Quiz Tab
+        with gr.Tab("Upload Notes & Generate Quiz"):
+            with gr.Row():
+                with gr.Column(scale=2):
+                    file_input = gr.File(label="Upload Lecture Notes (PDF, DOCX, PPTX)")
+                with gr.Column(scale=3):
+                    quiz_output = gr.Textbox(label="Generated Quiz", lines=10)
+
+            # Connect file input to document processing
+            file_input.change(process_document, inputs=file_input, outputs=quiz_output)
+
+        # Introduction Video Tab - Now with the working video
+        with gr.Tab("Introduction Video"):
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.Markdown("### Welcome to the Introduction Video")
+                    gr.Markdown("Music from Xu Mengyuan - China-O, musician Xu Mengyuan YUAN! | 徐梦圆 - China-O 音乐人徐梦圆YUAN!")
+                    # Use the local video file that's stored in your Space
+                    gr.Video("We_not_me_video.mp4", label="Introduction Video")
+
+        # Launch the application
+        gradio_app.launch(share=False)
+
+# Run both FastAPI and Gradio
+if __name__ == "__main__":
+    import threading
+    
+    # Start Gradio in a separate thread
+    gradio_thread = threading.Thread(target=tutor_ai_chatbot, daemon=True)
+    gradio_thread.start()
+    
+    # Start FastAPI
+    uvicorn.run(app, host="0.0.0.0", port=8000)