384 lines
12 KiB
Python
384 lines
12 KiB
Python
"""
|
|
Pattern Learning System - learns from interaction patterns to improve autonomy.
|
|
"""
|
|
|
|
import logging
|
|
import json
|
|
import os
|
|
from typing import Dict, List, Any, Optional
|
|
from datetime import datetime
|
|
from collections import defaultdict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PatternLearner:
|
|
"""
|
|
Learns from interaction patterns to improve Lyra's autonomous behavior.
|
|
|
|
Tracks:
|
|
- Topic frequencies (what users talk about)
|
|
- Time-of-day patterns (when users interact)
|
|
- User preferences (how users like responses)
|
|
- Successful response strategies (what works well)
|
|
"""
|
|
|
|
def __init__(self, patterns_file: str = "/app/data/learned_patterns.json"):
|
|
"""
|
|
Initialize pattern learner.
|
|
|
|
Args:
|
|
patterns_file: Path to persistent patterns storage
|
|
"""
|
|
self.patterns_file = patterns_file
|
|
self.patterns = self._load_patterns()
|
|
|
|
def _load_patterns(self) -> Dict[str, Any]:
|
|
"""Load patterns from disk."""
|
|
if os.path.exists(self.patterns_file):
|
|
try:
|
|
with open(self.patterns_file, 'r') as f:
|
|
patterns = json.load(f)
|
|
logger.info(f"[PATTERN_LEARNER] Loaded patterns from {self.patterns_file}")
|
|
return patterns
|
|
except Exception as e:
|
|
logger.error(f"[PATTERN_LEARNER] Failed to load patterns: {e}")
|
|
|
|
# Initialize empty patterns
|
|
return {
|
|
"topic_frequencies": {},
|
|
"time_patterns": {},
|
|
"user_preferences": {},
|
|
"successful_strategies": {},
|
|
"interaction_count": 0,
|
|
"last_updated": datetime.utcnow().isoformat()
|
|
}
|
|
|
|
def _save_patterns(self) -> None:
|
|
"""Save patterns to disk."""
|
|
try:
|
|
# Ensure directory exists
|
|
os.makedirs(os.path.dirname(self.patterns_file), exist_ok=True)
|
|
|
|
self.patterns["last_updated"] = datetime.utcnow().isoformat()
|
|
|
|
with open(self.patterns_file, 'w') as f:
|
|
json.dump(self.patterns, f, indent=2)
|
|
|
|
logger.debug(f"[PATTERN_LEARNER] Saved patterns to {self.patterns_file}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"[PATTERN_LEARNER] Failed to save patterns: {e}")
|
|
|
|
async def learn_from_interaction(
|
|
self,
|
|
user_prompt: str,
|
|
response: str,
|
|
monologue: Dict[str, Any],
|
|
context: Dict[str, Any]
|
|
) -> None:
|
|
"""
|
|
Learn from a single interaction.
|
|
|
|
Args:
|
|
user_prompt: User's message
|
|
response: Lyra's response
|
|
monologue: Inner monologue analysis
|
|
context: Full context state
|
|
"""
|
|
self.patterns["interaction_count"] += 1
|
|
|
|
# Learn topic frequencies
|
|
self._learn_topics(user_prompt, monologue)
|
|
|
|
# Learn time patterns
|
|
self._learn_time_patterns()
|
|
|
|
# Learn user preferences
|
|
self._learn_preferences(monologue, context)
|
|
|
|
# Learn successful strategies
|
|
self._learn_strategies(monologue, response, context)
|
|
|
|
# Save periodically (every 10 interactions)
|
|
if self.patterns["interaction_count"] % 10 == 0:
|
|
self._save_patterns()
|
|
|
|
def _learn_topics(self, user_prompt: str, monologue: Dict[str, Any]) -> None:
|
|
"""Track topic frequencies."""
|
|
intent = monologue.get("intent", "unknown")
|
|
|
|
# Increment topic counter
|
|
topic_freq = self.patterns["topic_frequencies"]
|
|
topic_freq[intent] = topic_freq.get(intent, 0) + 1
|
|
|
|
# Extract keywords (simple approach - words > 5 chars)
|
|
keywords = [word.lower() for word in user_prompt.split() if len(word) > 5]
|
|
|
|
for keyword in keywords:
|
|
topic_freq[f"keyword:{keyword}"] = topic_freq.get(f"keyword:{keyword}", 0) + 1
|
|
|
|
logger.debug(f"[PATTERN_LEARNER] Topic learned: {intent}")
|
|
|
|
def _learn_time_patterns(self) -> None:
|
|
"""Track time-of-day patterns."""
|
|
now = datetime.utcnow()
|
|
hour = now.hour
|
|
|
|
# Track interactions by hour
|
|
time_patterns = self.patterns["time_patterns"]
|
|
hour_key = f"hour_{hour:02d}"
|
|
time_patterns[hour_key] = time_patterns.get(hour_key, 0) + 1
|
|
|
|
# Track day of week
|
|
day_key = f"day_{now.strftime('%A').lower()}"
|
|
time_patterns[day_key] = time_patterns.get(day_key, 0) + 1
|
|
|
|
def _learn_preferences(self, monologue: Dict[str, Any], context: Dict[str, Any]) -> None:
|
|
"""Learn user preferences from detected tone and depth."""
|
|
tone = monologue.get("tone", "neutral")
|
|
depth = monologue.get("depth", "medium")
|
|
|
|
prefs = self.patterns["user_preferences"]
|
|
|
|
# Track preferred tone
|
|
prefs.setdefault("tone_counts", {})
|
|
prefs["tone_counts"][tone] = prefs["tone_counts"].get(tone, 0) + 1
|
|
|
|
# Track preferred depth
|
|
prefs.setdefault("depth_counts", {})
|
|
prefs["depth_counts"][depth] = prefs["depth_counts"].get(depth, 0) + 1
|
|
|
|
def _learn_strategies(
|
|
self,
|
|
monologue: Dict[str, Any],
|
|
response: str,
|
|
context: Dict[str, Any]
|
|
) -> None:
|
|
"""
|
|
Learn which response strategies are successful.
|
|
|
|
Success indicators:
|
|
- Executive was consulted and plan generated
|
|
- Response length matches depth request
|
|
- Tone matches request
|
|
"""
|
|
intent = monologue.get("intent", "unknown")
|
|
executive_used = context.get("executive_plan") is not None
|
|
|
|
strategies = self.patterns["successful_strategies"]
|
|
strategies.setdefault(intent, {})
|
|
|
|
# Track executive usage for this intent
|
|
if executive_used:
|
|
key = f"{intent}:executive_used"
|
|
strategies.setdefault(key, 0)
|
|
strategies[key] += 1
|
|
|
|
# Track response length patterns
|
|
response_length = len(response.split())
|
|
depth = monologue.get("depth", "medium")
|
|
|
|
length_key = f"{depth}:avg_words"
|
|
if length_key not in strategies:
|
|
strategies[length_key] = response_length
|
|
else:
|
|
# Running average
|
|
strategies[length_key] = (strategies[length_key] + response_length) / 2
|
|
|
|
# ========================================
|
|
# Pattern Analysis and Recommendations
|
|
# ========================================
|
|
|
|
def get_top_topics(self, limit: int = 10) -> List[tuple]:
|
|
"""
|
|
Get most frequent topics.
|
|
|
|
Args:
|
|
limit: Max number of topics to return
|
|
|
|
Returns:
|
|
List of (topic, count) tuples, sorted by count
|
|
"""
|
|
topics = self.patterns["topic_frequencies"]
|
|
sorted_topics = sorted(topics.items(), key=lambda x: x[1], reverse=True)
|
|
return sorted_topics[:limit]
|
|
|
|
def get_preferred_tone(self) -> str:
|
|
"""
|
|
Get user's most preferred tone.
|
|
|
|
Returns:
|
|
Preferred tone string
|
|
"""
|
|
prefs = self.patterns["user_preferences"]
|
|
tone_counts = prefs.get("tone_counts", {})
|
|
|
|
if not tone_counts:
|
|
return "neutral"
|
|
|
|
return max(tone_counts.items(), key=lambda x: x[1])[0]
|
|
|
|
def get_preferred_depth(self) -> str:
|
|
"""
|
|
Get user's most preferred response depth.
|
|
|
|
Returns:
|
|
Preferred depth string
|
|
"""
|
|
prefs = self.patterns["user_preferences"]
|
|
depth_counts = prefs.get("depth_counts", {})
|
|
|
|
if not depth_counts:
|
|
return "medium"
|
|
|
|
return max(depth_counts.items(), key=lambda x: x[1])[0]
|
|
|
|
def get_peak_hours(self, limit: int = 3) -> List[int]:
|
|
"""
|
|
Get peak interaction hours.
|
|
|
|
Args:
|
|
limit: Number of top hours to return
|
|
|
|
Returns:
|
|
List of hours (0-23)
|
|
"""
|
|
time_patterns = self.patterns["time_patterns"]
|
|
hour_counts = {k: v for k, v in time_patterns.items() if k.startswith("hour_")}
|
|
|
|
if not hour_counts:
|
|
return []
|
|
|
|
sorted_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)
|
|
top_hours = sorted_hours[:limit]
|
|
|
|
# Extract hour numbers
|
|
return [int(h[0].split("_")[1]) for h in top_hours]
|
|
|
|
def should_use_executive(self, intent: str) -> bool:
|
|
"""
|
|
Recommend whether to use executive for given intent based on patterns.
|
|
|
|
Args:
|
|
intent: Intent type
|
|
|
|
Returns:
|
|
True if executive is recommended
|
|
"""
|
|
strategies = self.patterns["successful_strategies"]
|
|
key = f"{intent}:executive_used"
|
|
|
|
# If we've used executive for this intent >= 3 times, recommend it
|
|
return strategies.get(key, 0) >= 3
|
|
|
|
def get_recommended_response_length(self, depth: str) -> int:
|
|
"""
|
|
Get recommended response length in words for given depth.
|
|
|
|
Args:
|
|
depth: Depth level (short/medium/deep)
|
|
|
|
Returns:
|
|
Recommended word count
|
|
"""
|
|
strategies = self.patterns["successful_strategies"]
|
|
key = f"{depth}:avg_words"
|
|
|
|
avg_length = strategies.get(key, None)
|
|
|
|
if avg_length:
|
|
return int(avg_length)
|
|
|
|
# Defaults if no pattern learned
|
|
defaults = {
|
|
"short": 50,
|
|
"medium": 150,
|
|
"deep": 300
|
|
}
|
|
|
|
return defaults.get(depth, 150)
|
|
|
|
def get_insights(self) -> Dict[str, Any]:
|
|
"""
|
|
Get high-level insights from learned patterns.
|
|
|
|
Returns:
|
|
{
|
|
"total_interactions": int,
|
|
"top_topics": [(topic, count), ...],
|
|
"preferred_tone": str,
|
|
"preferred_depth": str,
|
|
"peak_hours": [hours],
|
|
"learning_recommendations": [str]
|
|
}
|
|
"""
|
|
recommendations = []
|
|
|
|
# Check if user consistently prefers certain settings
|
|
preferred_tone = self.get_preferred_tone()
|
|
preferred_depth = self.get_preferred_depth()
|
|
|
|
if preferred_tone != "neutral":
|
|
recommendations.append(f"User prefers {preferred_tone} tone")
|
|
|
|
if preferred_depth != "medium":
|
|
recommendations.append(f"User prefers {preferred_depth} depth responses")
|
|
|
|
# Check for recurring topics
|
|
top_topics = self.get_top_topics(limit=3)
|
|
if top_topics:
|
|
top_topic = top_topics[0][0]
|
|
recommendations.append(f"Consider adding '{top_topic}' to learning queue")
|
|
|
|
return {
|
|
"total_interactions": self.patterns["interaction_count"],
|
|
"top_topics": self.get_top_topics(limit=5),
|
|
"preferred_tone": preferred_tone,
|
|
"preferred_depth": preferred_depth,
|
|
"peak_hours": self.get_peak_hours(limit=3),
|
|
"learning_recommendations": recommendations
|
|
}
|
|
|
|
def reset_patterns(self) -> None:
|
|
"""Reset all learned patterns (use with caution)."""
|
|
self.patterns = {
|
|
"topic_frequencies": {},
|
|
"time_patterns": {},
|
|
"user_preferences": {},
|
|
"successful_strategies": {},
|
|
"interaction_count": 0,
|
|
"last_updated": datetime.utcnow().isoformat()
|
|
}
|
|
self._save_patterns()
|
|
logger.warning("[PATTERN_LEARNER] Patterns reset")
|
|
|
|
def export_patterns(self) -> Dict[str, Any]:
|
|
"""
|
|
Export all patterns for analysis.
|
|
|
|
Returns:
|
|
Complete patterns dict
|
|
"""
|
|
return self.patterns.copy()
|
|
|
|
|
|
# Singleton instance
|
|
_learner_instance = None
|
|
|
|
|
|
def get_pattern_learner(patterns_file: str = "/app/data/learned_patterns.json") -> PatternLearner:
|
|
"""
|
|
Get singleton pattern learner instance.
|
|
|
|
Args:
|
|
patterns_file: Path to patterns file (only used on first call)
|
|
|
|
Returns:
|
|
PatternLearner instance
|
|
"""
|
|
global _learner_instance
|
|
if _learner_instance is None:
|
|
_learner_instance = PatternLearner(patterns_file=patterns_file)
|
|
return _learner_instance
|