Upload aco/router_v10.py with huggingface_hub

aco/router_v10.py

@@ -0,0 +1,151 @@
+"""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,
+        )