Upload 2 files

app.py

@@ -0,0 +1,323 @@
+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
+import queue
+import threading
+from collections import defaultdict
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+@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 RealTimeProcessor:
+    """Handle real-time audio processing and analysis"""
+    
+    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"""
+        try:
+            # Simulate real-time processing for Spaces
+            result = self.whisper_model.transcribe(audio_chunk)
+            return result["text"]
+        except Exception as e:
+            logger.error(f"Error processing audio chunk: {e}")
+            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
+        """
+        
+        try:
+            response = self.llm_pipeline(prompt, max_length=512)
+            return json.loads(response[0]['generated_text'])
+        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', []))
+
+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 = {}
+    
+    def create_custom_template(self, vc_style: VCStyle) -> Dict[str, Any]:
+        """Create a template based on VC's style"""
+        template = self.default_template.copy()
+        
+        # Add custom sections
+        for section in vc_style.custom_sections:
+            template[section] = {"notes": "", "actions": []}
+            
+        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}")
+
+class StyleManager:
+    """Manage VC's personal styles and preferences"""
+    
+    def __init__(self):
+        self.styles = {}
+        
+    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)
+
+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']
+        )
+        
+        # Update template
+        self.update_call_notes(meeting_id, transcript, insights)
+        
+        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
+
+def main():
+    st.set_page_config(page_title="VC Call Assistant", layout="wide")
+    
+    # Initialize managers
+    style_manager = StyleManager()
+    template_manager = DynamicTemplate()
+    
+    # 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"]
+        )
+        
+        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"])
+    
+    with tab1:
+        st.header("Live Call Analysis")
+        
+        # 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}
+                }
+            )
+            
+            # Audio input (simulated for Spaces)
+            uploaded_file = st.file_uploader(
+                "Upload audio segment", 
+                type=['wav', 'mp3', 'm4a']
+            )
+            
+            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
+                    )
+                    
+                    # 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', {}))
+                    
+                    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
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,14 @@
+streamlit==1.24.0
+whisper-openai==1.0.0
+pandas==1.5.3
+numpy==1.23.5
+torch==2.0.1
+transformers==4.35.2
+accelerate==0.24.1
+bitsandbytes==0.41.1
+scipy==1.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