Spaces:
Running
Running
File size: 12,294 Bytes
09801ca | 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 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 | """
📈 Training Validator Agent
Validates training behavior BEFORE final evaluation:
- Learning curve analysis
- Bias-variance diagnosis
- Training stability
- Early overfitting detection
KEY INNOVATION: Catch problems early, don't waste compute on bad models.
"""
import numpy as np
from typing import Dict, List, Any, Tuple, Optional
from dataclasses import dataclass
import logging
from .base import BaseAgent, AgentResult, AgentStatus, Phase, MessageType
logger = logging.getLogger(__name__)
@dataclass
class TrainingDiagnosis:
"""Diagnosis of training behavior"""
issue: str # high_bias, high_variance, unstable, overfitting
severity: str # low, medium, high
recommendation: str
target_agent: str # Which agent should fix this
class TrainingValidatorAgent(BaseAgent):
"""
Training Validator Agent
Validates training behavior → Diagnoses issues → Routes to fix agents
"""
name = "training_validator"
description = "Validates training behavior and catches issues early"
def __init__(self, memory=None):
super().__init__(memory)
self.diagnoses: List[TrainingDiagnosis] = []
def execute(self, **kwargs) -> AgentResult:
"""Main execution: validate training, diagnose issues"""
# Get trained models and data
X = self.read_state("features_engineered")
if X is None:
X = self.read_state("features")
y = self.read_state("target")
task_type = self.read_state("task_type")
best_score = self.read_state("best_score")
best_model_name = self.read_state("best_model_name")
if X is None or y is None:
return AgentResult(
status=AgentStatus.FAILED,
agent_name=self.name,
phase=self.current_phase,
errors=["Missing data"]
)
# Get best model artifact
model_artifact = self.memory.get_latest_artifact("model")
if model_artifact is None:
return AgentResult(
status=AgentStatus.FAILED,
agent_name=self.name,
phase=self.current_phase,
errors=["No trained model found"]
)
model = model_artifact.data
self.logger.info(f"📊 Validating: {best_model_name} (score={best_score:.4f})")
# Run validation based on phase
if self.is_fast_phase():
diagnoses = self._fast_validation(model, X, y, task_type, best_score)
else:
diagnoses = self._deep_validation(model, X, y, task_type, best_score)
self.diagnoses = diagnoses
# Check if issues found
if diagnoses:
critical_issues = [d for d in diagnoses if d.severity == "high"]
if critical_issues:
# Create retry messages for feedback loop
result = AgentResult(
status=AgentStatus.RETRY,
agent_name=self.name,
phase=self.current_phase,
data={"diagnoses": [d.__dict__ for d in diagnoses]},
recommendations=[d.recommendation for d in critical_issues]
)
# Add messages to route to fix agents
for issue in critical_issues:
result.add_message(
receiver=issue.target_agent,
msg_type=MessageType.RETRY,
payload={"issue": issue.issue, "recommendation": issue.recommendation}
)
self.logger.warning(f" ⚠️ {len(critical_issues)} critical issues found")
return result
# Validation passed
self.write_state("training_validated", True, self.name)
self.write_state("training_diagnoses", [d.__dict__ for d in diagnoses], self.name)
return AgentResult(
status=AgentStatus.SUCCESS,
agent_name=self.name,
phase=self.current_phase,
data={
"validated": True,
"minor_issues": len(diagnoses)
},
metrics={
"score": best_score
}
)
# =========================================================================
# FAST VALIDATION
# =========================================================================
def _fast_validation(self, model, X: np.ndarray, y: np.ndarray,
task_type: str, score: float) -> List[TrainingDiagnosis]:
"""Quick validation checks with production-realistic thresholds"""
diagnoses = []
# Get retry count from memory to be more lenient after attempts
retry_attempts = self.read_state("validation_retry_count") or 0
# After 2 retries, be more lenient (data limitation, not model issue)
is_lenient_mode = retry_attempts >= 2
# Adjusted thresholds based on retry attempts
if is_lenient_mode:
score_threshold = 0.3 if task_type == "classification" else -0.5 # Very lenient
gap_threshold = 0.6 # Accept higher gaps
self.logger.info(f" 📋 Lenient mode (attempt {retry_attempts + 1})")
else:
score_threshold = 0.5 if task_type == "classification" else 0.05
gap_threshold = 0.35 # More realistic for production data
# Check: Score too low
if score < score_threshold:
diagnoses.append(TrainingDiagnosis(
issue="low_performance",
severity="medium" if is_lenient_mode else "high",
recommendation="Try different features or algorithms",
target_agent="feature_engineer"
))
self.logger.warning(f" ⚠️ Low performance: {score:.4f} < {score_threshold}")
# Check: Train-test gap (quick check via predictions)
train_score = self._get_train_score(model, X, y, task_type)
gap = train_score - score
if gap > gap_threshold:
diagnoses.append(TrainingDiagnosis(
issue="overfitting",
severity="medium" if is_lenient_mode else ("high" if gap > 0.5 else "medium"),
recommendation="Reduce model complexity or add regularization",
target_agent="hyperparam"
))
self.logger.warning(f" ⚠️ Train-test gap: {gap:.4f}")
# Update retry count
self.write_state("validation_retry_count", retry_attempts + 1, self.name)
if not diagnoses:
self.logger.info(f" ✅ Fast validation passed")
elif is_lenient_mode and all(d.severity != "high" for d in diagnoses):
# In lenient mode with only medium issues, pass anyway
self.logger.info(f" ✅ Validation passed (lenient mode, {len(diagnoses)} minor issues)")
return [] # Clear issues to pass
return diagnoses
# =========================================================================
# DEEP VALIDATION
# =========================================================================
def _deep_validation(self, model, X: np.ndarray, y: np.ndarray,
task_type: str, score: float) -> List[TrainingDiagnosis]:
"""Deep validation with learning curves and stability checks"""
diagnoses = self._fast_validation(model, X, y, task_type, score)
# Learning curve analysis
lc_diagnosis = self._analyze_learning_curve(model, X, y, task_type)
if lc_diagnosis:
diagnoses.append(lc_diagnosis)
# Stability check (multiple seeds)
stability_diagnosis = self._check_stability(model, X, y, task_type)
if stability_diagnosis:
diagnoses.append(stability_diagnosis)
if not diagnoses:
self.logger.info(f" ✅ Deep validation passed")
return diagnoses
def _get_train_score(self, model, X: np.ndarray, y: np.ndarray,
task_type: str) -> float:
"""Get training score"""
try:
y_pred = model.predict(X)
if task_type == "classification":
from sklearn.metrics import accuracy_score
return accuracy_score(y, y_pred)
else:
from sklearn.metrics import r2_score
return r2_score(y, y_pred)
except:
return 1.0 # Assume perfect train score if can't calculate
def _analyze_learning_curve(self, model, X: np.ndarray, y: np.ndarray,
task_type: str) -> Optional[TrainingDiagnosis]:
"""Analyze learning curve for bias/variance"""
try:
from sklearn.model_selection import learning_curve
# Sample for speed
n_samples = min(5000, X.shape[0])
indices = np.random.choice(X.shape[0], n_samples, replace=False)
X_sample, y_sample = X[indices], y[indices]
train_sizes, train_scores, test_scores = learning_curve(
model.__class__(**model.get_params()),
X_sample, y_sample,
train_sizes=np.linspace(0.2, 1.0, 5),
cv=3,
n_jobs=-1,
scoring='accuracy' if task_type == "classification" else 'r2'
)
# Analyze curve
train_mean = train_scores.mean(axis=1)
test_mean = test_scores.mean(axis=1)
# High bias: both train and test scores are low
if train_mean[-1] < 0.6 and test_mean[-1] < 0.5:
return TrainingDiagnosis(
issue="high_bias",
severity="high",
recommendation="Model is too simple. Try more complex model or more features",
target_agent="model_strategy"
)
# High variance: train high, test low
final_gap = train_mean[-1] - test_mean[-1]
if final_gap > 0.2:
return TrainingDiagnosis(
issue="high_variance",
severity="high" if final_gap > 0.3 else "medium",
recommendation="Model is overfitting. Need more data or simpler model",
target_agent="hyperparam"
)
self.logger.info(f" 📈 Learning curve: train={train_mean[-1]:.3f}, test={test_mean[-1]:.3f}")
except Exception as e:
self.logger.warning(f" ⚠️ Learning curve failed: {str(e)[:30]}")
return None
def _check_stability(self, model, X: np.ndarray, y: np.ndarray,
task_type: str) -> Optional[TrainingDiagnosis]:
"""Check model stability across random seeds"""
try:
from sklearn.model_selection import cross_val_score, KFold
scores = []
for seed in [42, 123, 456]:
cv = KFold(n_splits=3, shuffle=True, random_state=seed)
cv_scores = cross_val_score(
model.__class__(**model.get_params()),
X, y, cv=cv,
scoring='accuracy' if task_type == "classification" else 'r2'
)
scores.append(cv_scores.mean())
std = np.std(scores)
if std > 0.05:
return TrainingDiagnosis(
issue="unstable",
severity="medium",
recommendation="Model is unstable across folds. Consider ensemble or more data",
target_agent="hyperparam"
)
self.logger.info(f" 📊 Stability: std={std:.4f}")
except Exception as e:
self.logger.warning(f" ⚠️ Stability check failed: {str(e)[:30]}")
return None
|