Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| from faster_whisper import WhisperModel | |
| import numpy as np | |
| import os | |
| import statistics | |
| from transformers import pipeline | |
| from textblob import TextBlob | |
| import torch | |
| import time | |
| class WebAITranscriber: | |
| def __init__(self): | |
| # Check if CUDA is available | |
| self.device = "cuda" if torch.cuda.is_available() else "cpu" | |
| self.compute_type = "float16" if torch.cuda.is_available() else "int8" | |
| # Initialize Whisper Model with optimized settings | |
| print(f"Initializing Whisper Model on {self.device}...") | |
| self.model = WhisperModel( | |
| "base", # Fixed: Directly specify the model size as a string | |
| device=self.device, | |
| compute_type=self.compute_type, | |
| cpu_threads=min(os.cpu_count(), 4), # Optimized thread count | |
| download_root=None, # Use default cache directory | |
| local_files_only=False # Allow downloading if needed | |
| ) | |
| # Optimize model settings | |
| self.model_settings = { | |
| 'beam_size': 1, # Reduced beam size for speed | |
| 'best_of': 1, # Take first result | |
| 'temperature': 0, # Reduce randomness | |
| 'compression_ratio_threshold': 2.4, | |
| 'condition_on_previous_text': True, | |
| 'no_speech_threshold': 0.6, | |
| 'initial_prompt': None | |
| } | |
| # AI Detection Markers (optimized for speed) | |
| self.ai_markers = { | |
| 'formal_phrases': [ | |
| 'moreover', 'furthermore', 'consequently', | |
| 'in conclusion', 'it is worth noting' | |
| ], | |
| 'ai_disclaimers': [ | |
| 'as an ai', 'i want to be clear', | |
| 'it is important to note' | |
| ] | |
| } | |
| # Initialize AI Detector with optimized settings | |
| print("Initializing AI Detection...") | |
| self.ai_detector = pipeline('text-classification', | |
| model='roberta-base-openai-detector', | |
| device=self.device) | |
| # Optimized buffer settings | |
| self.min_analysis_words = 10 # Minimum words before analysis | |
| self.max_buffer_size = 1000 # Maximum buffer size in words | |
| self.analysis_interval = 3 # Minimum seconds between analyses | |
| self.last_analysis_time = time.time() | |
| self.transcript_buffer = [] | |
| def process_realtime_audio(self, audio, state): | |
| """Process real-time audio with optimized settings""" | |
| if audio is None: | |
| return "", state | |
| try: | |
| # Initialize state if needed | |
| if state is None: | |
| state = {"full_transcript": "", "buffer": [], "pending_analysis": False} | |
| # Process audio in smaller chunks for real-time performance | |
| segments, _ = self.model.transcribe( | |
| audio[1], | |
| language="en", # Specify language for faster processing | |
| vad_filter=True, # Use Voice Activity Detection | |
| **self.model_settings | |
| ) | |
| # Process segments | |
| current_transcript = "" | |
| for segment in segments: | |
| current_transcript += segment.text + " " | |
| if not current_transcript.strip(): | |
| return state["full_transcript"], state | |
| # Update state | |
| state["full_transcript"] += " " + current_transcript | |
| state["buffer"].append(current_transcript) | |
| # Check if we should perform analysis | |
| current_time = time.time() | |
| buffer_text = " ".join(state["buffer"]) | |
| word_count = len(buffer_text.split()) | |
| if (word_count >= self.min_analysis_words and | |
| (current_time - self.last_analysis_time) >= self.analysis_interval): | |
| # Perform AI analysis | |
| if len(buffer_text.strip()) > 0: | |
| classification, probability, confidence = self.analyze_ai_content(buffer_text) | |
| analysis_result = f"\n\n---AI Analysis---\nClassification: {classification}\nProbability: {probability:.2f}\nConfidence: {confidence}\n---\n" | |
| state["full_transcript"] += analysis_result | |
| state["buffer"] = [] | |
| self.last_analysis_time = current_time | |
| # Trim buffer if it gets too large | |
| if word_count > self.max_buffer_size: | |
| words = state["full_transcript"].split() | |
| state["full_transcript"] = " ".join(words[-self.max_buffer_size:]) | |
| return state["full_transcript"], state | |
| except Exception as e: | |
| return f"Error processing audio: {str(e)}", state | |
| def analyze_ai_content(self, text): | |
| """Optimized AI content analysis""" | |
| if not text or len(text.split()) < self.min_analysis_words: | |
| return "Insufficient text", 0.0, "None" | |
| try: | |
| # Parallel processing of different analysis methods | |
| roberta_result = self.ai_detector(text[:512])[0] # Limit text length for speed | |
| ai_prob = roberta_result['score'] | |
| # Quick linguistic analysis | |
| linguistic_score = self.analyze_linguistic_patterns(text) | |
| # Calculate final score | |
| final_score = (ai_prob + linguistic_score) / 2 | |
| # Fast classification | |
| if final_score > 0.7: | |
| return "AI Generated", final_score, "High" | |
| elif final_score > 0.5: | |
| return "Likely AI", final_score, "Medium" | |
| elif final_score > 0.3: | |
| return "Possibly AI", final_score, "Low" | |
| return "Likely Human", final_score, "High" | |
| except Exception as e: | |
| print(f"Analysis error: {e}") | |
| return "Analysis Error", 0.0, "None" | |
| def analyze_linguistic_patterns(self, text): | |
| """Optimized linguistic pattern analysis""" | |
| text_lower = text.lower() | |
| ai_phrase_count = sum(1 for category in self.ai_markers.values() | |
| for phrase in category if phrase in text_lower) | |
| unique_words = len(set(text.split())) | |
| total_words = len(text.split()) | |
| return min((ai_phrase_count * 0.3) + (1 - (unique_words / total_words)) * 0.4, 1.0) | |
| def create_gradio_interface(): | |
| transcriber = WebAITranscriber() | |
| # Create the interface | |
| with gr.Blocks(title="Real-time AI Speech Analyzer") as interface: | |
| gr.Markdown(""" | |
| # Real-time AI Speech Analyzer | |
| This app uses Faster Whisper for real-time speech recognition and AI detection. | |
| """) | |
| with gr.Tab("Real-time Analysis"): | |
| with gr.Row(): | |
| with gr.Column(): | |
| audio_input = gr.Audio( | |
| sources=["microphone"], # Updated from 'source' to 'sources' | |
| streaming=True, | |
| type="numpy", | |
| label="🎤 Speak into your microphone" | |
| ) | |
| gr.Markdown(""" | |
| ### Tips for best performance: | |
| - Speak clearly and at a moderate pace | |
| - Minimize background noise | |
| - Wait a few seconds for initial processing | |
| """) | |
| with gr.Column(): | |
| realtime_output = gr.Textbox( | |
| label="Real-time Transcript and Analysis", | |
| lines=15, | |
| max_lines=30 | |
| ) | |
| audio_input.stream( | |
| transcriber.process_realtime_audio, | |
| inputs=[audio_input], | |
| outputs=[realtime_output], | |
| show_progress=False | |
| ) | |
| gr.Markdown(""" | |
| ### Technical Details: | |
| - Using Faster Whisper for optimized speech recognition | |
| - Real-time AI content analysis | |
| - Automatic voice activity detection | |
| - Optimized for low-latency processing | |
| """) | |
| return interface | |
| # Launch the app | |
| if __name__ == "__main__": | |
| interface = create_gradio_interface() | |
| interface.launch() |