autonomy phase 2
This commit is contained in:
serversdwn
1
cortex/autonomy/learning/__init__.py
Normal file
1
cortex/autonomy/learning/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Pattern learning and adaptation system."""
|
||||
383
cortex/autonomy/learning/pattern_learner.py
Normal file
383
cortex/autonomy/learning/pattern_learner.py
Normal file
@@ -0,0 +1,383 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user