| """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} |
| |
| |
| 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 |
| 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 |
| |
| 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, |
| ) |
|
|
