Upload app.py

app.py

@@ -1,6 +1,36 @@
-# Only showing the modified sections for brevity
+import streamlit as st
+import whisper
+import pandas as pd
+from datetime import datetime
+import tempfile
+import os
+import torch
+from transformers import (
+    AutoModelForCausalLM, 
+    AutoTokenizer, 
+    pipeline,
+    BitsAndBytesConfig
+)
+import gc
+from typing import Optional, Dict, Any, List
+import logging
+import json
+import numpy as np
+from dataclasses import dataclass, asdict
+from queue import Queue
+import threading
+from collections import defaultdict
 
-# Update MODEL_CONFIGS to use open models
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Constants for memory optimization
+CHUNK_SIZE = 30  # seconds
+MAX_AUDIO_LENGTH = 600  # seconds (10 minutes)
+BATCH_SIZE = 8
+
+# Model configurations with memory optimization
 MODEL_CONFIGS = {
     "FLAN-T5-Large": {
         "path": "google/flan-t5-large",
@@ -14,6 +44,25 @@ MODEL_CONFIGS = {
     }
 }
 
+@dataclass
+class VCStyle:
+    """Store VC's personal style preferences"""
+    name: str
+    note_format: Dict[str, Any]
+    key_interests: List[str]
+    custom_sections: List[str]
+    insight_preferences: Dict[str, float]
+
+@dataclass
+class LiveCallContext:
+    """Store context for live calls"""
+    meeting_id: str
+    participants: List[str]
+    topics: List[str]
+    key_points: List[str]
+    questions_asked: List[str]
+    action_items: List[str]
+
 class ModelManager:
     """Handles model loading and resource management"""
     
@@ -72,6 +121,122 @@ class ModelManager:
             st.error("Failed to load language model. Please try again.")
             return None
 
+class AudioProcessor:
+    """Handles audio processing with memory optimization"""
+    
+    def __init__(self, model):
+        self.model = model
+        self.chunk_queue = Queue()
+        
+    def process_audio_chunk(self, audio_chunk) -> Optional[str]:
+        try:
+            # Clear GPU memory before processing
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+                
+            result = self.model.transcribe(
+                audio_chunk,
+                language="en",
+                task="transcribe",
+                fp16=True  # Use half precision
+            )
+            return result["text"]
+            
+        except Exception as e:
+            logger.error(f"Error processing audio chunk: {e}")
+            return None
+        finally:
+            # Cleanup
+            gc.collect()
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+
+class ContentAnalyzer:
+    """Handles text analysis with optimized prompts"""
+    
+    def __init__(self, generator):
+        self.generator = generator
+        
+    def analyze_text(self, text: str, vc_style: VCStyle) -> Optional[Dict[str, Any]]:
+        try:
+            prompt = self._create_analysis_prompt(text, vc_style)
+            response = self._generate_response(prompt, max_length=512)
+            return self._parse_response(response)
+        except Exception as e:
+            logger.error(f"Analysis error: {e}")
+            return None
+
+    def _create_analysis_prompt(self, text: str, vc_style: VCStyle) -> str:
+        return f"""Analyze this startup pitch focusing on {', '.join(vc_style.key_interests)}:
+        
+        {text}
+        
+        Provide structured insights for:
+        1. Key Points
+        2. Metrics
+        3. Risks
+        4. Questions"""
+
+    def _generate_response(self, prompt: str, max_length: int) -> str:
+        try:
+            response = self.generator(
+                prompt,
+                max_new_tokens=max_length,
+                temperature=0.7,
+                top_p=0.95,
+                repetition_penalty=1.15
+            )
+            return response[0]['generated_text']
+        except Exception as e:
+            logger.error(f"Generation error: {e}")
+            return ""
+            
+    def _parse_response(self, response: str) -> Dict[str, Any]:
+        try:
+            # Simple parsing of the response into sections
+            sections = response.split('\n\n')
+            parsed_response = {}
+            current_section = "general"
+            
+            for section in sections:
+                if section.strip().endswith(':'):
+                    current_section = section.strip()[:-1].lower()
+                    parsed_response[current_section] = []
+                else:
+                    if current_section in parsed_response:
+                        parsed_response[current_section].append(section.strip())
+                    else:
+                        parsed_response[current_section] = [section.strip()]
+                        
+            return parsed_response
+        except Exception as e:
+            logger.error(f"Parsing error: {e}")
+            return {"error": "Failed to parse response"}
+
+class UIManager:
+    """Manages Streamlit UI with performance optimization"""
+    
+    @staticmethod
+    def setup_page():
+        st.set_page_config(
+            page_title="VC Call Assistant",
+            page_icon="🎙️",
+            layout="wide",
+            initial_sidebar_state="expanded"
+        )
+    
+    @staticmethod
+    def show_file_uploader() -> Optional[Any]:
+        return st.file_uploader(
+            "Upload Audio (Max 10 minutes)",
+            type=['wav', 'mp3', 'm4a'],
+            help="Supports WAV, MP3, M4A formats. Maximum duration: 10 minutes."
+        )
+    
+    @staticmethod
+    def show_progress(text: str) -> Any:
+        return st.progress(0, text=text)
+
 def main():
     try:
         # Initialize UI
@@ -91,4 +256,89 @@ def main():
             Memory Usage: {MODEL_CONFIGS[model_name]['memory_required']}
             Description: {MODEL_CONFIGS[model_name]['description']}""")
             
-            # Rest of the sidebar code remains the same
+            # VC Profile
+            vc_name = st.text_input("Your Name")
+            note_style = st.selectbox(
+                "Note Style",
+                ["Bullet Points", "Paragraphs", "Q&A"]
+            )
+            
+            interests = st.multiselect(
+                "Focus Areas",
+                ["Product", "Market", "Team", "Financials", "Technology"],
+                default=["Product", "Market"]
+            )
+
+        # Main content
+        st.title("🎙️ VC Call Assistant")
+        
+        if not vc_name:
+            st.warning("Please enter your name in the sidebar.")
+            return
+
+        # Initialize processors
+        with st.spinner("Loading models..."):
+            whisper_model = ModelManager.load_whisper()
+            llm = ModelManager.load_llm(model_name)
+            
+            if not whisper_model or not llm:
+                st.error("Failed to initialize models. Please refresh the page.")
+                return
+            
+            audio_processor = AudioProcessor(whisper_model)
+            analyzer = ContentAnalyzer(llm)
+
+        # File upload
+        audio_file = UIManager.show_file_uploader()
+        
+        if audio_file:
+            # Process audio
+            with st.spinner("Processing audio..."):
+                transcription = audio_processor.process_audio_chunk(audio_file)
+                
+                if transcription:
+                    # Display results
+                    col1, col2 = st.columns(2)
+                    
+                    with col1:
+                        st.subheader("📝 Transcript")
+                        st.write(transcription)
+                    
+                    with col2:
+                        st.subheader("🔍 Analysis")
+                        vc_style = VCStyle(
+                            name=vc_name,
+                            note_format={"style": note_style},
+                            key_interests=interests,
+                            custom_sections=[],
+                            insight_preferences={}
+                        )
+                        
+                        analysis = analyzer.analyze_text(transcription, vc_style)
+                        if analysis:
+                            st.write(analysis)
+                            
+                            # Export option
+                            st.download_button(
+                                "📥 Export Analysis",
+                                data=json.dumps({
+                                    "timestamp": datetime.now().isoformat(),
+                                    "transcription": transcription,
+                                    "analysis": analysis
+                                }, indent=2),
+                                file_name=f"vc_analysis_{datetime.now():%Y%m%d_%H%M%S}.json",
+                                mime="application/json"
+                            )
+
+    except Exception as e:
+        logger.error(f"Application error: {e}")
+        st.error("An unexpected error occurred. Please refresh the page.")
+        
+    finally:
+        # Cleanup
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+
+if __name__ == "__main__":
+    main()