#!/usr/bin/env python3 """ Proactive Insight Generation - Background analysis to surface useful patterns ============================================================================ Continuously analyzes accumulated data to proactively generate insights, patterns, and recommendations without being explicitly asked. This system runs in the background to surface valuable discoveries and suggestions. """ import os import json import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Any, Tuple, Optional from collections import defaultdict, Counter from dataclasses import dataclass, asdict from enum import Enum import statistics from core.mira_path_resolver import get_mira_memory_dir from conversations.comprehensive_indexer import ComprehensiveIndexer from intelligence.steward_profile import StewardProfile from intelligence.work_context_intelligence import get_work_context_intelligence from intelligence.emotional_resonance_tracking import get_emotional_memory_tracker from intelligence.relationship_evolution import get_relationship_evolution_tracker from intelligence.predictive_memory_surfacing import get_predictive_memory_surfacer from core.memory.memory_manager import MemoryManager from utils.logging import setup_logger logger = setup_logger(__name__) class InsightType(Enum): """Types of insights that can be generated""" PATTERN_DISCOVERY = "pattern_discovery" EFFICIENCY_OPTIMIZATION = "efficiency_optimization" LEARNING_OPPORTUNITY = "learning_opportunity" EMOTIONAL_WELLBEING = "emotional_wellbeing" COLLABORATION_IMPROVEMENT = "collaboration_improvement" PRODUCTIVITY_INSIGHT = "productivity_insight" KNOWLEDGE_GAP = "knowledge_gap" SUCCESS_PATTERN = "success_pattern" RISK_WARNING = "risk_warning" CREATIVE_OPPORTUNITY = "creative_opportunity" class InsightPriority(Enum): """Priority levels for insights""" CRITICAL = "critical" # Immediate attention needed HIGH = "high" # Should address soon MEDIUM = "medium" # Valuable but not urgent LOW = "low" # Nice to know @dataclass class ProactiveInsight: """A proactively generated insight""" id: str type: InsightType priority: InsightPriority title: str description: str evidence: List[str] # Supporting evidence recommendations: List[str] # Actionable recommendations confidence: float # 0.0 to 1.0 relevance_score: float # 0.0 to 1.0 generated_at: datetime expires_at: Optional[datetime] = None # When insight becomes stale tags: List[str] = None metadata: Dict[str, Any] = None def __post_init__(self): if self.tags is None: self.tags = [] if self.metadata is None: self.metadata = {} def to_dict(self) -> Dict[str, Any]: data = asdict(self) data['type'] = self.type.value data['priority'] = self.priority.value data['generated_at'] = self.generated_at.isoformat() if self.expires_at: data['expires_at'] = self.expires_at.isoformat() return data @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ProactiveInsight': data = data.copy() data['type'] = InsightType(data['type']) data['priority'] = InsightPriority(data['priority']) data['generated_at'] = datetime.fromisoformat(data['generated_at']) if data.get('expires_at'): data['expires_at'] = datetime.fromisoformat(data['expires_at']) return cls(**data) class PatternAnalyzer: """Analyzes various data sources for patterns""" def __init__(self): self.memory_manager = MemoryManager() self.indexer = ComprehensiveIndexer() def analyze_work_patterns(self) -> List[Dict[str, Any]]: """Analyze work patterns for insights""" patterns = [] try: work_intelligence = get_work_context_intelligence() work_analysis = work_intelligence.analyze_current_context(hours_back=168) # 1 week # Analyze productivity patterns if work_analysis.get('momentum_indicators'): momentum = work_analysis['momentum_indicators'] if momentum.get('momentum_score', 0) < 3: patterns.append({ 'type': 'low_productivity', 'severity': 'medium', 'evidence': f"Momentum score: {momentum.get('momentum_score', 0)}", 'data': momentum }) if momentum.get('pending_thread_count', 0) > 10: patterns.append({ 'type': 'task_overload', 'severity': 'high', 'evidence': f"Pending items: {momentum.get('pending_thread_count', 0)}", 'data': momentum }) # Analyze project diversity active_projects = work_analysis.get('active_projects', {}) if len(active_projects) > 5: patterns.append({ 'type': 'project_fragmentation', 'severity': 'medium', 'evidence': f"Active projects: {len(active_projects)}", 'data': {'project_count': len(active_projects), 'projects': list(active_projects.keys())} }) except Exception as e: logger.error(f"Error analyzing work patterns: {e}") return patterns def analyze_emotional_patterns(self) -> List[Dict[str, Any]]: """Analyze emotional patterns for insights""" patterns = [] try: emotional_tracker = get_emotional_memory_tracker() journey = emotional_tracker.analyze_emotional_journey(days_back=14) if journey.get('total_emotional_memories', 0) > 0: # Check for emotional imbalance emotion_dist = journey.get('emotion_distribution', {}) total_emotions = sum(emotion_dist.values()) # High frustration pattern frustration_ratio = emotion_dist.get('frustration', 0) / max(1, total_emotions) if frustration_ratio > 0.4: patterns.append({ 'type': 'high_frustration', 'severity': 'high', 'evidence': f"Frustration comprises {frustration_ratio:.1%} of emotional memories", 'data': {'frustration_ratio': frustration_ratio, 'distribution': emotion_dist} }) # Low emotional diversity if journey.get('emotional_diversity', 0) <= 2: patterns.append({ 'type': 'low_emotional_diversity', 'severity': 'medium', 'evidence': f"Only {journey.get('emotional_diversity', 0)} different emotions detected", 'data': journey }) # High emotional resonance (positive pattern) if journey.get('average_resonance', 0) > 0.7: patterns.append({ 'type': 'high_emotional_engagement', 'severity': 'positive', 'evidence': f"Average emotional resonance: {journey.get('average_resonance', 0):.1%}", 'data': journey }) except Exception as e: logger.error(f"Error analyzing emotional patterns: {e}") return patterns def analyze_learning_patterns(self) -> List[Dict[str, Any]]: """Analyze learning and knowledge patterns""" patterns = [] try: # Analyze memory types and learning progression memories_dir = os.path.join(get_mira_memory_dir(), 'memories') if os.path.exists(memories_dir): memory_types = defaultdict(int) learning_keywords = [] for filename in os.listdir(memories_dir): if filename.endswith('.json'): try: with open(os.path.join(memories_dir, filename), 'r') as f: memory = json.load(f) memory_type = memory.get('type', 'unknown') memory_types[memory_type] += 1 content = memory.get('content', '').lower() if any(word in content for word in ['learned', 'discovered', 'understood', 'realized']): learning_keywords.append(memory) except Exception: continue # Check for learning stagnation if memory_types.get('learning', 0) + memory_types.get('insight', 0) < 2: patterns.append({ 'type': 'learning_stagnation', 'severity': 'medium', 'evidence': f"Few learning-type memories: {memory_types.get('learning', 0) + memory_types.get('insight', 0)}", 'data': {'memory_types': dict(memory_types)} }) # Positive learning pattern if len(learning_keywords) > 5: patterns.append({ 'type': 'active_learning', 'severity': 'positive', 'evidence': f"Multiple learning instances detected: {len(learning_keywords)}", 'data': {'learning_count': len(learning_keywords)} }) except Exception as e: logger.error(f"Error analyzing learning patterns: {e}") return patterns def analyze_collaboration_patterns(self) -> List[Dict[str, Any]]: """Analyze collaboration and relationship patterns""" patterns = [] try: relationship_tracker = get_relationship_evolution_tracker() current_state = relationship_tracker.get_current_relationship_state() if 'current_snapshot' in current_state: snapshot = current_state['current_snapshot'] # Check for relationship issues if snapshot.get('trust_level', 0) < 0.5: patterns.append({ 'type': 'low_trust', 'severity': 'high', 'evidence': f"Trust level: {snapshot.get('trust_level', 0):.1%}", 'data': snapshot }) if snapshot.get('communication_comfort', 0) < 0.4: patterns.append({ 'type': 'communication_barriers', 'severity': 'medium', 'evidence': f"Communication comfort: {snapshot.get('communication_comfort', 0):.1%}", 'data': snapshot }) # Positive collaboration patterns if snapshot.get('technical_partnership_level', 0) > 0.7: patterns.append({ 'type': 'strong_technical_collaboration', 'severity': 'positive', 'evidence': f"Technical partnership: {snapshot.get('technical_partnership_level', 0):.1%}", 'data': snapshot }) except Exception as e: logger.error(f"Error analyzing collaboration patterns: {e}") return patterns class InsightGenerator: """Generates actionable insights from detected patterns""" def __init__(self): self.pattern_analyzer = PatternAnalyzer() def generate_insights(self) -> List[ProactiveInsight]: """Generate all types of insights""" insights = [] # Analyze all pattern types work_patterns = self.pattern_analyzer.analyze_work_patterns() emotional_patterns = self.pattern_analyzer.analyze_emotional_patterns() learning_patterns = self.pattern_analyzer.analyze_learning_patterns() collaboration_patterns = self.pattern_analyzer.analyze_collaboration_patterns() # Generate insights from patterns insights.extend(self._generate_work_insights(work_patterns)) insights.extend(self._generate_emotional_insights(emotional_patterns)) insights.extend(self._generate_learning_insights(learning_patterns)) insights.extend(self._generate_collaboration_insights(collaboration_patterns)) # Generate meta-insights (insights about insights) insights.extend(self._generate_meta_insights(insights)) return insights def _generate_work_insights(self, patterns: List[Dict[str, Any]]) -> List[ProactiveInsight]: """Generate work-related insights""" insights = [] for pattern in patterns: if pattern['type'] == 'low_productivity': insight = ProactiveInsight( id=f"work_productivity_{datetime.now().strftime('%Y%m%d')}", type=InsightType.PRODUCTIVITY_INSIGHT, priority=InsightPriority.MEDIUM, title="Low Productivity Momentum Detected", description="Your work momentum has decreased significantly. This could indicate burnout, unclear priorities, or external blockers.", evidence=[ pattern['evidence'], "Multiple pending items without resolution", "Decreased activity in recent work sessions" ], recommendations=[ "Review and prioritize your current task list", "Consider breaking large tasks into smaller, manageable pieces", "Schedule focused work blocks without interruptions", "Take a break if feeling overwhelmed" ], confidence=0.7, relevance_score=0.8, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=7), tags=['productivity', 'momentum', 'work'], metadata=pattern.get('data', {}) ) insights.append(insight) elif pattern['type'] == 'task_overload': insight = ProactiveInsight( id=f"task_overload_{datetime.now().strftime('%Y%m%d')}", type=InsightType.RISK_WARNING, priority=InsightPriority.HIGH, title="Task Overload Warning", description="You have an unusually high number of pending tasks, which may lead to stress and reduced quality of work.", evidence=[ pattern['evidence'], "High cognitive load from context switching", "Risk of important tasks being overlooked" ], recommendations=[ "Conduct a task audit - identify what can be delegated, deferred, or deleted", "Use the Eisenhower Matrix to prioritize tasks", "Set boundaries on new task acceptance", "Consider using time-boxing techniques" ], confidence=0.8, relevance_score=0.9, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=3), tags=['task_management', 'overload', 'stress'], metadata=pattern.get('data', {}) ) insights.append(insight) elif pattern['type'] == 'project_fragmentation': insight = ProactiveInsight( id=f"project_frag_{datetime.now().strftime('%Y%m%d')}", type=InsightType.EFFICIENCY_OPTIMIZATION, priority=InsightPriority.MEDIUM, title="Project Fragmentation Detected", description="You're working on many projects simultaneously, which may reduce focus and efficiency.", evidence=[ pattern['evidence'], "Increased context switching overhead", "Potential for reduced deep work" ], recommendations=[ "Consider consolidating or pausing some projects", "Implement time-blocking for different projects", "Create clear project boundaries and goals", "Use project batching techniques" ], confidence=0.6, relevance_score=0.7, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=14), tags=['project_management', 'focus', 'efficiency'], metadata=pattern.get('data', {}) ) insights.append(insight) return insights def _generate_emotional_insights(self, patterns: List[Dict[str, Any]]) -> List[ProactiveInsight]: """Generate emotional wellbeing insights""" insights = [] for pattern in patterns: if pattern['type'] == 'high_frustration': insight = ProactiveInsight( id=f"high_frustration_{datetime.now().strftime('%Y%m%d')}", type=InsightType.EMOTIONAL_WELLBEING, priority=InsightPriority.HIGH, title="Elevated Frustration Levels", description="A significant portion of your recent emotional experiences involve frustration, which may impact wellbeing and productivity.", evidence=[ pattern['evidence'], "Pattern of struggle with technical challenges", "Potential stress accumulation" ], recommendations=[ "Identify common sources of frustration and address them systematically", "Consider pair programming or collaboration for challenging problems", "Implement stress management techniques (breaks, exercise, mindfulness)", "Celebrate small wins to balance frustration with achievement" ], confidence=0.8, relevance_score=0.9, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=7), tags=['emotional_health', 'frustration', 'wellbeing'], metadata=pattern.get('data', {}) ) insights.append(insight) elif pattern['type'] == 'high_emotional_engagement': insight = ProactiveInsight( id=f"emotional_engagement_{datetime.now().strftime('%Y%m%d')}", type=InsightType.SUCCESS_PATTERN, priority=InsightPriority.MEDIUM, title="High Emotional Investment in Work", description="Your work generates strong emotional responses, indicating high engagement and personal investment.", evidence=[ pattern['evidence'], "Strong emotional connection to outcomes", "Deep personal investment in projects" ], recommendations=[ "Leverage this engagement for creative and innovative work", "Share your passion with team members to inspire others", "Balance emotional investment with self-care", "Document emotional insights for future reference" ], confidence=0.7, relevance_score=0.8, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=30), tags=['engagement', 'passion', 'positive_pattern'], metadata=pattern.get('data', {}) ) insights.append(insight) return insights def _generate_learning_insights(self, patterns: List[Dict[str, Any]]) -> List[ProactiveInsight]: """Generate learning and knowledge insights""" insights = [] for pattern in patterns: if pattern['type'] == 'learning_stagnation': insight = ProactiveInsight( id=f"learning_stagnation_{datetime.now().strftime('%Y%m%d')}", type=InsightType.LEARNING_OPPORTUNITY, priority=InsightPriority.MEDIUM, title="Learning Activity Appears Low", description="Recent memory patterns suggest fewer learning and insight moments than typical for active development.", evidence=[ pattern['evidence'], "Limited knowledge acquisition patterns", "Fewer 'aha' moments recorded" ], recommendations=[ "Explore new technologies or approaches in your field", "Engage with learning communities or forums", "Take on a challenging problem that requires new skills", "Document learning experiences more actively" ], confidence=0.6, relevance_score=0.7, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=21), tags=['learning', 'growth', 'development'], metadata=pattern.get('data', {}) ) insights.append(insight) elif pattern['type'] == 'active_learning': insight = ProactiveInsight( id=f"active_learning_{datetime.now().strftime('%Y%m%d')}", type=InsightType.SUCCESS_PATTERN, priority=InsightPriority.LOW, title="Strong Learning Pattern Detected", description="You're demonstrating excellent learning momentum with frequent knowledge acquisition and insights.", evidence=[ pattern['evidence'], "Regular discovery and understanding moments", "Active knowledge building pattern" ], recommendations=[ "Consider sharing your learnings with others", "Document insights for future reference", "Connect new learnings to existing knowledge", "Explore teaching or mentoring opportunities" ], confidence=0.8, relevance_score=0.7, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=30), tags=['learning', 'growth', 'positive_pattern'], metadata=pattern.get('data', {}) ) insights.append(insight) return insights def _generate_collaboration_insights(self, patterns: List[Dict[str, Any]]) -> List[ProactiveInsight]: """Generate collaboration and relationship insights""" insights = [] for pattern in patterns: if pattern['type'] == 'low_trust': insight = ProactiveInsight( id=f"low_trust_{datetime.now().strftime('%Y%m%d')}", type=InsightType.COLLABORATION_IMPROVEMENT, priority=InsightPriority.HIGH, title="Trust Level Needs Attention", description="The collaboration relationship shows lower trust levels, which may impact communication and effectiveness.", evidence=[ pattern['evidence'], "Limited vulnerability and openness", "Cautious communication patterns" ], recommendations=[ "Focus on consistent, reliable assistance", "Be transparent about capabilities and limitations", "Follow through on commitments", "Encourage open feedback and questions" ], confidence=0.7, relevance_score=0.8, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=14), tags=['trust', 'collaboration', 'relationship'], metadata=pattern.get('data', {}) ) insights.append(insight) elif pattern['type'] == 'strong_technical_collaboration': insight = ProactiveInsight( id=f"tech_collab_{datetime.now().strftime('%Y%m%d')}", type=InsightType.SUCCESS_PATTERN, priority=InsightPriority.LOW, title="Excellent Technical Partnership", description="You've developed a strong technical collaboration that enables effective problem-solving and learning.", evidence=[ pattern['evidence'], "Effective technical communication", "Successful joint problem-solving" ], recommendations=[ "Leverage this partnership for complex technical challenges", "Explore advanced topics together", "Document successful collaboration patterns", "Share this model with other partnerships" ], confidence=0.8, relevance_score=0.7, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=60), tags=['technical', 'collaboration', 'partnership'], metadata=pattern.get('data', {}) ) insights.append(insight) return insights def _generate_meta_insights(self, insights: List[ProactiveInsight]) -> List[ProactiveInsight]: """Generate insights about the insights themselves""" meta_insights = [] if len(insights) == 0: # No patterns detected - could be good or concerning meta_insight = ProactiveInsight( id=f"no_patterns_{datetime.now().strftime('%Y%m%d')}", type=InsightType.PATTERN_DISCOVERY, priority=InsightPriority.LOW, title="Stable Patterns Detected", description="No significant patterns or issues detected in recent analysis, suggesting stable work and emotional patterns.", evidence=[ "No concerning patterns in work productivity", "Emotional balance appears stable", "Learning and collaboration within normal ranges" ], recommendations=[ "Continue current practices that are working well", "Consider proactive improvements during this stable period", "Document what's working for future reference" ], confidence=0.6, relevance_score=0.5, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=7), tags=['meta', 'stability', 'maintenance'] ) meta_insights.append(meta_insight) elif len([i for i in insights if i.priority in [InsightPriority.CRITICAL, InsightPriority.HIGH]]) > 2: # Multiple high-priority issues meta_insight = ProactiveInsight( id=f"multiple_issues_{datetime.now().strftime('%Y%m%d')}", type=InsightType.RISK_WARNING, priority=InsightPriority.HIGH, title="Multiple Areas Need Attention", description="Several high-priority patterns have been detected across different areas, suggesting a need for systematic attention.", evidence=[ f"Multiple high-priority insights: {len([i for i in insights if i.priority == InsightPriority.HIGH])}", "Issues span different domains (work, emotional, collaboration)", "Potential for compound effects" ], recommendations=[ "Prioritize addressing the most critical issues first", "Look for common root causes across different problems", "Consider seeking additional support or resources", "Create a systematic improvement plan" ], confidence=0.8, relevance_score=0.9, generated_at=datetime.now(), expires_at=datetime.now() + timedelta(days=3), tags=['meta', 'multiple_issues', 'systematic'] ) meta_insights.append(meta_insight) return meta_insights class ProactiveInsightManager: """Manages the generation, storage, and retrieval of proactive insights""" def __init__(self): self.memory_dir = get_mira_memory_dir() self.insights_file = os.path.join(self.memory_dir, 'proactive_insights.json') self.generator = InsightGenerator() def generate_fresh_insights(self) -> List[ProactiveInsight]: """Generate new insights and store them""" try: # Generate new insights new_insights = self.generator.generate_insights() # Load existing insights existing_insights = self._load_insights() # Remove expired insights current_time = datetime.now() active_insights = [ insight for insight in existing_insights if not insight.expires_at or insight.expires_at > current_time ] # Deduplicate by ID and add new insights insight_ids = {insight.id for insight in active_insights} for insight in new_insights: if insight.id not in insight_ids: active_insights.append(insight) # Save updated insights self._save_insights(active_insights) logger.info(f"Generated {len(new_insights)} new insights, {len(active_insights)} total active") return new_insights except Exception as e: logger.error(f"Error generating fresh insights: {e}") return [] def get_active_insights(self, priority_filter: List[InsightPriority] = None, type_filter: List[InsightType] = None) -> List[ProactiveInsight]: """Get currently active insights with optional filtering""" try: insights = self._load_insights() current_time = datetime.now() # Filter out expired insights active_insights = [ insight for insight in insights if not insight.expires_at or insight.expires_at > current_time ] # Apply filters if priority_filter: active_insights = [i for i in active_insights if i.priority in priority_filter] if type_filter: active_insights = [i for i in active_insights if i.type in type_filter] # Sort by priority and relevance priority_order = { InsightPriority.CRITICAL: 4, InsightPriority.HIGH: 3, InsightPriority.MEDIUM: 2, InsightPriority.LOW: 1 } active_insights.sort( key=lambda x: (priority_order[x.priority], x.relevance_score, x.confidence), reverse=True ) return active_insights except Exception as e: logger.error(f"Error getting active insights: {e}") return [] def dismiss_insight(self, insight_id: str) -> bool: """Dismiss an insight (mark as expired)""" try: insights = self._load_insights() for insight in insights: if insight.id == insight_id: insight.expires_at = datetime.now() break self._save_insights(insights) return True except Exception as e: logger.error(f"Error dismissing insight {insight_id}: {e}") return False def _load_insights(self) -> List[ProactiveInsight]: """Load insights from storage""" try: if os.path.exists(self.insights_file): with open(self.insights_file, 'r') as f: data = json.load(f) return [ProactiveInsight.from_dict(insight_data) for insight_data in data] return [] except Exception as e: logger.error(f"Error loading insights: {e}") return [] def _save_insights(self, insights: List[ProactiveInsight]): """Save insights to storage""" try: data = [insight.to_dict() for insight in insights] with open(self.insights_file, 'w') as f: json.dump(data, f, indent=2) except Exception as e: logger.error(f"Error saving insights: {e}") def get_proactive_insight_manager() -> ProactiveInsightManager: """Get the global proactive insight manager instance""" return ProactiveInsightManager() if __name__ == "__main__": # Test the proactive insight generation manager = get_proactive_insight_manager() print("Testing proactive insight generation...") new_insights = manager.generate_fresh_insights() print(f"Generated {len(new_insights)} new insights:") for insight in new_insights: print(f"- {insight.title} ({insight.priority.value}, {insight.confidence:.2f})") print("\nActive insights:") active_insights = manager.get_active_insights() for insight in active_insights: print(f"- {insight.title}") print(f" Priority: {insight.priority.value}, Type: {insight.type.value}") print(f" Description: {insight.description}") print()