File size: 6,088 Bytes
8f09a93 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """
optimizer.py — Agentic optimization decision policy.
Monitors agent performance and decides what to optimize:
improving → continue (don't fix what works)
plateau → optimize prompts/skills/router first (epigenetic)
epigenetic_plateau → suggest LoRA/distillation only if ROI positive
degrading → rollback immediately
Never distills without eval data AND ROI check.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class OptimizationState(str, Enum):
IMPROVING = "improving"
PLATEAU = "plateau"
EPIGENETIC_PLATEAU = "epigenetic_plateau"
DEGRADING = "degrading"
@dataclass
class PerformanceWindow:
"""Rolling window of performance scores."""
scores: list[float] = field(default_factory=list)
window_size: int = 10
def add(self, score: float) -> None:
self.scores.append(score)
if len(self.scores) > self.window_size * 3:
self.scores = self.scores[-self.window_size * 2:]
@property
def recent(self) -> list[float]:
return self.scores[-self.window_size:]
@property
def previous(self) -> list[float]:
if len(self.scores) < self.window_size * 2:
return self.scores[:len(self.scores)//2]
return self.scores[-self.window_size*2:-self.window_size]
@property
def trend(self) -> float:
"""Positive = improving, negative = degrading, ~0 = plateau."""
r = self.recent
p = self.previous
if not r or not p:
return 0.0
return (sum(r)/len(r)) - (sum(p)/len(p))
@dataclass
class OptimizationRecommendation:
"""What the optimizer recommends."""
state: OptimizationState
action: str # "continue", "optimize_prompts", "optimize_skills", "suggest_lora", "rollback"
reason: str
details: dict[str, Any] = field(default_factory=dict)
class AgenticOptimizer:
"""
Monitors performance and recommends optimization actions.
Usage:
optimizer = AgenticOptimizer()
# After each task:
optimizer.record_score(phi_score)
# Periodically check:
rec = optimizer.recommend()
if rec.action == "optimize_prompts":
# Rebuild prompt pack with new skills
elif rec.action == "rollback":
# Revert to previous configuration
"""
def __init__(
self,
plateau_threshold: float = 0.05,
degradation_threshold: float = -0.1,
min_samples: int = 5,
epigenetic_attempts_before_lora: int = 3,
):
self.plateau_threshold = plateau_threshold
self.degradation_threshold = degradation_threshold
self.min_samples = min_samples
self.epigenetic_attempts = epigenetic_attempts_before_lora
self.window = PerformanceWindow()
self._epigenetic_count = 0
self._recommendations: list[OptimizationRecommendation] = []
def record_score(self, score: float) -> None:
"""Record a performance score (typically final Φ)."""
self.window.add(score)
def recommend(self) -> OptimizationRecommendation:
"""Get optimization recommendation based on current performance trend."""
if len(self.window.scores) < self.min_samples:
rec = OptimizationRecommendation(
state=OptimizationState.IMPROVING,
action="continue",
reason=f"Insufficient data ({len(self.window.scores)}/{self.min_samples} samples)",
)
self._recommendations.append(rec)
return rec
trend = self.window.trend
state = self._classify_state(trend)
if state == OptimizationState.IMPROVING:
rec = OptimizationRecommendation(
state=state, action="continue",
reason=f"Performance improving (trend={trend:+.3f})",
)
elif state == OptimizationState.DEGRADING:
rec = OptimizationRecommendation(
state=state, action="rollback",
reason=f"Performance degrading (trend={trend:+.3f}). Rollback to previous config.",
)
elif state == OptimizationState.PLATEAU:
self._epigenetic_count += 1
if self._epigenetic_count >= self.epigenetic_attempts:
rec = OptimizationRecommendation(
state=OptimizationState.EPIGENETIC_PLATEAU,
action="suggest_lora",
reason=f"Plateau persists after {self._epigenetic_count} epigenetic attempts. Consider LoRA/distillation.",
details={"epigenetic_attempts": self._epigenetic_count},
)
else:
rec = OptimizationRecommendation(
state=state,
action="optimize_prompts" if self._epigenetic_count <= 1 else "optimize_skills",
reason=f"Performance plateau (trend={trend:+.3f}). Trying epigenetic optimization #{self._epigenetic_count}.",
)
else:
rec = OptimizationRecommendation(state=state, action="continue", reason="Unknown state")
self._recommendations.append(rec)
return rec
def _classify_state(self, trend: float) -> OptimizationState:
if trend > self.plateau_threshold:
return OptimizationState.IMPROVING
elif trend < self.degradation_threshold:
return OptimizationState.DEGRADING
else:
return OptimizationState.PLATEAU
def reset_epigenetic_counter(self) -> None:
"""Call after successful epigenetic optimization breaks plateau."""
self._epigenetic_count = 0
@property
def current_state(self) -> OptimizationState:
if len(self.window.scores) < self.min_samples:
return OptimizationState.IMPROVING
return self._classify_state(self.window.trend)
@property
def history(self) -> list[OptimizationRecommendation]:
return self._recommendations
|