ab_model_routing（ab 模型路由）

ab_model_routing

Implementa un sistema de routing A/B para evaluar nuevas versiones de modelos de machine learning en producción dentro del pipeline de verificación de identidad. Permite dirigir un porcentaje configurable del tráfico de verificación a un modelo candidato (ej: ArcFace v2) mientras el modelo de control (ej: ArcFace v1) sirve el resto, recopilando métricas comparativas de FAR, FRR y latencia para tomar decisiones de promoción basadas en datos.

When to use

Usa esta skill cuando necesites evaluar una nueva versión de modelo ML en producción dentro del model_server_agent. Aplica cuando se tenga un modelo candidato que haya pasado validación offline pero se requiera confirmar su rendimiento con tráfico real de verificación antes de hacer un rollout completo.

Instructions

1. Definir la configuración del experimento A/B: @dataclass class ABExperiment: experiment_id: str model_name: str # ej: "arcface" control_version: str # ej: "1.0.0" candidate_version: str # ej: "2.0.0" traffic_split: float # 0.0 a 1.0, porcentaje al candidato start_date: datetime min_samples: int # mínimo de verificaciones para significancia status: str # "running", "completed", "aborted" metrics_config: dict # métricas a comparar: ["far", "frr", "latency_p95"]
2. Implementar el router que asigna cada solicitud de verificación al modelo control o candidato: class ABRouter: def __init__(self, experiment: ABExperiment): self.experiment = experiment self.control_model = load_model(experiment.control_version) self.candidate_model = load_model(experiment.candidate_version) def route(self, session_id: str) -> tuple[str, Model]: # Hash determinista por session_id para consistencia hash_value = int(hashlib.sha256(session_id.encode()).hexdigest(), 16) if (hash_value % 100) / 100 < self.experiment.traffic_split: return "candidate", self.candidate_model return "control", self.control_model
3. Registrar los resultados de cada verificación asociados al grupo (control/candidato): class ABMetricsCollector: def record(self, experiment_id: str, group: str, session_id: str, prediction: dict, ground_truth: dict = None): self.db.insert({"experiment_id": experiment_id, "group": group, "session_id": session_id, "similarity_score": prediction["score"], "is_match": prediction["is_match"], "latency_ms": prediction["latency_ms"], "timestamp": datetime.utcnow()})
4. Implementar el análisis estadístico para determinar si la diferencia entre modelos es significativa: from scipy import stats def analyze_experiment(self, experiment_id: str) -> dict: control_scores = self.get_scores(experiment_id, "control") candidate_scores = self.get_scores(experiment_id, "candidate") t_stat, p_value = stats.ttest_ind(control_scores, candidate_scores) control_far = self.calculate_far(experiment_id, "control") candidate_far = self.calculate_far(experiment_id, "candidate") return {"p_value": p_value, "control_far": control_far, "candidate_far": candidate_far, "is_significant": p_value < 0.05, "recommendation": "promote" if candidate_far < control_far and p_value < 0.05 else "keep_control"}
5. Configurar guardrails automáticos que detengan el experimento si el modelo candidato degrada métricas críticas: def check_guardrails(self, experiment_id: str): candidate_far = self.calculate_far(experiment_id, "candidate") if candidate_far > 0.001: # FAR > 0.1% -> abortar self.abort_experiment(experiment_id) self.alert("AB experiment aborted: candidate FAR exceeded threshold")
6. Exponer un dashboard o endpoint con métricas en tiempo real del experimento: FAR, FRR, latencia p50/p95, volumen de muestras por grupo y significancia estadística actual.
7. Al completar el experimento, generar un informe automático y, si el candidato es superior, integrarse con la skill model_versioning para promover la nueva versión.

Notes

• El routing debe ser determinista por session_id para que todas las llamadas de una misma verificación (embedding, comparación) usen el mismo modelo, evitando inconsistencias en los resultados.
• Comenzar con un traffic split conservador (5-10% al candidato) e incrementar gradualmente solo si los guardrails no se activan; nunca iniciar un experimento con mas del 20% en el candidato sin validación previa.
• Los experimentos A/B en modelos de seguridad KYC requieren un volumen mínimo significativo (recomendado >1000 verificaciones por grupo) antes de tomar decisiones de promoción; resultados prematuros pueden ser engañosos.