#!/usr/bin/env python3 """ Deep Behavioral Analysis - Advanced pattern recognition for steward profiling. This module implements sophisticated behavioral analysis to understand: - Communication patterns and preferences - Decision-making styles - Work rhythms and productivity patterns - Technical preferences and coding styles - Emotional states and stress indicators - Collaboration preferences """ import json import os import re from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from collections import defaultdict, Counter import numpy as np from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent.parent)) from config import MEMORY_DIR from utils.logging import setup_logger logger = setup_logger(__name__) class DeepBehavioralAnalyzer: """Performs deep analysis of user behavior patterns.""" def __init__(self): self.analysis_dir = os.path.join(MEMORY_DIR, "behavioral_analysis") os.makedirs(self.analysis_dir, exist_ok=True) # Pattern recognizers self.communication_patterns = CommunicationPatternAnalyzer() self.work_rhythm_analyzer = WorkRhythmAnalyzer() self.decision_pattern_analyzer = DecisionPatternAnalyzer() self.emotional_state_analyzer = EmotionalStateAnalyzer() self.technical_preference_analyzer = TechnicalPreferenceAnalyzer() # Historical data self.interaction_history = self._load_interaction_history() def analyze_message(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]: """ Perform comprehensive behavioral analysis on a message. Returns detailed insights about behavioral patterns. """ timestamp = datetime.now() # Analyze different aspects communication = self.communication_patterns.analyze(message) emotional = self.emotional_state_analyzer.analyze(message, context) technical = self.technical_preference_analyzer.analyze(message, context) # Store interaction interaction = { "timestamp": timestamp.isoformat(), "message": message, "context": context or {}, "analysis": { "communication": communication, "emotional": emotional, "technical": technical } } self.interaction_history.append(interaction) self._save_interaction_history() # Update analyzers with new data self.work_rhythm_analyzer.update(timestamp) self.decision_pattern_analyzer.analyze_decision(message, context) return { "immediate_insights": { "communication_style": communication["style"], "emotional_state": emotional["state"], "urgency_level": communication["urgency"], "technical_focus": technical["focus_areas"] }, "patterns": self._extract_patterns() } def _extract_patterns(self) -> Dict[str, Any]: """Extract high-level patterns from accumulated data.""" if len(self.interaction_history) < 10: return {"status": "insufficient_data"} return { "work_rhythm": self.work_rhythm_analyzer.get_patterns(), "communication_preferences": self.communication_patterns.get_preferences(), "decision_style": self.decision_pattern_analyzer.get_style(), "emotional_patterns": self.emotional_state_analyzer.get_patterns(), "technical_preferences": self.technical_preference_analyzer.get_preferences() } def _load_interaction_history(self) -> List[Dict]: """Load historical interaction data.""" history_file = os.path.join(self.analysis_dir, "interaction_history.json") if os.path.exists(history_file): try: with open(history_file, 'r') as f: return json.load(f) except: return [] return [] def _save_interaction_history(self): """Save interaction history (keep last 1000 interactions).""" history_file = os.path.join(self.analysis_dir, "interaction_history.json") # Keep only recent history self.interaction_history = self.interaction_history[-1000:] with open(history_file, 'w') as f: json.dump(self.interaction_history, f, indent=2) def analyze_behavioral_patterns(self, timeframe: str = "recent", include_personality: bool = True, include_patterns: bool = True) -> Dict[str, Any]: """MCP Interface method - analyze behavioral patterns over timeframe""" try: patterns = self._extract_patterns() if patterns.get("status") == "insufficient_data": return { "status": "insufficient_data", "message": "Need more interactions to build behavioral profile", "timeframe": timeframe } result = { "patterns": patterns, "timeframe": timeframe } if include_personality: result["personality_profile"] = { "communication_style": patterns["communication_preferences"].get("primary_style", "unknown"), "work_style": patterns["work_rhythm"].get("work_style", "unknown"), "decision_style": patterns["decision_style"].get("primary_style", "unknown"), "emotional_baseline": patterns["emotional_patterns"].get("dominant_category", "neutral") } if include_patterns: result["behavioral_insights"] = { "peak_productivity": patterns["work_rhythm"].get("peak_hours", []), "communication_preferences": patterns["communication_preferences"], "emotional_stability": patterns["emotional_patterns"].get("emotional_stability", "unknown") } return result except Exception as e: logger.error(f"Error in behavioral pattern analysis: {e}") return { "status": "error", "error": str(e), "timeframe": timeframe } def get_user_context_summary(self) -> str: """Get a brief summary of user context for MCP interface""" try: patterns = self._extract_patterns() if patterns.get("status") == "insufficient_data": return "Insufficient interaction data for behavioral summary" comm_style = patterns.get("communication_preferences", {}).get("primary_style", "unknown") work_style = patterns.get("work_rhythm", {}).get("work_style", "unknown") return f"Communication: {comm_style}, Work style: {work_style}" except Exception as e: logger.error(f"Error getting user context summary: {e}") return "Context summary unavailable" def get_steward_personality_summary(self) -> Dict[str, Any]: """Get steward personality summary for MCP interface""" try: patterns = self._extract_patterns() if patterns.get("status") == "insufficient_data": return {"status": "insufficient_data"} return { "communication_patterns": patterns.get("communication_preferences", {}), "work_patterns": patterns.get("work_rhythm", {}), "decision_patterns": patterns.get("decision_style", {}), "emotional_patterns": patterns.get("emotional_patterns", {}), "technical_patterns": patterns.get("technical_preferences", {}) } except Exception as e: logger.error(f"Error getting steward personality summary: {e}") return {"status": "error", "error": str(e)} class CommunicationPatternAnalyzer: """Analyzes communication style and patterns.""" def __init__(self): self.style_indicators = { "formal": ["please", "kindly", "would you", "could you", "thank you"], "casual": ["hey", "gonna", "wanna", "yeah", "cool", "awesome"], "direct": ["do", "need", "want", "must", "now", "asap"], "questioning": ["how", "what", "why", "when", "where", "can you explain"], "collaborative": ["let's", "we", "our", "together", "help me"] } self.urgency_indicators = { "high": ["urgent", "asap", "immediately", "now", "critical", "emergency"], "medium": ["soon", "today", "when you can", "priority"], "low": ["whenever", "no rush", "when convenient", "eventually"] } self.preference_counts = defaultdict(int) def analyze(self, message: str) -> Dict[str, Any]: """Analyze communication patterns in a message.""" message_lower = message.lower() # Detect communication style style_scores = {} for style, indicators in self.style_indicators.items(): score = sum(1 for word in indicators if word in message_lower) style_scores[style] = score primary_style = max(style_scores, key=style_scores.get) if any(style_scores.values()) else "neutral" # Detect urgency urgency = "normal" for level, indicators in self.urgency_indicators.items(): if any(word in message_lower for word in indicators): urgency = level break # Detect question vs statement vs command message_type = "statement" if message.strip().endswith("?"): message_type = "question" elif any(message.lower().startswith(cmd) for cmd in ["do", "please", "can you", "could you"]): message_type = "request" elif any(message.lower().startswith(cmd) for cmd in ["fix", "add", "remove", "change"]): message_type = "command" # Update preferences self.preference_counts[primary_style] += 1 return { "style": primary_style, "urgency": urgency, "message_type": message_type, "style_scores": style_scores, "politeness_level": self._calculate_politeness(message) } def _calculate_politeness(self, message: str) -> str: """Calculate politeness level of the message.""" polite_words = ["please", "thank", "kindly", "appreciate", "grateful", "sorry"] polite_count = sum(1 for word in polite_words if word in message.lower()) if polite_count >= 3: return "very_polite" elif polite_count >= 1: return "polite" else: return "neutral" def get_preferences(self) -> Dict[str, Any]: """Get learned communication preferences.""" total = sum(self.preference_counts.values()) if total == 0: return {"primary_style": "unknown"} return { "primary_style": max(self.preference_counts, key=self.preference_counts.get), "style_distribution": { style: count/total for style, count in self.preference_counts.items() } } class WorkRhythmAnalyzer: """Analyzes work patterns and productivity rhythms.""" def __init__(self): self.activity_log = defaultdict(list) # hour -> timestamps self.daily_patterns = defaultdict(int) # day_of_week -> activity count self.session_durations = [] self.last_activity = None def update(self, timestamp: datetime): """Update activity tracking.""" hour = timestamp.hour day = timestamp.weekday() self.activity_log[hour].append(timestamp) self.daily_patterns[day] += 1 # Track session duration if self.last_activity: gap = (timestamp - self.last_activity).total_seconds() / 60 # minutes if gap < 30: # Same session if less than 30 min gap if self.session_durations and isinstance(self.session_durations[-1], list): self.session_durations[-1].append(timestamp) else: self.session_durations.append([self.last_activity, timestamp]) else: # New session self.session_durations.append([timestamp]) self.last_activity = timestamp def get_patterns(self) -> Dict[str, Any]: """Extract work rhythm patterns.""" if not self.activity_log: return {"status": "no_data"} # Find peak hours hour_counts = {hour: len(timestamps) for hour, timestamps in self.activity_log.items()} peak_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)[:3] # Find peak days day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] peak_days = sorted(self.daily_patterns.items(), key=lambda x: x[1], reverse=True)[:3] # Calculate average session duration session_lengths = [] for session in self.session_durations: if len(session) >= 2: duration = (session[-1] - session[0]).total_seconds() / 60 session_lengths.append(duration) avg_session = np.mean(session_lengths) if session_lengths else 0 return { "peak_hours": [hour for hour, _ in peak_hours], "peak_days": [day_names[day] for day, _ in peak_days], "average_session_minutes": round(avg_session, 1), "total_sessions": len(self.session_durations), "work_style": self._determine_work_style(hour_counts) } def _determine_work_style(self, hour_counts: Dict[int, int]) -> str: """Determine work style based on activity distribution.""" if not hour_counts: return "unknown" morning = sum(hour_counts.get(h, 0) for h in range(6, 12)) afternoon = sum(hour_counts.get(h, 0) for h in range(12, 18)) evening = sum(hour_counts.get(h, 0) for h in range(18, 24)) night = sum(hour_counts.get(h, 0) for h in list(range(0, 6)) + list(range(0, 6))) total = morning + afternoon + evening + night if total == 0: return "unknown" if morning / total > 0.4: return "morning_person" elif evening / total > 0.4: return "evening_person" elif night / total > 0.3: return "night_owl" else: return "flexible" class DecisionPatternAnalyzer: """Analyzes decision-making patterns and preferences.""" def __init__(self): self.decisions = [] self.decision_keywords = { "analytical": ["data", "metrics", "analyze", "compare", "evaluate", "consider"], "intuitive": ["feel", "think", "seems", "probably", "maybe", "gut"], "decisive": ["definitely", "absolutely", "must", "will", "going to"], "cautious": ["might", "could", "perhaps", "possibly", "careful", "safe"], "collaborative": ["discuss", "agree", "consensus", "team", "together"], "independent": ["I'll", "my way", "decided", "myself", "alone"] } def analyze_decision(self, message: str, context: Optional[Dict] = None): """Analyze decision-making patterns in a message.""" message_lower = message.lower() # Check if this is a decision point decision_indicators = ["let's", "should", "will", "going to", "decided", "choose", "pick"] if not any(ind in message_lower for ind in decision_indicators): return # Analyze decision style style_scores = {} for style, keywords in self.decision_keywords.items(): score = sum(1 for word in keywords if word in message_lower) style_scores[style] = score if any(style_scores.values()): primary_style = max(style_scores, key=style_scores.get) self.decisions.append({ "timestamp": datetime.now().isoformat(), "style": primary_style, "context": context or {}, "confidence": self._assess_confidence(message) }) def _assess_confidence(self, message: str) -> str: """Assess confidence level in decision.""" high_confidence = ["definitely", "absolutely", "certainly", "must", "will"] low_confidence = ["maybe", "perhaps", "might", "could", "possibly"] message_lower = message.lower() if any(word in message_lower for word in high_confidence): return "high" elif any(word in message_lower for word in low_confidence): return "low" else: return "medium" def get_style(self) -> Dict[str, Any]: """Get decision-making style analysis.""" if not self.decisions: return {"primary_style": "unknown"} style_counts = Counter(d["style"] for d in self.decisions) confidence_counts = Counter(d["confidence"] for d in self.decisions) return { "primary_style": style_counts.most_common(1)[0][0], "style_distribution": dict(style_counts), "average_confidence": confidence_counts.most_common(1)[0][0], "decision_count": len(self.decisions) } class EmotionalStateAnalyzer: """Analyzes emotional patterns and stress indicators.""" def __init__(self): self.emotional_indicators = { "positive": { "excited": ["excited", "awesome", "amazing", "fantastic", "great", "love"], "happy": ["happy", "glad", "pleased", "delighted", "joy"], "grateful": ["thank", "appreciate", "grateful", "thanks"], "confident": ["confident", "sure", "certain", "definitely"] }, "negative": { "frustrated": ["frustrated", "annoying", "irritating", "ugh", "argh"], "stressed": ["stressed", "overwhelmed", "too much", "pressure", "deadline"], "confused": ["confused", "don't understand", "unclear", "lost", "what"], "worried": ["worried", "concern", "afraid", "nervous", "anxious"] }, "neutral": { "focused": ["focus", "concentrate", "working on", "deep dive"], "curious": ["wonder", "curious", "interesting", "how does", "why"], "calm": ["calm", "relaxed", "steady", "no rush"] } } self.emotional_history = [] def analyze(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]: """Analyze emotional state from message.""" message_lower = message.lower() # Count emotional indicators emotional_scores = defaultdict(lambda: defaultdict(int)) for category, emotions in self.emotional_indicators.items(): for emotion, indicators in emotions.items(): score = sum(1 for word in indicators if word in message_lower) if score > 0: emotional_scores[category][emotion] = score # Determine primary emotional state primary_category = "neutral" primary_emotion = "calm" max_score = 0 for category, emotions in emotional_scores.items(): for emotion, score in emotions.items(): if score > max_score: max_score = score primary_category = category primary_emotion = emotion # Analyze stress level stress_level = self._analyze_stress_level(message, primary_category) # Store in history self.emotional_history.append({ "timestamp": datetime.now().isoformat(), "category": primary_category, "emotion": primary_emotion, "stress_level": stress_level }) return { "state": primary_emotion, "category": primary_category, "stress_level": stress_level, "indicators": dict(emotional_scores) } def _analyze_stress_level(self, message: str, emotional_category: str) -> str: """Analyze stress level from message characteristics.""" stress_indicators = { "high": ["!!!", "ASAP", "URGENT", "immediately", "critical"], "medium": ["soon", "quickly", "need", "important"], "low": ["whenever", "no rush", "relaxed", "calm"] } # Check message characteristics exclamation_count = message.count("!") caps_ratio = sum(1 for c in message if c.isupper()) / max(len(message), 1) # Determine stress level if emotional_category == "negative" or exclamation_count > 2 or caps_ratio > 0.3: return "high" elif any(ind in message.lower() for ind in stress_indicators["high"]): return "high" elif any(ind in message.lower() for ind in stress_indicators["medium"]): return "medium" else: return "low" def get_patterns(self) -> Dict[str, Any]: """Get emotional patterns over time.""" if not self.emotional_history: return {"status": "no_data"} # Analyze emotional trends recent = self.emotional_history[-20:] # Last 20 interactions category_counts = Counter(e["category"] for e in recent) emotion_counts = Counter(e["emotion"] for e in recent) stress_counts = Counter(e["stress_level"] for e in recent) return { "dominant_category": category_counts.most_common(1)[0][0], "frequent_emotions": [e for e, _ in emotion_counts.most_common(3)], "average_stress": stress_counts.most_common(1)[0][0], "emotional_stability": self._calculate_stability() } def _calculate_stability(self) -> str: """Calculate emotional stability based on variation.""" if len(self.emotional_history) < 5: return "unknown" recent = self.emotional_history[-10:] categories = [e["category"] for e in recent] # Count category changes changes = sum(1 for i in range(1, len(categories)) if categories[i] != categories[i-1]) change_ratio = changes / (len(categories) - 1) if change_ratio < 0.3: return "stable" elif change_ratio < 0.6: return "moderate" else: return "variable" class TechnicalPreferenceAnalyzer: """Analyzes technical preferences and coding style.""" def __init__(self): self.preferences = defaultdict(int) self.technical_keywords = { "languages": { "python": ["python", "py", "pip", "django", "flask"], "javascript": ["javascript", "js", "node", "npm", "react", "vue"], "typescript": ["typescript", "ts", "types", "interface"], "rust": ["rust", "cargo", "rustc"], "go": ["golang", "go ", "goroutine"] }, "paradigms": { "functional": ["functional", "pure", "immutable", "map", "reduce"], "oop": ["class", "object", "inheritance", "encapsulation"], "procedural": ["procedure", "function", "script"] }, "practices": { "testing": ["test", "unit test", "integration", "tdd", "coverage"], "documentation": ["document", "readme", "comments", "docstring"], "refactoring": ["refactor", "clean", "optimize", "improve"], "security": ["security", "vulnerability", "authentication", "encryption"] } } def analyze(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]: """Analyze technical preferences from message.""" message_lower = message.lower() detected = defaultdict(list) for category, items in self.technical_keywords.items(): for name, keywords in items.items(): if any(kw in message_lower for kw in keywords): detected[category].append(name) self.preferences[f"{category}:{name}"] += 1 return { "focus_areas": dict(detected), "technical_depth": self._assess_technical_depth(message), "implementation_style": self._assess_implementation_style(message) } def _assess_technical_depth(self, message: str) -> str: """Assess the technical depth of the message.""" deep_indicators = ["architecture", "design pattern", "algorithm", "complexity", "performance"] surface_indicators = ["basic", "simple", "quick", "easy"] message_lower = message.lower() deep_count = sum(1 for ind in deep_indicators if ind in message_lower) surface_count = sum(1 for ind in surface_indicators if ind in message_lower) if deep_count > surface_count: return "deep" elif surface_count > deep_count: return "surface" else: return "balanced" def _assess_implementation_style(self, message: str) -> str: """Assess preferred implementation style.""" styles = { "pragmatic": ["works", "practical", "simple", "straightforward"], "elegant": ["elegant", "beautiful", "clean", "refined"], "efficient": ["fast", "efficient", "optimized", "performance"], "robust": ["robust", "reliable", "stable", "error handling"] } message_lower = message.lower() style_scores = {} for style, keywords in styles.items(): score = sum(1 for kw in keywords if kw in message_lower) if score > 0: style_scores[style] = score if style_scores: return max(style_scores, key=style_scores.get) return "balanced" def get_preferences(self) -> Dict[str, Any]: """Get learned technical preferences.""" if not self.preferences: return {"status": "no_data"} # Extract top preferences by category language_prefs = {} paradigm_prefs = {} practice_prefs = {} for pref, count in self.preferences.items(): category, name = pref.split(":", 1) if category == "languages": language_prefs[name] = count elif category == "paradigms": paradigm_prefs[name] = count elif category == "practices": practice_prefs[name] = count return { "preferred_languages": sorted(language_prefs.items(), key=lambda x: x[1], reverse=True)[:3], "preferred_paradigms": sorted(paradigm_prefs.items(), key=lambda x: x[1], reverse=True)[:2], "valued_practices": sorted(practice_prefs.items(), key=lambda x: x[1], reverse=True)[:3] } def create_behavioral_profile(analyzer: 'DeepBehavioralAnalyzer') -> Dict[str, Any]: """Create a comprehensive behavioral profile from analysis.""" patterns = analyzer._extract_patterns() if patterns.get("status") == "insufficient_data": return {"status": "insufficient_data", "message": "Need more interactions to build profile"} return { "personality_profile": { "communication_style": patterns["communication_preferences"].get("primary_style", "unknown"), "work_style": patterns["work_rhythm"].get("work_style", "unknown"), "decision_style": patterns["decision_style"].get("primary_style", "unknown"), "emotional_baseline": patterns["emotional_patterns"].get("dominant_category", "neutral") }, "work_patterns": { "peak_productivity": patterns["work_rhythm"].get("peak_hours", []), "preferred_days": patterns["work_rhythm"].get("peak_days", []), "session_style": f"{patterns['work_rhythm'].get('average_session_minutes', 0)} min sessions" }, "technical_profile": { "languages": patterns["technical_preferences"].get("preferred_languages", []), "paradigms": patterns["technical_preferences"].get("preferred_paradigms", []), "practices": patterns["technical_preferences"].get("valued_practices", []) }, "interaction_preferences": { "politeness_preference": patterns["communication_preferences"].get("style_distribution", {}), "emotional_stability": patterns["emotional_patterns"].get("emotional_stability", "unknown"), "stress_tolerance": patterns["emotional_patterns"].get("average_stress", "unknown") } }