""" Modal Labs GPU Processing GPU-accelerated sentiment analysis and text processing """ import modal import asyncio import logging import os from typing import List, Dict, Any, Optional import json from datetime import datetime import numpy as np from dotenv import load_dotenv # Load environment variables load_dotenv() logger = logging.getLogger(__name__) # Modal app definition app = modal.App("product-feature-agent") # Define the Modal image with all required dependencies image = ( modal.Image.debian_slim() .pip_install([ "torch>=2.1.0", "transformers>=4.35.0", "numpy>=1.24.0", "scikit-learn>=1.3.0", "textblob>=0.17.1", "vaderSentiment>=3.3.2", "pandas>=2.1.0", "accelerate>=0.24.0", "python-dotenv>=1.0.0" ]) .run_commands([ "python -c \"import nltk; nltk.download('punkt'); nltk.download('punkt_tab'); nltk.download('vader_lexicon'); nltk.download('averaged_perceptron_tagger')\"", "python -m textblob.download_corpora", "python -c \"import textblob; textblob.TextBlob('test').tags\"" ]) ) # Shared volume for model caching model_volume = modal.Volume.from_name("feature-agent-models", create_if_missing=True) @app.function( image=image, gpu="T4", # Use T4 GPU for cost efficiency memory=4096, # 4GB RAM timeout=300, # 5 minute timeout volumes={"/models": model_volume}, min_containers=1 # Keep one instance warm for faster response ) def gpu_batch_sentiment_analysis(texts: List[str], batch_size: int = 32) -> List[Dict[str, Any]]: """ Perform GPU-accelerated sentiment analysis on a batch of texts Args: texts: List of text strings to analyze batch_size: Batch size for processing Returns: List of sentiment analysis results """ import torch from transformers.pipelines import pipeline from transformers import AutoTokenizer, AutoModelForSequenceClassification from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer from textblob import TextBlob import time start_time = time.time() logger.info(f"Starting GPU sentiment analysis for {len(texts)} texts") try: # Initialize VADER analyzer (CPU-based, fast for social media text) vader_analyzer = SentimentIntensityAnalyzer() # Load transformer model for more sophisticated analysis model_name = "cardiffnlp/twitter-roberta-base-sentiment-latest" cache_dir = "/models/sentiment" # Use GPU if available device = 0 if torch.cuda.is_available() else -1 # Initialize sentiment pipeline with caching sentiment_pipeline = pipeline( "sentiment-analysis", model=model_name, tokenizer=model_name, device=device, model_kwargs={"cache_dir": cache_dir}, max_length=512, truncation=True ) results = [] # Process texts in batches for i in range(0, len(texts), batch_size): batch_texts = texts[i:i + batch_size] batch_results = [] # GPU-accelerated transformer analysis transformer_results = sentiment_pipeline(batch_texts) # Process each text in the batch for j, text in enumerate(batch_texts): if not text or len(text.strip()) == 0: batch_results.append({ "text": text, "error": "Empty text", "processing_time": 0 }) continue text_start = time.time() # VADER sentiment (good for social media, slang, emojis) vader_scores = vader_analyzer.polarity_scores(text) # TextBlob sentiment (grammar-based) try: blob = TextBlob(text) textblob_sentiment = blob.sentiment except: textblob_sentiment = None # Transformer sentiment (deep learning) transformer_result = transformer_results[j] # type: ignore # Map transformer labels to consistent format transformer_label = transformer_result["label"].lower() # type: ignore if transformer_label in ["positive", "pos"]: transformer_sentiment = "positive" elif transformer_label in ["negative", "neg"]: transformer_sentiment = "negative" else: transformer_sentiment = "neutral" # Combine all sentiment scores combined_result = { "text": text[:100] + "..." if len(text) > 100 else text, # Truncate for storage "vader": { "compound": vader_scores["compound"], "positive": vader_scores["pos"], "negative": vader_scores["neg"], "neutral": vader_scores["neu"] }, "textblob": { "polarity": textblob_sentiment.polarity if textblob_sentiment else 0, # type: ignore "subjectivity": textblob_sentiment.subjectivity if textblob_sentiment else 0 # type: ignore }, "transformer": { "label": transformer_sentiment, "confidence": transformer_result["score"] # type: ignore }, "consensus": _calculate_consensus_sentiment( vader_scores["compound"], textblob_sentiment.polarity if textblob_sentiment else 0, # type: ignore transformer_sentiment, transformer_result["score"] # type: ignore ), "processing_time": time.time() - text_start } batch_results.append(combined_result) results.extend(batch_results) # Log progress processed = min(i + batch_size, len(texts)) logger.info(f"Processed {processed}/{len(texts)} texts") total_time = time.time() - start_time logger.info(f"GPU sentiment analysis completed in {total_time:.2f}s") # Add summary statistics summary = _calculate_batch_summary(results) return { "results": results, "summary": summary, "processing_stats": { "total_texts": len(texts), "total_time": total_time, "texts_per_second": len(texts) / total_time, "gpu_used": torch.cuda.is_available(), "model_used": model_name } } # type: ignore except Exception as e: logger.error(f"Error in GPU sentiment analysis: {str(e)}") return { "error": str(e), "processing_stats": { "total_texts": len(texts), "total_time": time.time() - start_time, "gpu_used": False } } # type: ignore def _calculate_consensus_sentiment(vader_compound: float, textblob_polarity: float, transformer_label: str, transformer_confidence: float) -> Dict[str, Any]: """Calculate consensus sentiment from multiple models""" # Convert transformer to numeric transformer_score = 0 if transformer_label == "positive": transformer_score = transformer_confidence elif transformer_label == "negative": transformer_score = -transformer_confidence # Weight the scores (transformer gets higher weight due to confidence) weights = { "vader": 0.3, "textblob": 0.2, "transformer": 0.5 } weighted_score = ( vader_compound * weights["vader"] + textblob_polarity * weights["textblob"] + transformer_score * weights["transformer"] ) # Classify final sentiment if weighted_score >= 0.1: consensus_label = "positive" elif weighted_score <= -0.1: consensus_label = "negative" else: consensus_label = "neutral" # Calculate confidence based on agreement scores = [vader_compound, textblob_polarity, transformer_score] agreement = 1.0 - (np.std(scores) / 2.0) # Higher agreement = lower std dev return { "label": consensus_label, "score": weighted_score, "confidence": max(0.0, min(1.0, agreement)) # type: ignore } def _calculate_batch_summary(results: List[Dict[str, Any]]) -> Dict[str, Any]: """Calculate summary statistics for batch results""" if not results: return {} valid_results = [r for r in results if "error" not in r] if not valid_results: return {"error": "No valid results"} # Count sentiment labels sentiment_counts = {"positive": 0, "negative": 0, "neutral": 0} total_confidence = 0 total_processing_time = 0 for result in valid_results: consensus = result.get("consensus", {}) label = consensus.get("label", "neutral") confidence = consensus.get("confidence", 0) sentiment_counts[label] += 1 total_confidence += confidence total_processing_time += result.get("processing_time", 0) total_valid = len(valid_results) return { "total_analyzed": total_valid, "sentiment_distribution": { "positive": sentiment_counts["positive"], "negative": sentiment_counts["negative"], "neutral": sentiment_counts["neutral"], "positive_pct": (sentiment_counts["positive"] / total_valid) * 100, "negative_pct": (sentiment_counts["negative"] / total_valid) * 100, "neutral_pct": (sentiment_counts["neutral"] / total_valid) * 100 }, "average_confidence": total_confidence / total_valid, "average_processing_time": total_processing_time / total_valid, "dominant_sentiment": max(sentiment_counts, key=sentiment_counts.get) # type: ignore } @app.function( image=image, gpu="T4", memory=2048, timeout=180 ) def gpu_keyword_extraction(texts: List[str], max_keywords: int = 10) -> Dict[str, Any]: """ Extract keywords using GPU-accelerated NLP models """ import torch from transformers.pipelines import pipeline from textblob import TextBlob from sklearn.feature_extraction.text import TfidfVectorizer import time start_time = time.time() logger.info(f"Starting GPU keyword extraction for {len(texts)} texts") try: # Combine all texts combined_text = " ".join(texts) # Use TextBlob for noun phrase extraction blob = TextBlob(combined_text) noun_phrases = list(blob.noun_phrases) # type: ignore # Use TF-IDF for important terms vectorizer = TfidfVectorizer( max_features=max_keywords * 2, ngram_range=(1, 3), stop_words="english" ) if texts: tfidf_matrix = vectorizer.fit_transform(texts) feature_names = vectorizer.get_feature_names_out() tfidf_scores = tfidf_matrix.sum(axis=0).A1 # type: ignore # Get top TF-IDF terms top_indices = tfidf_scores.argsort()[-max_keywords:][::-1] tfidf_keywords = [(feature_names[i], tfidf_scores[i]) for i in top_indices] else: tfidf_keywords = [] # Combine and rank keywords all_keywords = {} # Add noun phrases for phrase in noun_phrases: if len(phrase) > 3: # Filter short phrases all_keywords[phrase] = all_keywords.get(phrase, 0) + 1 # Add TF-IDF terms for term, score in tfidf_keywords: all_keywords[term] = all_keywords.get(term, 0) + score # Sort by importance sorted_keywords = sorted(all_keywords.items(), key=lambda x: x[1], reverse=True) result = { "keywords": sorted_keywords[:max_keywords], "total_texts": len(texts), "processing_time": time.time() - start_time, "method": "hybrid_tfidf_nlp" } logger.info(f"Keyword extraction completed in {result['processing_time']:.2f}s") return result except Exception as e: logger.error(f"Error in keyword extraction: {str(e)}") return {"error": str(e)} class GPUProcessor: """Client interface for Modal GPU processing""" def __init__(self): """Initialize GPU processor client""" self.app = app self.modal_available = False self.setup_modal_client() def setup_modal_client(self): """Setup Modal client with credentials and test connection""" logger.info("⚙️ Modal Labs client setup - Started") try: # Check for Modal token modal_token = os.getenv("MODAL_TOKEN") if not modal_token: error_msg = "No Modal token found in environment variables (MODAL_TOKEN)" logger.error(f"❌ Modal Labs API failed: {error_msg}") self.modal_available = False return logger.info(f"🔌 Attempting to connect to Modal Labs API with token ending in ...{modal_token[-4:]}") # Test Modal connection by trying to import and validate try: import modal # Try to create a simple app to test connection test_app = modal.App("connection-test") logger.info("✅ Modal Labs API connected successfully - client initialized") self.modal_available = True logger.info("⚙️ Modal Labs client setup - Completed") except Exception as modal_error: error_msg = f"Modal Labs connection test failed: {str(modal_error)}" logger.error(f"❌ Modal Labs API failed: {error_msg}") self.modal_available = False except Exception as e: error_msg = f"Modal Labs client setup failed: {str(e)}" logger.error(f"❌ Modal Labs API failed: {error_msg}") logger.error("⚙️ Modal Labs client setup - Failed") self.modal_available = False async def batch_sentiment_analysis(self, data_sources: List[Any]) -> Dict[str, Any]: """ Process multiple data sources with sentiment analysis (GPU if available, fallback to local) Args: data_sources: List of data from different collectors Returns: Comprehensive sentiment analysis """ try: # Extract texts from different data sources all_texts = [] source_mapping = {} for i, source in enumerate(data_sources): source_texts = self._extract_texts_from_source(source) # Track which texts come from which source start_idx = len(all_texts) all_texts.extend(source_texts) end_idx = len(all_texts) source_mapping[f"source_{i}"] = { "start": start_idx, "end": end_idx, "count": len(source_texts), "type": self._identify_source_type(source), "data": source # Store original data for fallback } if not all_texts: return {"error": "No texts found in data sources"} # Try Modal GPU processing if available if self.modal_available: try: logger.info(f"Sending {len(all_texts)} texts to Modal GPU processing") with app.run(): sentiment_results = gpu_batch_sentiment_analysis.remote(all_texts) # Reorganize results by source organized_results = self._organize_results_by_source( sentiment_results, source_mapping # type: ignore ) return organized_results except Exception as modal_error: logger.warning(f"Modal GPU processing failed: {str(modal_error)}") logger.info("Falling back to local sentiment analysis") # Fallback to local sentiment analysis logger.info(f"Processing {len(all_texts)} texts with local sentiment analysis") sentiment_results = await self._local_sentiment_analysis(all_texts) # Reorganize results by source organized_results = self._organize_results_by_source( sentiment_results, source_mapping ) return organized_results except Exception as e: logger.error(f"Error in batch sentiment analysis: {str(e)}") return {"error": str(e)} async def extract_keywords(self, texts: List[str], max_keywords: int = 20) -> Dict[str, Any]: """Extract keywords using GPU acceleration""" try: with app.run(): result = gpu_keyword_extraction.remote(texts, max_keywords) return result except Exception as e: logger.error(f"Error in keyword extraction: {str(e)}") return {"error": str(e)} async def _local_sentiment_analysis(self, texts: List[str]) -> Dict[str, Any]: """ Local sentiment analysis fallback using VADER and TextBlob """ try: from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer from textblob import TextBlob import time start_time = time.time() vader_analyzer = SentimentIntensityAnalyzer() results = [] for text in texts: if not text or len(text.strip()) == 0: results.append({ "text": text, "error": "Empty text", "processing_time": 0 }) continue text_start = time.time() # VADER sentiment vader_scores = vader_analyzer.polarity_scores(text) # TextBlob sentiment try: blob = TextBlob(text) textblob_sentiment = blob.sentiment except: textblob_sentiment = None # Simple consensus (weighted average) vader_compound = vader_scores["compound"] textblob_polarity = textblob_sentiment.polarity if textblob_sentiment else 0 # Calculate consensus weighted_score = (vader_compound * 0.6) + (textblob_polarity * 0.4) if weighted_score >= 0.1: consensus_label = "positive" elif weighted_score <= -0.1: consensus_label = "negative" else: consensus_label = "neutral" # Calculate confidence based on agreement agreement = 1.0 - abs(vader_compound - textblob_polarity) / 2.0 confidence = max(0.0, min(1.0, agreement)) result = { "text": text[:100] + "..." if len(text) > 100 else text, "vader": { "compound": vader_compound, "positive": vader_scores["pos"], "negative": vader_scores["neg"], "neutral": vader_scores["neu"] }, "textblob": { "polarity": textblob_polarity, "subjectivity": textblob_sentiment.subjectivity if textblob_sentiment else 0 }, "consensus": { "label": consensus_label, "score": weighted_score, "confidence": confidence }, "processing_time": time.time() - text_start } results.append(result) total_time = time.time() - start_time # Calculate summary summary = _calculate_batch_summary(results) return { "results": results, "summary": summary, "processing_stats": { "total_texts": len(texts), "total_time": total_time, "texts_per_second": len(texts) / total_time if total_time > 0 else 0, "gpu_used": False, "model_used": "local_vader_textblob" } } except Exception as e: logger.error(f"Error in local sentiment analysis: {str(e)}") return {"error": str(e)} def _extract_texts_from_source(self, source: Any) -> List[str]: """Extract text content from different data source formats""" texts = [] if isinstance(source, dict): # App Store reviews if "apps" in source: for app_name, app_data in source["apps"].items(): if "reviews" in app_data: for review in app_data["reviews"]: title = review.get("title", "") content = review.get("content", "") combined = f"{title} {content}".strip() if combined: texts.append(combined) # Reddit posts elif "posts" in source: for post in source["posts"]: title = post.get("title", "") selftext = post.get("selftext", "") combined = f"{title} {selftext}".strip() if combined: texts.append(combined) # News articles (check multiple possible structures) elif "articles" in source: for article in source["articles"]: title = article.get("title", "") description = article.get("description", "") combined = f"{title} {description}".strip() if combined: texts.append(combined) # News search results structure elif "search_results" in source: for search_term, results in source["search_results"].items(): if "articles" in results: for article in results["articles"]: title = article.get("title", "") description = article.get("description", "") combined = f"{title} {description}".strip() if combined: texts.append(combined) # Reddit query results structure elif "query_results" in source: for query, result in source["query_results"].items(): if "posts" in result: for post in result["posts"]: title = post.get("title", "") selftext = post.get("selftext", "") combined = f"{title} {selftext}".strip() if combined: texts.append(combined) return texts def _identify_source_type(self, source: Any) -> str: """Identify the type of data source""" if isinstance(source, dict): if "apps" in source: return "app_store" elif "posts" in source: return "reddit" elif "articles" in source: return "news" return "unknown" def _organize_results_by_source(self, sentiment_results: Dict[str, Any], source_mapping: Dict[str, Any]) -> Dict[str, Any]: """Organize sentiment results by original data source""" if "error" in sentiment_results: return sentiment_results results = sentiment_results.get("results", []) summary = sentiment_results.get("summary", {}) processing_stats = sentiment_results.get("processing_stats", {}) organized = { "by_source": {}, "overall_summary": summary, "processing_stats": processing_stats } # Split results back to sources for source_id, mapping in source_mapping.items(): start_idx = mapping["start"] end_idx = mapping["end"] source_results = results[start_idx:end_idx] # Calculate source-specific summary source_summary = _calculate_batch_summary(source_results) organized["by_source"][source_id] = { "type": mapping["type"], "count": mapping["count"], "results": source_results, "summary": source_summary } return organized # Example usage and testing functions async def test_gpu_processor(): """Test function for GPU processor""" processor = GPUProcessor() # Test data test_texts = [ "This product is amazing! I love using it every day.", "Terrible experience, would not recommend to anyone.", "It's okay, nothing special but does the job.", "Outstanding features and great customer service!", "Complete waste of money, very disappointed." ] # Test sentiment analysis print("Testing GPU sentiment analysis...") mock_sources = [ {"posts": [{"title": text, "selftext": ""} for text in test_texts[:3]]}, {"articles": [{"title": text, "description": ""} for text in test_texts[3:]]} ] sentiment_result = await processor.batch_sentiment_analysis(mock_sources) print(f"Sentiment analysis completed: {sentiment_result.get('processing_stats', {}).get('total_texts', 0)} texts processed") # Test keyword extraction print("Testing GPU keyword extraction...") keyword_result = await processor.extract_keywords(test_texts) print(f"Keyword extraction: {len(keyword_result.get('keywords', []))} keywords found") return sentiment_result, keyword_result if __name__ == "__main__": # Run test asyncio.run(test_gpu_processor())