CHKIM79/scalable-ai-agent-system

1"""2Natural Language Processing Module3Implements sentiment analysis, intent recognition, entity extraction, and text generation4"""5import asyncio6import logging7import re8from typing import Dict, List, Any, Optional, Tuple, Union9from dataclasses import dataclass, field10from enum import Enum11import openai12import numpy as np13from transformers import pipeline, AutoTokenizer, AutoModel14import torch15from sentence_transformers import SentenceTransformer16 17 18class SentimentLabel(Enum):19    POSITIVE = "positive"20    NEGATIVE = "negative"21    NEUTRAL = "neutral"22    MIXED = "mixed"23 24 25class IntentCategory(Enum):26    QUESTION = "question"27    REQUEST = "request"28    COMPLAINT = "complaint"29    COMPLIMENT = "compliment"30    INFORMATION_SEEKING = "information_seeking"31    TASK_EXECUTION = "task_execution"32    CONVERSATION = "conversation"33    UNKNOWN = "unknown"34 35 36@dataclass37class SentimentResult:38    label: SentimentLabel39    confidence: float40    emotional_intensity: float41    emotional_categories: Dict[str, float] = field(default_factory=dict)42 43 44@dataclass45class IntentResult:46    category: IntentCategory47    confidence: float48    subcategory: Optional[str] = None49    urgency_level: float = 0.050    complexity_level: float = 0.051 52 53@dataclass54class Entity:55    text: str56    label: str57    start_pos: int58    end_pos: int59    confidence: float60    metadata: Dict[str, Any] = field(default_factory=dict)61 62 63@dataclass64class NLPResult:65    original_text: str66    sentiment: SentimentResult67    intent: IntentResult68    entities: List[Entity]69    key_phrases: List[str]70    language: str = "en"71    processing_time: float = 0.072 73 74class SentimentAnalyzer:75    """Advanced sentiment analysis with emotional intelligence"""76    77    def __init__(self):78        self.sentiment_pipeline = pipeline(79            "sentiment-analysis",80            model="cardiffnlp/twitter-roberta-base-sentiment-latest",81            return_all_scores=True82        )83        self.emotion_pipeline = pipeline(84            "text-classification",85            model="j-hartmann/emotion-english-distilroberta-base",86            return_all_scores=True87        )88    89    async def analyze(self, text: str) -> SentimentResult:90        """Perform comprehensive sentiment analysis"""91        # Basic sentiment92        sentiment_scores = self.sentiment_pipeline(text)[0]93        94        # Map model labels to our enum95        label_mapping = {96            "LABEL_0": SentimentLabel.NEGATIVE,97            "LABEL_1": SentimentLabel.NEUTRAL,98            "LABEL_2": SentimentLabel.POSITIVE99        }100        101        # Find highest confidence sentiment102        best_sentiment = max(sentiment_scores, key=lambda x: x['score'])103        sentiment_label = label_mapping.get(best_sentiment['label'], SentimentLabel.NEUTRAL)104        105        # Emotional analysis106        emotion_scores = self.emotion_pipeline(text)[0]107        emotional_categories = {item['label']: item['score'] for item in emotion_scores}108        109        # Calculate emotional intensity110        emotional_intensity = max(emotional_categories.values())111        112        # Check for mixed emotions113        top_emotions = sorted(emotion_scores, key=lambda x: x['score'], reverse=True)[:2]114        if abs(top_emotions[0]['score'] - top_emotions[1]['score']) < 0.1:115            sentiment_label = SentimentLabel.MIXED116        117        return SentimentResult(118            label=sentiment_label,119            confidence=best_sentiment['score'],120            emotional_intensity=emotional_intensity,121            emotional_categories=emotional_categories122        )123 124 125class IntentClassifier:126    """Intent recognition and classification"""127    128    def __init__(self):129        self.intent_patterns = {130            IntentCategory.QUESTION: [131                r'\b(what|how|why|when|where|who|which)\b',132                r'\?',133                r'\b(explain|tell me|help me understand)\b'134            ],135            IntentCategory.REQUEST: [136                r'\b(please|can you|could you|would you)\b',137                r'\b(help|assist|support)\b',138                r'\b(need|want|require)\b'139            ],140            IntentCategory.COMPLAINT: [141                r'\b(problem|issue|error|bug|wrong|broken)\b',142                r'\b(not working|doesn\'t work|failed)\b',143                r'\b(frustrated|annoyed|upset)\b'144            ],145            IntentCategory.COMPLIMENT: [146                r'\b(thank|thanks|appreciate|grateful)\b',147                r'\b(great|excellent|amazing|wonderful)\b',148                r'\b(good job|well done)\b'149            ],150            IntentCategory.TASK_EXECUTION: [151                r'\b(do|execute|run|perform|start|begin)\b',152                r'\b(create|make|build|generate)\b',153                r'\b(calculate|compute|analyze)\b'154            ]155        }156        157        self.urgency_keywords = [158            "urgent", "asap", "immediately", "emergency", "critical", "now"159        ]160        161        self.complexity_keywords = [162            "complex", "complicated", "multiple", "several", "various", "detailed"163        ]164    165    async def classify(self, text: str) -> IntentResult:166        """Classify intent of the text"""167        text_lower = text.lower()168        169        # Pattern matching for intent categories170        intent_scores = {}171        for category, patterns in self.intent_patterns.items():172            score = 0173            for pattern in patterns:174                matches = len(re.findall(pattern, text_lower))175                score += matches176            intent_scores[category] = score177        178        # Determine primary intent179        if max(intent_scores.values()) == 0:180            primary_intent = IntentCategory.UNKNOWN181            confidence = 0.3182        else:183            primary_intent = max(intent_scores, key=intent_scores.get)184            total_matches = sum(intent_scores.values())185            confidence = intent_scores[primary_intent] / total_matches if total_matches > 0 else 0.5186        187        # Calculate urgency level188        urgency_level = sum(1 for keyword in self.urgency_keywords if keyword in text_lower) / len(self.urgency_keywords)189        190        # Calculate complexity level191        complexity_level = sum(1 for keyword in self.complexity_keywords if keyword in text_lower) / len(self.complexity_keywords)192        complexity_level += len(text.split()) / 100  # Longer text = more complex193        complexity_level = min(complexity_level, 1.0)194        195        return IntentResult(196            category=primary_intent,197            confidence=min(confidence, 1.0),198            urgency_level=urgency_level,199            complexity_level=complexity_level200        )201 202 203class EntityExtractor:204    """Named Entity Recognition and extraction"""205    206    def __init__(self):207        self.ner_pipeline = pipeline(208            "ner",209            model="dbmdz/bert-large-cased-finetuned-conll03-english",210            aggregation_strategy="simple"211        )212        213        # Custom entity patterns214        self.custom_patterns = {215            "EMAIL": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',216            "PHONE": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',217            "URL": r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+',218            "MONEY": r'\$\d+(?:,\d{3})*(?:\.\d{2})?',219            "DATE_TIME": r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b|\b\d{4}-\d{2}-\d{2}\b'220        }221    222    async def extract(self, text: str) -> List[Entity]:223        """Extract entities from text"""224        entities = []225        226        # Standard NER227        ner_results = self.ner_pipeline(text)228        for result in ner_results:229            entities.append(Entity(230                text=result['word'],231                label=result['entity_group'],232                start_pos=result['start'],233                end_pos=result['end'],234                confidence=result['score']235            ))236        237        # Custom pattern matching238        for label, pattern in self.custom_patterns.items():239            matches = re.finditer(pattern, text)240            for match in matches:241                entities.append(Entity(242                    text=match.group(),243                    label=label,244                    start_pos=match.start(),245                    end_pos=match.end(),246                    confidence=0.9247                ))248        249        # Remove duplicates and overlaps250        entities = self._remove_overlapping_entities(entities)251        252        return entities253    254    def _remove_overlapping_entities(self, entities: List[Entity]) -> List[Entity]:255        """Remove overlapping entities, keeping the one with higher confidence"""256        entities.sort(key=lambda x: x.start_pos)257        filtered_entities = []258        259        for entity in entities:260            overlap = False261            for existing in filtered_entities:262                if (entity.start_pos < existing.end_pos and 263                    entity.end_pos > existing.start_pos):264                    if entity.confidence <= existing.confidence:265                        overlap = True266                        break267                    else:268                        # Remove the existing entity with lower confidence269                        filtered_entities.remove(existing)270                        break271            272            if not overlap:273                filtered_entities.append(entity)274        275        return filtered_entities276 277 278class KeyPhraseExtractor:279    """Extract key phrases and topics from text"""280    281    def __init__(self):282        self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')283    284    async def extract(self, text: str, max_phrases: int = 10) -> List[str]:285        """Extract key phrases using various techniques"""286        # Simple n-gram extraction287        words = text.lower().split()288        289        # Remove stop words290        stop_words = {291            'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for',292            'of', 'with', 'by', 'is', 'are', 'was', 'were', 'be', 'been', 'have',293            'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should'294        }295        296        filtered_words = [word for word in words if word not in stop_words and len(word) > 2]297        298        # Extract bigrams and trigrams299        phrases = []300        301        # Unigrams (important single words)302        word_freq = {}303        for word in filtered_words:304            word_freq[word] = word_freq.get(word, 0) + 1305        306        # Get top unigrams307        top_unigrams = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:max_phrases//2]308        phrases.extend([word for word, _ in top_unigrams])309        310        # Bigrams311        for i in range(len(filtered_words) - 1):312            bigram = f"{filtered_words[i]} {filtered_words[i+1]}"313            phrases.append(bigram)314        315        # Trigrams316        for i in range(len(filtered_words) - 2):317            trigram = f"{filtered_words[i]} {filtered_words[i+1]} {filtered_words[i+2]}"318            phrases.append(trigram)319        320        # Remove duplicates and limit321        unique_phrases = list(set(phrases))[:max_phrases]322        323        return unique_phrases324 325 326class TextGenerator:327    """Advanced text generation with transformer models"""328    329    def __init__(self, llm_client=None):330        self.llm_client = llm_client or openai.AsyncOpenAI()331    332    async def generate_response(self, 333                              prompt: str, 334                              style: str = "professional",335                              max_tokens: int = 500,336                              temperature: float = 0.7) -> str:337        """Generate contextual response"""338        339        style_prompts = {340            "professional": "Respond in a professional, clear, and helpful manner.",341            "casual": "Respond in a casual, friendly, and conversational tone.",342            "empathetic": "Respond with empathy, understanding, and emotional support.",343            "technical": "Respond with technical accuracy and detailed explanations.",344            "concise": "Respond concisely and directly to the point."345        }346        347        system_prompt = style_prompts.get(style, style_prompts["professional"])348        349        try:350            response = await self.llm_client.chat.completions.create(351                model="gpt-4",352                messages=[353                    {"role": "system", "content": system_prompt},354                    {"role": "user", "content": prompt}355                ],356                max_tokens=max_tokens,357                temperature=temperature358            )359            return response.choices[0].message.content360        except Exception as e:361            return f"Error generating response: {str(e)}"362 363 364class NLPProcessor:365    """Main NLP processing orchestrator"""366    367    def __init__(self, llm_client=None):368        self.sentiment_analyzer = SentimentAnalyzer()369        self.intent_classifier = IntentClassifier()370        self.entity_extractor = EntityExtractor()371        self.keyphrase_extractor = KeyPhraseExtractor()372        self.text_generator = TextGenerator(llm_client)373        self.logger = logging.getLogger(__name__)374    375    async def process(self, text: str) -> NLPResult:376        """Comprehensive NLP processing"""377        import time378        start_time = time.time()379        380        try:381            # Run all analyses concurrently382            sentiment_task = self.sentiment_analyzer.analyze(text)383            intent_task = self.intent_classifier.classify(text)384            entities_task = self.entity_extractor.extract(text)385            keyphrases_task = self.keyphrase_extractor.extract(text)386            387            sentiment, intent, entities, key_phrases = await asyncio.gather(388                sentiment_task, intent_task, entities_task, keyphrases_task389            )390            391            processing_time = time.time() - start_time392            393            result = NLPResult(394                original_text=text,395                sentiment=sentiment,396                intent=intent,397                entities=entities,398                key_phrases=key_phrases,399                processing_time=processing_time400            )401            402            self.logger.info(f"NLP processing completed in {processing_time:.2f}s")403            return result404            405        except Exception as e:406            self.logger.error(f"NLP processing failed: {e}")407            # Return basic result on error408            return NLPResult(409                original_text=text,410                sentiment=SentimentResult(SentimentLabel.NEUTRAL, 0.0, 0.0),411                intent=IntentResult(IntentCategory.UNKNOWN, 0.0),412                entities=[],413                key_phrases=[],414                processing_time=time.time() - start_time415            )416    417    async def generate_contextual_response(self, 418                                         nlp_result: NLPResult,419                                         context: Dict[str, Any] = None) -> str:420        """Generate contextual response based on NLP analysis"""421        422        # Determine response style based on sentiment and intent423        if nlp_result.sentiment.label == SentimentLabel.NEGATIVE:424            style = "empathetic"425        elif nlp_result.intent.category == IntentCategory.COMPLAINT:426            style = "empathetic"427        elif nlp_result.intent.category in [IntentCategory.QUESTION, IntentCategory.INFORMATION_SEEKING]:428            style = "professional"429        elif nlp_result.intent.urgency_level > 0.5:430            style = "concise"431        else:432            style = "professional"433        434        # Create enhanced prompt with NLP insights435        prompt = f"""436        User message: {nlp_result.original_text}437        438        Analysis:439        - Sentiment: {nlp_result.sentiment.label.value} (confidence: {nlp_result.sentiment.confidence:.2f})440        - Intent: {nlp_result.intent.category.value} (urgency: {nlp_result.intent.urgency_level:.2f})441        - Key entities: {', '.join([e.text for e in nlp_result.entities[:5]])}442        - Key phrases: {', '.join(nlp_result.key_phrases[:5])}443        444        Context: {context or {}}445        446        Please provide an appropriate response.447        """448        449        return await self.text_generator.generate_response(prompt, style)450    451    async def initialize(self):452        """Initialize NLP processor"""453        self.logger.info("NLP processor initialized")454    455    async def shutdown(self):456        """Shutdown NLP processor"""457        self.logger.info("NLP processor shutdown")458