Upload 2 files

app.py

@@ -17,7 +17,7 @@ import logging
 import json
 import numpy as np
 from dataclasses import dataclass, asdict
-import queue
+from queue import Queue
 import threading
 from collections import defaultdict
 
@@ -25,6 +25,20 @@ from collections import defaultdict
 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 = {
+    "Mistral-7B-Instruct": {
+        "path": "mistralai/Mistral-7B-Instruct-v0.1",
+        "description": "Efficient model for real-time analysis",
+        "memory_required": "16GB"
+    }
+}
+
 @dataclass
 class VCStyle:
     """Store VC's personal style preferences"""
@@ -44,280 +58,257 @@ class LiveCallContext:
     questions_asked: List[str]
     action_items: List[str]
 
-class RealTimeProcessor:
-    """Handle real-time audio processing and analysis"""
+class ModelManager:
+    """Handles model loading and resource management"""
     
-    def __init__(self, whisper_model, llm_pipeline):
-        self.whisper_model = whisper_model
-        self.llm_pipeline = llm_pipeline
-        self.audio_buffer = queue.Queue()
-        self.transcript_buffer = queue.Queue()
-        self.context = defaultdict(list)
-        
-    def process_audio_chunk(self, audio_chunk):
-        """Process incoming audio chunk"""
+    @staticmethod
+    @st.cache_resource
+    def load_whisper():
         try:
-            # Simulate real-time processing for Spaces
-            result = self.whisper_model.transcribe(audio_chunk)
-            return result["text"]
+            return whisper.load_model("base")
         except Exception as e:
-            logger.error(f"Error processing audio chunk: {e}")
+            logger.error(f"Failed to load Whisper model: {e}")
+            st.error("Failed to load speech recognition model. Please refresh the page.")
             return None
 
-    def generate_live_insights(self, transcript: str, vc_style: VCStyle) -> Dict[str, Any]:
-        """Generate real-time insights based on transcript"""
-        prompt = f"""Given this conversation transcript and VC's interests, provide real-time insights:
-        Transcript: {transcript}
-        VC Interests: {vc_style.key_interests}
-        
-        Provide insights in these areas:
-        1. Key Discussion Points
-        2. Potential Red Flags
-        3. Follow-up Questions
-        4. Market Insights
-        """
-        
+    @staticmethod
+    @st.cache_resource
+    def load_llm(model_name: str):
         try:
-            response = self.llm_pipeline(prompt, max_length=512)
-            return json.loads(response[0]['generated_text'])
+            config = MODEL_CONFIGS[model_name]
+            
+            # Optimized quantization config
+            bnb_config = BitsAndBytesConfig(
+                load_in_4bit=True,
+                bnb_4bit_quant_type="nf4",
+                bnb_4bit_compute_dtype=torch.float16,
+                bnb_4bit_use_double_quant=True,
+            )
+            
+            tokenizer = AutoTokenizer.from_pretrained(
+                config["path"],
+                token=st.secrets.get("HF_TOKEN"),
+                trust_remote_code=True
+            )
+            
+            model = AutoModelForCausalLM.from_pretrained(
+                config["path"],
+                token=st.secrets.get("HF_TOKEN"),
+                quantization_config=bnb_config,
+                device_map="auto",
+                torch_dtype=torch.float16,
+                low_cpu_mem_usage=True
+            )
+            
+            pipe = pipeline(
+                "text-generation",
+                model=model,
+                tokenizer=tokenizer,
+                max_new_tokens=512,  # Reduced for memory
+                temperature=0.7,
+                top_p=0.95,
+                repetition_penalty=1.15,
+                batch_size=BATCH_SIZE
+            )
+            
+            return pipe
+            
         except Exception as e:
-            logger.error(f"Error generating insights: {e}")
-            return {}
-
-    def update_context(self, transcript: str, insights: Dict[str, Any]):
-        """Update call context with new information"""
-        self.context['transcripts'].append(transcript)
-        self.context['insights'].extend(insights.get('key_points', []))
-        self.context['questions'].extend(insights.get('questions', []))
+            logger.error(f"Failed to load LLM {model_name}: {e}")
+            st.error("Failed to load language model. Please try again.")
+            return None
 
-class DynamicTemplate:
-    """Handle dynamic template management"""
-    
-    def __init__(self):
-        self.default_template = {
-            "product": {"problem": "", "solution": "", "other": ""},
-            "finances": {
-                "capital_raised": "", "cash_on_hand": "", 
-                "monthly_burn": "", "gross_margin": "", 
-                "deal_dynamics": ""
-            },
-            "market": {
-                "revenue": "", "yoy_growth": "", "tam": "",
-                "pricing": "", "acv_arpa": "", "churn": "",
-                "gtm": "", "competition": "", "other": ""
-            },
-            "concerns": {"risks": "", "mitigations": "", "other": ""},
-            "free_form": "",
-            "spv_actions": []
-        }
-        self.active_templates = {}
+class AudioProcessor:
+    """Handles audio processing with memory optimization"""
     
-    def create_custom_template(self, vc_style: VCStyle) -> Dict[str, Any]:
-        """Create a template based on VC's style"""
-        template = self.default_template.copy()
+    def __init__(self, model):
+        self.model = model
+        self.chunk_queue = Queue()
         
-        # Add custom sections
-        for section in vc_style.custom_sections:
-            template[section] = {"notes": "", "actions": []}
+    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"]
             
-        return template
-    
-    def update_template(self, template_id: str, section: str, content: str):
-        """Update template content in real-time"""
-        if template_id in self.active_templates:
-            try:
-                section_path = section.split('.')
-                target = self.active_templates[template_id]
-                for key in section_path[:-1]:
-                    target = target[key]
-                target[section_path[-1]] = content
-            except Exception as e:
-                logger.error(f"Error updating template: {e}")
+        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 StyleManager:
-    """Manage VC's personal styles and preferences"""
+class ContentAnalyzer:
+    """Handles text analysis with optimized prompts"""
     
-    def __init__(self):
-        self.styles = {}
+    def __init__(self, generator):
+        self.generator = generator
         
-    def create_style(self, vc_name: str, preferences: Dict[str, Any]) -> VCStyle:
-        """Create or update VC style"""
-        style = VCStyle(
-            name=vc_name,
-            note_format=preferences.get('note_format', {}),
-            key_interests=preferences.get('key_interests', []),
-            custom_sections=preferences.get('custom_sections', []),
-            insight_preferences=preferences.get('insight_preferences', {})
-        )
-        self.styles[vc_name] = style
-        return style
-    
-    def get_style(self, vc_name: str) -> Optional[VCStyle]:
-        """Retrieve VC's style"""
-        return self.styles.get(vc_name)
+    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
 
-class LiveCallManager:
-    """Manage live call processing and analysis"""
-    
-    def __init__(self, processor: RealTimeProcessor, template_manager: DynamicTemplate):
-        self.processor = processor
-        self.template_manager = template_manager
-        self.active_calls = {}
-        
-    def start_call(self, meeting_id: str, vc_style: VCStyle) -> str:
-        """Initialize a new call session"""
-        self.active_calls[meeting_id] = {
-            'context': LiveCallContext(
-                meeting_id=meeting_id,
-                participants=[],
-                topics=[],
-                key_points=[],
-                questions_asked=[],
-                action_items=[]
-            ),
-            'template': self.template_manager.create_custom_template(vc_style),
-            'style': vc_style
-        }
-        return meeting_id
-    
-    def process_call_segment(self, meeting_id: str, audio_chunk) -> Dict[str, Any]:
-        """Process a segment of the call"""
-        if meeting_id not in self.active_calls:
-            raise ValueError("Invalid meeting ID")
-            
-        # Process audio
-        transcript = self.processor.process_audio_chunk(audio_chunk)
-        if not transcript:
-            return {}
-            
-        # Generate insights
-        call_data = self.active_calls[meeting_id]
-        insights = self.processor.generate_live_insights(
-            transcript, 
-            call_data['style']
-        )
+    def _create_analysis_prompt(self, text: str, vc_style: VCStyle) -> str:
+        return f"""Analyze this startup pitch focusing on {', '.join(vc_style.key_interests)}:
         
-        # Update template
-        self.update_call_notes(meeting_id, transcript, insights)
+        {text}
         
-        return {
-            'transcript': transcript,
-            'insights': insights,
-            'template': call_data['template']
-        }
-    
-    def update_call_notes(self, meeting_id: str, transcript: str, insights: Dict[str, Any]):
-        """Update call notes with new information"""
-        call_data = self.active_calls[meeting_id]
-        template = call_data['template']
-        
-        # Update relevant sections based on insights
-        for key, value in insights.items():
-            if key in template:
-                if isinstance(template[key], dict):
-                    template[key].update(value)
-                else:
-                    template[key] = value
+        Provide structured insights for:
+        1. Key Points
+        2. Metrics
+        3. Risks
+        4. Questions"""
 
-def main():
-    st.set_page_config(page_title="VC Call Assistant", layout="wide")
-    
-    # Initialize managers
-    style_manager = StyleManager()
-    template_manager = DynamicTemplate()
+    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 ""
+
+class UIManager:
+    """Manages Streamlit UI with performance optimization"""
     
-    # Sidebar for VC profile
-    with st.sidebar:
-        st.header("VC Profile")
-        vc_name = st.text_input("VC Name")
-        
-        # Style preferences
-        st.subheader("Style Preferences")
-        note_format = st.multiselect(
-            "Preferred Note Format",
-            ["Bullet Points", "Paragraphs", "Q&A Format"],
-            default=["Bullet Points"]
+    @staticmethod
+    def setup_page():
+        st.set_page_config(
+            page_title="VC Call Assistant",
+            page_icon="🎙️",
+            layout="wide",
+            initial_sidebar_state="expanded"
         )
-        
-        key_interests = st.multiselect(
-            "Key Interest Areas",
-            ["Technical", "Financial", "Market", "Team", "Product"],
-            default=["Financial", "Market"]
-        )
-        
-        custom_sections = st.multiselect(
-            "Custom Sections",
-            ["Competition Deep Dive", "Technical Assessment", "Team Background"],
-            default=[]
-        )
-    
-    # Main content area
-    st.title("VC Call Assistant")
     
-    # Create tabs for different modes
-    tab1, tab2 = st.tabs(["Live Call", "Batch Processing"])
+    @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."
+        )
     
-    with tab1:
-        st.header("Live Call Analysis")
+    @staticmethod
+    def show_progress(text: str) -> Any:
+        return st.progress(0, text=text)
+
+def main():
+    try:
+        # Initialize UI
+        UIManager.setup_page()
         
-        # Initialize call managers if VC profile exists
-        if vc_name:
-            # Create/update VC style
-            vc_style = style_manager.create_style(
-                vc_name,
-                {
-                    'note_format': note_format,
-                    'key_interests': key_interests,
-                    'custom_sections': custom_sections,
-                    'insight_preferences': {'technical': 0.8, 'financial': 0.9}
-                }
+        # Sidebar
+        with st.sidebar:
+            st.title("VC Assistant Settings")
+            model_name = "Mistral-7B-Instruct"  # Fixed for stability
+            
+            st.info(f"""Using {model_name}
+            Memory Usage: {MODEL_CONFIGS[model_name]['memory_required']}
+            Description: {MODEL_CONFIGS[model_name]['description']}""")
+            
+            # VC Profile
+            vc_name = st.text_input("Your Name")
+            note_style = st.selectbox(
+                "Note Style",
+                ["Bullet Points", "Paragraphs", "Q&A"]
             )
             
-            # Audio input (simulated for Spaces)
-            uploaded_file = st.file_uploader(
-                "Upload audio segment", 
-                type=['wav', 'mp3', 'm4a']
+            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 uploaded_file:
-                # Process audio in chunks (simulated)
-                with st.spinner("Processing audio..."):
-                    # Initialize processors
-                    whisper_model = whisper.load_model("base")
-                    llm = load_llm("Mixtral-8x7B-Instruct")  # Function from previous code
-                    processor = RealTimeProcessor(whisper_model, llm)
-                    live_call_manager = LiveCallManager(processor, template_manager)
-                    
-                    # Start call session
-                    meeting_id = live_call_manager.start_call(
-                        str(datetime.now()), 
-                        vc_style
-                    )
-                    
-                    # Process call
-                    results = live_call_manager.process_call_segment(
-                        meeting_id,
-                        uploaded_file
-                    )
-                    
+            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("Live Transcript")
-                        st.write(results.get('transcript', ''))
-                        
-                        st.subheader("Real-time Insights")
-                        st.write(results.get('insights', {}))
+                        st.subheader("📝 Transcript")
+                        st.write(transcription)
                     
                     with col2:
-                        st.subheader("Smart Notes")
-                        st.write(results.get('template', {}))
-        else:
-            st.warning("Please enter VC profile information in the sidebar")
-    
-    with tab2:
-        st.header("Batch Processing")
-        # Previous batch processing code here
+                        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()

requirements.txt

@@ -11,4 +11,3 @@ sentencepiece==0.1.99
 huggingface-hub==0.19.4
 python-dotenv==1.0.0
 dataclasses-json==0.5.7
-queue==1.0.2