""" Contextual Awareness System ========================== Advanced contextual awareness implementation that identifies and interprets subtle context clues, implicit information, and situational variables across multiple dimensions. """ import asyncio import json import re import logging from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple, Set from dataclasses import dataclass, field from enum import Enum import numpy as np from collections import defaultdict, Counter from ai_agent_framework.core.context_engineering_agent import ( ContextElement, ContextModality, ContextDimension, ContextAwareLLM, ContextualMemoryManager ) logger = logging.getLogger(__name__) class ContextualClue(Enum): """Types of contextual clues to detect.""" TEMPORAL = "temporal" SPATIAL = "spatial" SOCIAL = "social" EMOTIONAL = "emotional" LINGUISTIC = "linguistic" BEHAVIORAL = "behavioral" ENVIRONMENTAL = "environmental" CULTURAL = "cultural" DOMAIN_SPECIFIC = "domain_specific" @dataclass class ContextualSignal: """Represents a detected contextual signal.""" id: str clue_type: ContextualClue content: str confidence: float strength: float timestamp: datetime source_elements: List[str] implied_information: List[str] situational_variables: Dict[str, Any] def __post_init__(self): if not self.timestamp: self.timestamp = datetime.utcnow() if not self.source_elements: self.source_elements = [] if not self.implied_information: self.implied_information = [] if not self.situational_variables: self.situational_variables = {} @dataclass class ImplicitInformation: """Represents information that is implied but not explicitly stated.""" id: str implied_by: List[str] # Elements that imply this information inference_type: str # logical, contextual, cultural, etc. confidence: float evidence_strength: float timestamp: datetime def __post_init__(self): if not self.timestamp: self.timestamp = datetime.utcnow() @dataclass class SituationalContext: """Complete situational context from detected signals.""" situation_type: str primary_signals: List[ContextualSignal] secondary_signals: List[ContextualSignal] confidence_level: float situational_variables: Dict[str, Any] time_relevance: Dict[str, Any] def __post_init__(self): if not self.primary_signals: self.primary_signals = [] if not self.secondary_signals: self.secondary_signals = [] class ContextualAwarenessEngine: """Advanced contextual awareness system for detecting subtle clues and implicit information.""" def __init__(self): self.sensitivity_levels = { ContextualClue.TEMPORAL: 0.8, ContextualClue.SPATIAL: 0.7, ContextualClue.SOCIAL: 0.6, ContextualClue.EMOTIONAL: 0.9, ContextualClue.LINGUISTIC: 0.8, ContextualClue.BEHAVIORAL: 0.7, ContextualClue.ENVIRONMENTAL: 0.6, ContextualClue.CULTURAL: 0.5, ContextualClue.DOMAIN_SPECIFIC: 0.8 } self.pattern_library = self._initialize_pattern_library() self.inference_rules = self._initialize_inference_rules() self.cultural_contexts = self._initialize_cultural_contexts() # Signal tracking self.detected_signals = {} self.signal_history = [] self.context_snapshots = [] def _initialize_pattern_library(self) -> Dict[ContextualClue, List[Dict[str, Any]]]: """Initialize pattern library for detecting contextual clues.""" return { ContextualClue.TEMPORAL: [ { "pattern": r"(?i)\b(immediately|urgent|asap|now|right away)\b", "signal_type": "urgency_temporal", "confidence": 0.8, "implications": ["time_pressure", "stress_level_high"] }, { "pattern": r"(?i)\b(later|tomorrow|next week|eventually|sometime)\b", "signal_type": "future_focused", "confidence": 0.7, "implications": ["planning_mode", "flexible_timeline"] } ], ContextualClue.SOCIAL: [ { "pattern": r"(?i)\b(we|us|our team|our group)\b", "signal_type": "collective_identity", "confidence": 0.7, "implications": ["collaborative_context", "group_oriented"] }, { "pattern": r"(?i)\b(sorry|excuse me|thank you|please)\b", "signal_type": "politeness_markers", "confidence": 0.8, "implications": ["formal_interaction", "respectful_communication"] } ], ContextualClue.EMOTIONAL: [ { "pattern": r"(?i)\b(frustrated|annoyed|mad|angry|furious)\b", "signal_type": "negative_emotion", "confidence": 0.9, "implications": ["emotional_state_negative", "potential_conflict"] }, { "pattern": r"(?i)\b(happy|excited|great|wonderful|amazing)\b", "signal_type": "positive_emotion", "confidence": 0.8, "implications": ["emotional_state_positive", "open_to_ideas"] } ], ContextualClue.LINGUISTIC: [ { "pattern": r"(?i)\b(however|but|although|nevertheless)\b", "signal_type": "contrast_indicators", "confidence": 0.8, "implications": ["complex_reasoning", "qualifying_statements"] }, { "pattern": r"(?i)\b(therefore|thus|hence|consequently)\b", "signal_type": "conclusion_indicators", "confidence": 0.7, "implications": ["logical_reasoning", "result_focused"] } ], ContextualClue.BEHAVIORAL: [ { "pattern": r"(?i)\b(let me check|I'll look into that|I need to verify)\b", "signal_type": "verification_behavior", "confidence": 0.7, "implications": ["accuracy_focused", "thorough_approach"] }, { "pattern": r"(?i)\b(I'm not sure|I don't know|I'm guessing)\b", "signal_type": "uncertainty_expression", "confidence": 0.9, "implications": ["honest_communication", "humility_in_expertise"] } ] } def _initialize_inference_rules(self) -> List[Dict[str, Any]]: """Initialize rules for inferring implicit information.""" return [ { "rule": "temporal_urgency_implies_stress", "conditions": ["urgency_temporal"], "inference": "person_experiencing_time_pressure", "confidence": 0.7 }, { "rule": "collective_language_implies_collaboration", "conditions": ["collective_identity"], "inference": "group_oriented_work_environment", "confidence": 0.8 }, { "rule": "emotional_indicators_imply_sensitivity", "conditions": ["negative_emotion", "politeness_markers"], "inference": "emotionally_aware_interaction", "confidence": 0.6 }, { "rule": "uncertainty_expressions_imply_honesty", "conditions": ["uncertainty_expression", "verification_behavior"], "inference": "truth_seeking_communication_style", "confidence": 0.8 } ] def _initialize_cultural_contexts(self) -> Dict[str, Dict[str, Any]]: """Initialize cultural context knowledge.""" return { "western_business": { "communication_style": "direct", "decision_making": "individual", "time_orientation": "future", "formality_level": "moderate" }, "eastern_business": { "communication_style": "indirect", "decision_making": "consensus", "time_orientation": "present", "formality_level": "high" }, "academic_research": { "communication_style": "analytical", "decision_making": "evidence_based", "time_orientation": "detailed", "formality_level": "high" }, "casual_conversation": { "communication_style": "informal", "decision_making": "spontaneous", "time_orientation": "immediate", "formality_level": "low" } } async def detect_contextual_signals( self, input_text: str, context_elements: List[ContextElement], user_profile: Optional[Any] = None ) -> Tuple[List[ContextualSignal], List[ImplicitInformation]]: """Detect contextual signals and implicit information from input.""" try: # Parse input for explicit signals explicit_signals = await self._parse_explicit_signals(input_text) # Detect implicit information implicit_information = await self._detect_implicit_information( input_text, explicit_signals, context_elements ) # Identify situational context situational_context = await self._identify_situational_context( explicit_signals, implicit_information, user_profile ) # Generate composite signals composite_signals = await self._generate_composite_signals( explicit_signals, implicit_information, situational_context ) return explicit_signals + composite_signals, implicit_information except Exception as e: logger.error(f"Contextual signal detection failed: {e}") return [], [] async def _parse_explicit_signals(self, text: str) -> List[ContextualSignal]: """Parse text for explicit contextual signals.""" signals = [] # Apply pattern matching for different clue types for clue_type, patterns in self.pattern_library.items(): for pattern_info in patterns: matches = re.findall(pattern_info["pattern"], text) for match in matches: # Calculate signal strength based on confidence and frequency frequency_factor = len(matches) / max(len(text.split()), 1) strength = pattern_info["confidence"] * min(1.0, frequency_factor * 2) signal = ContextualSignal( id=f"signal_{clue_type.value}_{hash(match)}", clue_type=clue_type, content=match, confidence=pattern_info["confidence"], strength=strength, timestamp=datetime.utcnow(), source_elements=[match], implied_information=pattern_info.get("implications", []), situational_variables={ "pattern_matched": pattern_info["pattern"], "signal_type": pattern_info["signal_type"], "frequency": len(matches) } ) signals.append(signal) # Detect linguistic patterns beyond regex linguistic_signals = await self._detect_linguistic_patterns(text) signals.extend(linguistic_signals) # Detect emotional indicators emotional_signals = await self._detect_emotional_patterns(text) signals.extend(emotional_signals) return signals async def _detect_linguistic_patterns(self, text: str) -> List[ContextualSignal]: """Detect sophisticated linguistic patterns.""" signals = [] # Sentence complexity analysis sentences = text.split('.') avg_sentence_length = np.mean([len(s.split()) for s in sentences if s.strip()]) if avg_sentence_length > 15: signals.append(ContextualSignal( id="linguistic_complexity_high", clue_type=ContextualClue.LINGUISTIC, content=f"Complex sentence structure (avg: {avg_sentence_length:.1f} words)", confidence=0.7, strength=0.6, timestamp=datetime.utcnow(), source_elements=[text[:100]], implied_information=["analytical_thinking", "detail_oriented"], situational_variables={ "metric": "average_sentence_length", "value": avg_sentence_length } )) # Question vs statement ratio questions = text.count('?') statements = len(sentences) - questions if questions > statements: signals.append(ContextualSignal( id="linguistic_inquiry_heavy", clue_type=ContextualClue.LINGUISTIC, content="Inquiry-heavy communication pattern", confidence=0.8, strength=0.7, timestamp=datetime.utcnow(), source_elements=[text], implied_information=["exploratory_mindset", "information_seeking"], situational_variables={ "question_ratio": questions / max(statements, 1), "total_questions": questions } )) return signals async def _detect_emotional_patterns(self, text: str) -> List[ContextualSignal]: """Detect emotional patterns in communication.""" signals = [] # Emotional intensity indicators intensity_words = { "high": ["extremely", "incredibly", "absolutely", "totally", "completely"], "medium": ["very", "quite", "rather", "pretty", "fairly"], "low": ["somewhat", "a bit", "slightly", "minor"] } intensity_count = {"high": 0, "medium": 0, "low": 0} for intensity, words in intensity_words.items(): for word in words: if word.lower() in text.lower(): intensity_count[intensity] += text.lower().count(word.lower()) total_intensity = sum(intensity_count.values()) if total_intensity > 0: primary_intensity = max(intensity_count, key=intensity_count.get) signals.append(ContextualSignal( id="emotional_intensity_pattern", clue_type=ContextualClue.EMOTIONAL, content=f"Emotional intensity: {primary_intensity} level", confidence=0.6, strength=total_intensity / max(len(text.split()), 1) * 10, # Normalize timestamp=datetime.utcnow(), source_elements=[text], implied_information=["emotional_expression", "emphasis_seeking"], situational_variables={ "intensity_distribution": intensity_count, "primary_intensity": primary_intensity } )) return signals async def _detect_implicit_information( self, text: str, signals: List[ContextualSignal], context_elements: List[ContextElement] ) -> List[ImplicitInformation]: """Detect implicit information using inference rules.""" implicit_info = [] # Extract signal types for rule matching signal_types = {} for signal in signals: signal_type = signal.situational_variables.get("signal_type", "") if signal_type: signal_types[signal_type] = signal_types.get(signal_type, 0) + 1 # Apply inference rules for rule in self.inference_rules: conditions = rule["conditions"] # Check if all conditions are met conditions_met = any( condition in signal_types for condition in conditions ) if conditions_met: implied = ImplicitInformation( id=f"implicit_{rule['rule']}_{len(implicit_info)}", implied_by=[str(signal.id) for signal in signals], inference_type=rule["rule"], confidence=rule["confidence"], evidence_strength=sum( signal.confidence for signal in signals if signal.situational_variables.get("signal_type") in conditions ) / max(len(conditions), 1), timestamp=datetime.utcnow() ) implicit_info.append(implied) # Context-based inference context_implicit = await self._infer_from_context(context_elements) implicit_info.extend(context_implicit) return implicit_info async def _infer_from_context(self, context_elements: List[ContextElement]) -> List[ImplicitInformation]: """Infer implicit information from context elements.""" implicit_info = [] # Pattern analysis in context elements element_themes = defaultdict(int) for element in context_elements: # Extract themes from content content_str = str(element.content).lower() # Simple theme extraction (in production would use NLP) if "urgent" in content_str or "asap" in content_str: element_themes["time_pressure"] += 1 if "team" in content_str or "group" in content_str: element_themes["collaborative_work"] += 1 if "customer" in content_str or "client" in content_str: element_themes["customer_focus"] += 1 # Generate implicit information from patterns if element_themes: for theme, count in element_themes.items(): if count >= 2: # Pattern needs at least 2 occurrences implicit = ImplicitInformation( id=f"context_pattern_{theme}", implied_by=[element.id for element in context_elements], inference_type="context_pattern_analysis", confidence=min(0.9, count * 0.3), evidence_strength=count / max(len(context_elements), 1), timestamp=datetime.utcnow() ) implicit_info.append(implicit) return implicit_info async def _identify_situational_context( self, signals: List[ContextualSignal], implicit_info: List[ImplicitInformation], user_profile: Optional[Any] = None ) -> SituationalContext: """Identify the overall situational context.""" # Categorize signals by strength and confidence strong_signals = [s for s in signals if s.strength > 0.7 and s.confidence > 0.6] moderate_signals = [s for s in signals if 0.4 <= s.strength <= 0.7] # Determine situation type situation_type = self._classify_situation_type(strong_signals, moderate_signals) # Calculate confidence level if strong_signals: confidence_level = np.mean([s.confidence for s in strong_signals]) else: confidence_level = np.mean([s.confidence for s in signals]) if signals else 0.5 # Extract situational variables situational_variables = {} for signal in strong_signals: situational_variables.update(signal.situational_variables) # Add user profile context if available if user_profile: situational_variables.update({ "user_communication_style": getattr(user_profile, 'communication_style', {}), "user_preferences": getattr(user_profile, 'preferences', {}) }) # Calculate time relevance time_relevance = self._calculate_time_relevance(signals) return SituationalContext( situation_type=situation_type, primary_signals=strong_signals, secondary_signals=moderate_signals, confidence_level=confidence_level, situational_variables=situational_variables, time_relevance=time_relevance ) def _classify_situation_type( self, strong_signals: List[ContextualSignal], moderate_signals: List[ContextualSignal] ) -> str: """Classify the overall situation type based on signals.""" # Count signal types clue_counts = defaultdict(int) for signal in strong_signals + moderate_signals: clue_counts[signal.clue_type] += 1 # Determine dominant context if clue_counts[ContextualClue.TEMPORAL] > 2: return "time_pressured" elif clue_counts[ContextualClue.SOCIAL] > 2: return "collaborative" elif clue_counts[ContextualClue.EMOTIONAL] > 2: return "emotionally_charged" elif clue_counts[ContextualClue.LINGUISTIC] > 3: return "complex_communication" elif clue_counts[ContextualClue.BEHAVIORAL] > 1: return "behavioral_analysis" else: return "general_interaction" def _calculate_time_relevance(self, signals: List[ContextualSignal]) -> Dict[str, Any]: """Calculate time-based relevance of signals.""" current_time = datetime.utcnow() time_relevance = {} for signal in signals: age_minutes = (current_time - signal.timestamp).total_seconds() / 60 # Calculate recency score (higher = more recent) recency_score = max(0, 1 - age_minutes / 60) # Decay over 1 hour time_relevance[signal.id] = { "age_minutes": age_minutes, "recency_score": recency_score, "freshness": "fresh" if age_minutes < 10 else "recent" if age_minutes < 60 else "stale" } return time_relevance async def _generate_composite_signals( self, explicit_signals: List[ContextualSignal], implicit_info: List[ImplicitInformation], situational_context: SituationalContext ) -> List[ContextualSignal]: """Generate composite signals from combinations of explicit signals.""" composite_signals = [] # Detect signal combinations signal_combinations = self._find_signal_combinations(explicit_signals) for combination in signal_combinations: if len(combination) >= 2: # Create composite signal from combination combined_content = " + ".join([s.content for s in combination[:3]]) avg_confidence = np.mean([s.confidence for s in combination]) combined_strength = np.mean([s.strength for s in combination]) composite_signal = ContextualSignal( id=f"composite_{len(composite_signals)}", clue_type=ContextualClue.DOMAIN_SPECIFIC, content=f"Composite pattern: {combined_content}", confidence=avg_confidence * 0.8, # Reduce confidence for composites strength=combined_strength, timestamp=datetime.utcnow(), source_elements=[s.id for s in combination], implied_information=[f"combined_pattern_{i}" for i in range(len(combination))], situational_variables={ "combination_size": len(combination), "primary_clue_types": [s.clue_type.value for s in combination], "situational_context": situational_context.situation_type } ) composite_signals.append(composite_signal) return composite_signals def _find_signal_combinations(self, signals: List[ContextualSignal]) -> List[List[ContextualSignal]]: """Find meaningful combinations of signals.""" combinations = [] # Look for complementary signal types for i, signal1 in enumerate(signals): for j, signal2 in enumerate(signals[i+1:], i+1): # Check for complementary clue types complementary_pairs = { (ContextualClue.TEMPORAL, ContextualClue.EMOTIONAL), (ContextualClue.SOCIAL, ContextualClue.LINGUISTIC), (ContextualClue.EMOTIONAL, ContextualClue.BEHAVIORAL) } if (signal1.clue_type, signal2.clue_type) in complementary_pairs: combinations.append([signal1, signal2]) elif signal1.clue_type == signal2.clue_type and signal1.clue_type == ContextualClue.LINGUISTIC: # Multiple linguistic signals combinations.append([signal1, signal2]) return combinations def get_awareness_metrics(self) -> Dict[str, Any]: """Get metrics about the awareness system's performance.""" return { "total_signals_detected": len(self.detected_signals), "signal_types_distribution": dict(Counter( signal.clue_type.value for signal in self.detected_signals.values() )) if self.detected_signals else {}, "average_confidence": np.mean([s.confidence for s in self.detected_signals.values()]) if self.detected_signals else 0, "recent_signal_count": len([s for s in self.detected_signals.values() if (datetime.utcnow() - s.timestamp).total_seconds() < 3600]) if self.detected_signals else 0, "pattern_library_coverage": { clue_type.value: len(patterns) for clue_type, patterns in self.pattern_library.items() } } # Context-aware processing integration class ContextualAwarenessProcessor: """Processor that integrates contextual awareness with the main system.""" def __init__(self): self.awareness_engine = ContextualAwarenessEngine() self.memory_manager = None self.sensitivity_adjustment = {} async def initialize(self, memory_manager: ContextualMemoryManager): """Initialize with memory manager.""" self.memory_manager = memory_manager async def process_input_with_awareness( self, input_text: str, context_elements: List[ContextElement], user_profile: Optional[Any] = None ) -> Dict[str, Any]: """Process input with full contextual awareness.""" # Detect contextual signals and implicit information signals, implicit_info = await self.awareness_engine.detect_contextual_signals( input_text, context_elements, user_profile ) # Store detected signals in memory await self._store_signals_in_memory(signals, implicit_info) # Generate contextual understanding report understanding_report = await self._generate_understanding_report( signals, implicit_info, context_elements ) return { "contextual_signals": [self._signal_to_dict(signal) for signal in signals], "implicit_information": [self._implicit_to_dict(info) for info in implicit_info], "understanding_report": understanding_report, "awareness_metrics": self.awareness_engine.get_awareness_metrics() } async def _store_signals_in_memory( self, signals: List[ContextualSignal], implicit_info: List[ImplicitInformation] ): """Store detected signals and implicit information in memory.""" if not self.memory_manager: return # Store signals as context elements for signal in signals: context_element = ContextElement( id=signal.id, content=signal.content, modality=ContextModality.BEHAVIORAL, timestamp=signal.timestamp, relevance_score=signal.strength, confidence=signal.confidence, expires_at=signal.timestamp + timedelta(hours=2), # Signals expire relatively quickly source="contextual_awareness", metadata={ "signal_type": signal.clue_type.value, "implied_information": signal.implied_information, "situational_variables": signal.situational_variables, "is_signal": True }, tags={"contextual_signal", signal.clue_type.value} ) await self.memory_manager.store_context(context_element) # Store implicit information for info in implicit_info: context_element = ContextElement( id=info.id, content=f"Implicit: {info.inference_type}", modality=ContextModality.BEHAVIORAL, timestamp=info.timestamp, relevance_score=info.confidence, confidence=info.evidence_strength, expires_at=info.timestamp + timedelta(hours=4), # Implicit info lasts longer source="contextual_awareness", metadata={ "inferred_by": info.implied_by, "inference_type": info.inference_type, "is_implicit": True }, tags={"implicit_information", info.inference_type} ) await self.memory_manager.store_context(context_element) async def _generate_understanding_report( self, signals: List[ContextualSignal], implicit_info: List[ImplicitInformation], context_elements: List[ContextElement] ) -> Dict[str, Any]: """Generate a comprehensive contextual understanding report.""" # Analyze signal patterns signal_patterns = self._analyze_signal_patterns(signals) # Assess context completeness context_completeness = self._assess_context_completeness(signals, implicit_info, context_elements) # Generate situational assessment situational_assessment = self._generate_situational_assessment(signals) return { "signal_analysis": { "total_signals": len(signals), "high_confidence_signals": len([s for s in signals if s.confidence > 0.7]), "signal_type_distribution": dict(Counter(s.clue_type.value for s in signals)), "patterns_detected": signal_patterns }, "implicit_information_analysis": { "total_implicit_items": len(implicit_info), "high_confidence_inferences": len([i for i in implicit_info if i.confidence > 0.6]), "evidence_strength_distribution": [i.evidence_strength for i in implicit_info] }, "context_assessment": { "completeness_score": context_completeness["completeness"], "confidence_level": context_completeness["confidence"], "coverage_gaps": context_completeness["gaps"] }, "situational_assessment": situational_assessment } def _analyze_signal_patterns(self, signals: List[ContextualSignal]) -> Dict[str, Any]: """Analyze patterns in detected signals.""" if not signals: return {"status": "no_signals"} # Frequency analysis signal_types = Counter(s.clue_type.value for s in signals) content_patterns = Counter(s.content.lower() for s in signals) # Temporal patterns recent_signals = [s for s in signals if (datetime.utcnow() - s.timestamp).total_seconds() < 300] return { "dominant_signal_types": dict(signal_types.most_common(3)), "recurring_content": dict(content_patterns.most_common(5)), "temporal_density": len(recent_signals) / max(len(signals), 1), "confidence_distribution": { "high": len([s for s in signals if s.confidence > 0.7]), "medium": len([s for s in signals if 0.4 <= s.confidence <= 0.7]), "low": len([s for s in signals if s.confidence < 0.4]) } } def _assess_context_completeness( self, signals: List[ContextualSignal], implicit_info: List[ImplicitInformation], context_elements: List[ContextElement] ) -> Dict[str, Any]: """Assess how complete the contextual understanding is.""" # Check coverage of different dimensions covered_dimensions = set(s.clue_type for s in signals) total_dimensions = len(ContextualClue) # Calculate completeness score completeness = len(covered_dimensions) / total_dimensions # Calculate average confidence all_confidences = [s.confidence for s in signals] + [i.confidence for i in implicit_info] avg_confidence = np.mean(all_confidences) if all_confidences else 0 # Identify coverage gaps gaps = [] for dimension in ContextualClue: if dimension not in covered_dimensions: gaps.append(dimension.value) return { "completeness": completeness, "confidence": avg_confidence, "gaps": gaps, "dimensions_covered": len(covered_dimensions), "total_dimensions": total_dimensions } def _generate_situational_assessment(self, signals: List[ContextualSignal]) -> Dict[str, Any]: """Generate assessment of the current situation.""" if not signals: return {"status": "insufficient_data"} # Determine dominant characteristics signal_strengths = [s.strength for s in signals] signal_confidences = [s.confidence for s in signals] # Categorize situation situation_characteristics = [] if any(s.clue_type == ContextualClue.EMOTIONAL for s in signals): emotional_signals = [s for s in signals if s.clue_type == ContextualClue.EMOTIONAL] avg_emotional_intensity = np.mean([s.strength for s in emotional_signals]) if avg_emotional_intensity > 0.6: situation_characteristics.append("emotionally_intense") elif avg_emotional_intensity > 0.4: situation_characteristics.append("moderately_emotional") if any(s.clue_type == ContextualClue.TEMPORAL for s in signals): if any("urgent" in s.content.lower() or "asap" in s.content.lower() for s in signals): situation_characteristics.append("time_pressured") if any(s.clue_type == ContextualClue.SOCIAL for s in signals): if any("team" in s.content.lower() or "we" in s.content.lower() for s in signals): situation_characteristics.append("collaborative") return { "situation_characteristics": situation_characteristics, "overall_intensity": np.mean(signal_strengths), "confidence_level": np.mean(signal_confidences), "assessment": self._generate_situation_description(situation_characteristics) } def _generate_situation_description(self, characteristics: List[str]) -> str: """Generate natural language description of the situation.""" if not characteristics: return "General interaction with moderate contextual awareness." descriptions = { "emotionally_intense": "High emotional engagement detected", "time_pressured": "Urgency and time pressure indicators present", "collaborative": "Collaborative and team-oriented communication style", "analytical": "Detailed analytical thinking patterns observed" } primary_desc = descriptions.get(characteristics[0], characteristics[0]) if len(characteristics) > 1: return f"{primary_desc}, with additional {characteristics[1]} patterns." else: return f"{primary_desc}." def _signal_to_dict(self, signal: ContextualSignal) -> Dict[str, Any]: """Convert signal to dictionary for serialization.""" return { "id": signal.id, "clue_type": signal.clue_type.value, "content": signal.content, "confidence": signal.confidence, "strength": signal.strength, "timestamp": signal.timestamp.isoformat(), "implied_information": signal.implied_information, "situational_variables": signal.situational_variables } def _implicit_to_dict(self, info: ImplicitInformation) -> Dict[str, Any]: """Convert implicit information to dictionary for serialization.""" return { "id": info.id, "inference_type": info.inference_type, "confidence": info.confidence, "evidence_strength": info.evidence_strength, "timestamp": info.timestamp.isoformat(), "implied_by": info.implied_by } if __name__ == "__main__": print("Contextual Awareness System Initialized") print("=" * 50) processor = ContextualAwarenessProcessor() print("Ready to detect subtle contextual clues and implicit information!")