"""ACO v10 Router: Trained on REAL SWE-Router execution data. Key difference from v8: Uses XGBoost models trained on 500 real SWE-bench tasks across 8 models, not synthetic data. Routes based on problem-statement features → per-tier P(success) → optimal tier selection. Supports cascade + feedback escalation. """ import numpy as np import pickle, os, json from typing import Dict, Optional, Tuple from dataclasses import dataclass CODE_KW = ["python","javascript","code","function","bug","debug","refactor", "implement","test","compile","runtime","segfault","thread","async","class", "module","import","error","traceback"] CRITICAL_KW = ["critical","production","urgent","emergency","live","deployed", "safety","security"] SIMPLE_KW = ["typo","simple","quick","brief","minor","small","easy","trivial","just"] RESEARCH_KW = ["research","investigate","compare","analyze","survey","paper"] TOOL_KW = ["search","fetch","retrieve","query","api","database","scrape","aggregate"] LONG_KW = ["plan","project","roadmap","orchestrate","migrate","pipeline","deploy","architecture"] FEAT_KEYS = sorted([ 'req_len','num_words','has_code','n_code','has_legal','has_research', 'has_tool','has_critical','has_simple','has_long','has_math', 'has_error_msg','has_file_path','n_lines','has_version','has_add', 'has_fix','has_change','has_remove','has_test','has_doc', 'has_see_also','has_steps_to_reproduce', ]) TIER_TO_MODEL = { 1: 'deepseek-v4-flash', 2: 'gpt-5-mini', 3: 'gemini-2.5-pro', 4: 'claude-opus-4.7', 5: 'gemini-3-pro', } TIER_COST = {1:0.01, 2:0.05, 3:0.15, 4:0.30, 5:0.50} @dataclass class V10RoutingDecision: tier: int model: str confidence: float cost_estimate: float tier_probs: Dict[int, float] escalated: bool = False class V10Router: def __init__(self, model_path: str = None, success_threshold: float = 0.5): self.success_threshold = success_threshold self.tier_clfs = None self.tier_calibs = None self.opt_clf = None self.feat_keys = FEAT_KEYS if model_path and os.path.exists(model_path): self._load(model_path) def _load(self, path): bundle = pickle.load(open(path, 'rb')) self.tier_clfs = {int(k):v for k,v in bundle.get('tier_clfs',{}).items()} self.tier_calibs = {int(k):v for k,v in bundle.get('tier_calibrators',{}).items()} self.opt_clf = bundle.get('opt_clf', None) self.feat_keys = bundle.get('feat_keys', FEAT_KEYS) def _extract(self, text: str) -> np.ndarray: r = text.lower() feats = { 'req_len': len(text), 'num_words': len(text.split()), 'has_code': int(any(k in r for k in CODE_KW)), 'n_code': sum(1 for k in CODE_KW if k in r), 'has_legal': int(any(k in r for k in ["contract","legal","compliance"])), 'has_research': int(any(k in r for k in RESEARCH_KW)), 'has_tool': int(any(k in r for k in TOOL_KW)), 'has_critical': int(any(k in r for k in CRITICAL_KW)), 'has_simple': int(any(k in r for k in SIMPLE_KW)), 'has_long': int(any(k in r for k in LONG_KW)), 'has_math': int(any(k in r for k in ["calculate","compute","solve","equation"])), 'has_error_msg': int('error' in r or 'traceback' in r or 'exception' in r), 'has_file_path': int('/' in r), 'n_lines': text.count('\n') + 1, 'has_version': int('version' in r or 'update' in r), 'has_add': int('add' in r or 'new' in r or 'create' in r), 'has_fix': int('fix' in r or 'bug' in r or 'issue' in r), 'has_change': int('change' in r or 'modify' in r), 'has_remove': int('remove' in r or 'delete' in r), 'has_test': int('test' in r or 'spec' in r), 'has_doc': int('doc' in r or 'readme' in r), 'has_see_also': int('see also' in r or 'related' in r), 'has_steps_to_reproduce': int('reproduce' in r or 'steps' in r), } return np.array([float(feats.get(k,0.0)) for k in self.feat_keys], dtype=np.float32).reshape(1,-1) def route_cascade(self, text: str) -> V10RoutingDecision: """Route to cheapest tier with P(success) >= threshold.""" x = self._extract(text) tier_probs = {} if self.tier_clfs: for t in range(1, 6): if t in self.tier_clfs: p_raw = self.tier_clfs[t].predict_proba(x)[0,1] p_cal = float(self.tier_calibs[t].transform([p_raw])[0]) tier_probs[t] = p_cal else: tier_probs[t] = 0.5 else: tier_probs = {1:0.67,2:0.72,3:0.50,4:0.84,5:0.70} # Find cheapest tier above threshold selected_tier = 5 for t in range(1, 6): if tier_probs.get(t, 0) >= self.success_threshold: selected_tier = t break model = TIER_TO_MODEL.get(selected_tier, 'claude-opus-4.7') return V10RoutingDecision( tier=selected_tier, model=model, confidence=tier_probs.get(selected_tier, 0.5), cost_estimate=TIER_COST.get(selected_tier, 0.30), tier_probs=tier_probs, ) def route_direct(self, text: str) -> V10RoutingDecision: """Predict optimal tier directly.""" x = self._extract(text) if self.opt_clf: tier = int(self.opt_clf.predict(x)[0]) + 1 else: tier = 4 # fallback model = TIER_TO_MODEL.get(tier, 'claude-opus-4.7') return V10RoutingDecision( tier=tier, model=model, confidence=0.8, cost_estimate=TIER_COST.get(tier, 0.30), tier_probs={}, ) def route_with_feedback(self, text: str, initial_success: bool = True) -> V10RoutingDecision: """Route with feedback: start cheap, escalate on failure.""" initial = self.route_cascade(text) if initial_success: return initial # Escalate escalated_tier = min(initial.tier + 1, 5) model = TIER_TO_MODEL.get(escalated_tier, 'claude-opus-4.7') return V10RoutingDecision( tier=escalated_tier, model=model, confidence=initial.tier_probs.get(escalated_tier, 0.8), cost_estimate=TIER_COST.get(initial.tier, 0.01) + TIER_COST.get(escalated_tier, 0.30), tier_probs=initial.tier_probs, escalated=True, )