tuner.py
1 """ 2 Resonance Weight Tuner 3 4 Learns optimal resonance weights per operator based on their 5 selection behavior. When operators consistently choose bullets 6 that score low on a factor, that factor's weight decreases. 7 8 This implements implicit preference learning - the system adapts 9 to each operator's actual behavior without explicit configuration. 10 """ 11 12 from dataclasses import dataclass, field 13 from datetime import datetime, timedelta 14 from typing import Optional, List, Dict, Any, Tuple 15 import math 16 17 from .scoring import ResonanceScoringEngine, ResonanceScore, ResonanceFactors 18 19 20 @dataclass 21 class SelectionEvent: 22 """ 23 Records when an operator selects a bullet from presented options. 24 25 Used to learn which resonance factors matter to this operator. 26 """ 27 selected_uuid: str 28 presented_uuids: List[str] # All bullets that were shown 29 selected_score: ResonanceScore 30 presented_scores: List[ResonanceScore] 31 timestamp: datetime = field(default_factory=datetime.now) 32 context: Optional[Dict[str, Any]] = None 33 34 35 @dataclass 36 class FactorImportance: 37 """ 38 Learned importance of a resonance factor for an operator. 39 """ 40 factor_name: str 41 current_weight: float 42 selection_correlation: float # How often high factor → selection 43 adjustment_history: List[Tuple[datetime, float]] = field(default_factory=list) 44 45 46 class ResonanceWeightTuner: 47 """ 48 Learns and adjusts resonance weights based on operator behavior. 49 50 The tuner observes selection events and adjusts weights to better 51 predict what the operator will find relevant. 52 53 Key principle: If an operator consistently selects bullets that 54 score high on a particular factor, that factor becomes more important. 55 Conversely, if they ignore high-scoring bullets, the factor loses weight. 56 """ 57 58 # How quickly weights adapt (0-1, higher = faster adaptation) 59 DEFAULT_LEARNING_RATE = 0.1 60 61 # Minimum weight for any factor (prevent complete elimination) 62 MIN_WEIGHT = 0.01 63 64 # Maximum weight for any factor (prevent dominance) 65 MAX_WEIGHT = 0.5 66 67 # How many recent events to consider 68 WINDOW_SIZE = 100 69 70 def __init__( 71 self, 72 scoring_engine: ResonanceScoringEngine, 73 operator_id: str, 74 learning_rate: float = DEFAULT_LEARNING_RATE 75 ): 76 """ 77 Initialize the weight tuner. 78 79 Args: 80 scoring_engine: The resonance scoring engine to tune 81 operator_id: Identifier for this operator's preferences 82 learning_rate: How quickly to adapt weights 83 """ 84 self.scoring_engine = scoring_engine 85 self.operator_id = operator_id 86 self.learning_rate = learning_rate 87 88 # Selection history 89 self.selection_events: List[SelectionEvent] = [] 90 91 # Track factor correlations 92 self.factor_importance: Dict[str, FactorImportance] = {} 93 self._initialize_factor_importance() 94 95 def _initialize_factor_importance(self): 96 """Initialize factor importance tracking from current weights.""" 97 for factor, weight in self.scoring_engine.weights.items(): 98 self.factor_importance[factor] = FactorImportance( 99 factor_name=factor, 100 current_weight=weight, 101 selection_correlation=0.5, # Start neutral 102 adjustment_history=[(datetime.now(), weight)] 103 ) 104 105 def record_selection( 106 self, 107 selected_uuid: str, 108 presented_uuids: List[str], 109 selected_score: ResonanceScore, 110 presented_scores: List[ResonanceScore], 111 context: Optional[Dict[str, Any]] = None 112 ): 113 """ 114 Record a selection event and update weights. 115 116 Args: 117 selected_uuid: UUID of the bullet the operator selected 118 presented_uuids: All UUIDs that were presented 119 selected_score: Resonance score of selected bullet 120 presented_scores: Scores of all presented bullets 121 context: Optional context about the selection 122 """ 123 event = SelectionEvent( 124 selected_uuid=selected_uuid, 125 presented_uuids=presented_uuids, 126 selected_score=selected_score, 127 presented_scores=presented_scores, 128 timestamp=datetime.now(), 129 context=context 130 ) 131 132 self.selection_events.append(event) 133 134 # Keep window size bounded 135 if len(self.selection_events) > self.WINDOW_SIZE * 2: 136 self.selection_events = self.selection_events[-self.WINDOW_SIZE:] 137 138 # Update weights based on this selection 139 self._update_weights(event) 140 141 def _update_weights(self, event: SelectionEvent): 142 """ 143 Update resonance weights based on a selection event. 144 145 The logic: 146 - For each factor, compare the selected bullet's score to the average 147 - If selected bullet scores above average on a factor, increase weight 148 - If selected bullet scores below average, decrease weight 149 - Normalize weights to sum to 1.0 150 """ 151 selected_factors = event.selected_score.factors 152 153 # Compute average factor values across all presented 154 avg_factors: Dict[str, float] = {} 155 for factor in self.scoring_engine.weights.keys(): 156 values = [getattr(s.factors, factor) for s in event.presented_scores] 157 avg_factors[factor] = sum(values) / len(values) if values else 0.5 158 159 # Compute adjustment for each factor 160 adjustments: Dict[str, float] = {} 161 for factor in self.scoring_engine.weights.keys(): 162 selected_value = getattr(selected_factors, factor) 163 avg_value = avg_factors[factor] 164 165 # Positive if selected was above average, negative if below 166 delta = selected_value - avg_value 167 168 # Scale by learning rate 169 adjustment = delta * self.learning_rate 170 adjustments[factor] = adjustment 171 172 # Apply adjustments 173 new_weights = {} 174 for factor, current_weight in self.scoring_engine.weights.items(): 175 new_weight = current_weight + adjustments[factor] 176 # Clamp to bounds 177 new_weight = max(self.MIN_WEIGHT, min(self.MAX_WEIGHT, new_weight)) 178 new_weights[factor] = new_weight 179 180 # Record in history 181 self.factor_importance[factor].current_weight = new_weight 182 self.factor_importance[factor].adjustment_history.append( 183 (datetime.now(), new_weight) 184 ) 185 186 # Normalize to sum to 1.0 187 total = sum(new_weights.values()) 188 if total > 0: 189 new_weights = {k: v / total for k, v in new_weights.items()} 190 191 # Update engine weights 192 self.scoring_engine.weights = new_weights 193 194 def compute_factor_correlations(self) -> Dict[str, float]: 195 """ 196 Compute correlation between each factor and selection. 197 198 Returns: 199 Dict mapping factor names to selection correlation (-1 to 1) 200 """ 201 if len(self.selection_events) < 5: 202 return {f: 0.5 for f in self.scoring_engine.weights.keys()} 203 204 correlations = {} 205 206 for factor in self.scoring_engine.weights.keys(): 207 # For each event, check if high factor value → selection 208 selection_when_high = 0 209 selection_when_low = 0 210 total_high = 0 211 total_low = 0 212 213 for event in self.selection_events[-self.WINDOW_SIZE:]: 214 # Get median factor value for this presentation 215 factor_values = [getattr(s.factors, factor) for s in event.presented_scores] 216 if not factor_values: 217 continue 218 219 median = sorted(factor_values)[len(factor_values) // 2] 220 selected_value = getattr(event.selected_score.factors, factor) 221 222 if selected_value >= median: 223 selection_when_high += 1 224 total_high += 1 225 else: 226 selection_when_low += 1 227 total_low += 1 228 229 # Correlation: how much more likely to select high-scoring 230 if total_high + total_low > 0: 231 correlation = (selection_when_high - selection_when_low) / (total_high + total_low) 232 else: 233 correlation = 0.0 234 235 correlations[factor] = correlation 236 self.factor_importance[factor].selection_correlation = correlation 237 238 return correlations 239 240 def get_weight_report(self) -> Dict[str, Any]: 241 """ 242 Get a report on current weights and their evolution. 243 244 Returns: 245 Dict with weight analysis 246 """ 247 correlations = self.compute_factor_correlations() 248 249 report = { 250 'operator_id': self.operator_id, 251 'total_selections': len(self.selection_events), 252 'learning_rate': self.learning_rate, 253 'current_weights': dict(self.scoring_engine.weights), 254 'factor_correlations': correlations, 255 'top_factors': sorted( 256 self.scoring_engine.weights.items(), 257 key=lambda x: x[1], 258 reverse=True 259 )[:3], 260 'bottom_factors': sorted( 261 self.scoring_engine.weights.items(), 262 key=lambda x: x[1] 263 )[:3], 264 } 265 266 return report 267 268 def reset_weights(self): 269 """Reset weights to defaults.""" 270 self.scoring_engine.weights = ResonanceScoringEngine.DEFAULT_WEIGHTS.copy() 271 self._initialize_factor_importance() 272 273 def export_weights(self) -> Dict[str, float]: 274 """Export current weights for persistence.""" 275 return dict(self.scoring_engine.weights) 276 277 def import_weights(self, weights: Dict[str, float]): 278 """Import weights from persistence.""" 279 self.scoring_engine.weights = dict(weights) 280 self._initialize_factor_importance() 281 282 283 # Quick test 284 if __name__ == "__main__": 285 import tempfile 286 import os 287 from ..database import GodDatabase 288 289 # Create temp database 290 db_path = os.path.join(tempfile.gettempdir(), "test_tuner.db") 291 db = GodDatabase(db_path) 292 293 print("=== Resonance Weight Tuner Test ===\n") 294 295 # Create engine and tuner 296 engine = ResonanceScoringEngine(db) 297 tuner = ResonanceWeightTuner(engine, "rick") 298 299 print("Initial weights:") 300 for factor, weight in sorted(engine.weights.items(), key=lambda x: -x[1]): 301 print(f" {factor}: {weight:.3f}") 302 303 print("\nSimulating selections that favor topic_relevance...") 304 305 # Create test bullets 306 bullets = [] 307 for i in range(10): 308 bullet = db.create_bullet( 309 content=f"Test bullet {i}", 310 blanket_id="test" 311 ) 312 bullets.append(bullet) 313 314 # Simulate selection events where operator prefers topic-relevant items 315 from .scoring import ResonanceContext 316 317 context = ResonanceContext(current_topic="architecture") 318 319 for _ in range(20): 320 # Score all bullets 321 scores = [engine.compute_resonance(b.uuid, context) for b in bullets] 322 scores = [s for s in scores if s is not None] 323 324 if len(scores) < 2: 325 continue 326 327 # Simulate operator selecting the most topic-relevant 328 sorted_by_topic = sorted(scores, key=lambda s: s.factors.topic_relevance, reverse=True) 329 selected = sorted_by_topic[0] 330 331 tuner.record_selection( 332 selected_uuid=selected.uuid, 333 presented_uuids=[s.uuid for s in scores], 334 selected_score=selected, 335 presented_scores=scores 336 ) 337 338 print("\nWeights after topic-focused selections:") 339 for factor, weight in sorted(engine.weights.items(), key=lambda x: -x[1]): 340 print(f" {factor}: {weight:.3f}") 341 342 # Get report 343 report = tuner.get_weight_report() 344 print(f"\nTotal selections: {report['total_selections']}") 345 print(f"Top factors: {report['top_factors']}") 346 347 # Cleanup 348 os.remove(db_path) 349 print("\n=== Test Complete ===")