Upload aco/pareto.py

aco/pareto.py

@@ -0,0 +1,97 @@
+"""Cost-Quality Pareto Frontier analysis.
+
+Implements RouterBench NDCH (arxiv:2403.12031) and
+RouteLLM CPT/APGR (arxiv:2406.18665) metrics.
+"""
+import numpy as np
+from typing import Dict, List, Optional, Tuple
+
+
+def compute_ndch(points: np.ndarray) -> np.ndarray:
+    """Non-Decreasing Convex Hull from RouterBench Sec 3.2.
+    
+    Args:
+        points: Nx2 array of (cost, quality) pairs
+    Returns:
+        NDCH vertices sorted by cost
+    """
+    from scipy.spatial import ConvexHull
+    if len(points) < 3:
+        idx = np.argsort(points[:, 0])
+        return points[idx]
+    hull = ConvexHull(points)
+    hull_pts = points[hull.vertices]
+    hull_pts = hull_pts[hull_pts[:, 0].argsort()]
+    ndch = [hull_pts[0]]
+    for pt in hull_pts[1:]:
+        if pt[1] >= ndch[-1][1]:
+            ndch.append(pt)
+    return np.array(ndch)
+
+
+def cost_at_quality(ndch: np.ndarray, target_q: float) -> Optional[float]:
+    """Find minimum cost on NDCH achieving target_quality."""
+    for i in range(len(ndch) - 1):
+        q1, q2 = ndch[i, 1], ndch[i+1, 1]
+        if q1 <= target_q <= q2:
+            c1, c2 = ndch[i, 0], ndch[i+1, 0]
+            return c1 + (c2 - c1) * (target_q - q1) / (q2 - q1)
+    return None
+
+
+def cost_savings_at_iso(ndch_opt, ndch_base, target_q):
+    """Cost reduction at iso-quality: 1 - c_opt/c_base."""
+    c_opt = cost_at_quality(ndch_opt, target_q)
+    c_base = cost_at_quality(ndch_base, target_q)
+    if c_opt is None or c_base is None:
+        return None
+    return 1.0 - c_opt / c_base
+
+
+def compute_aiq(ndch, c_min, c_max):
+    """Average Improvement in Quality (RouterBench)."""
+    costs = np.linspace(c_min, c_max, 100)
+    quals = np.interp(costs, ndch[:, 0], ndch[:, 1])
+    return float(np.trapz(quals, costs) / (c_max - c_min))
+
+
+def compute_apgr(ndch_router, ndch_random, c_max):
+    """Average Performance Gap Recovery (RouteLLM).
+    APGR = integral of PGR over cost fraction [0,1].
+    PGR(c) = (q_router(c) - q_weak(c)) / (q_strong(c) - q_weak(c))
+    """
+    costs = np.linspace(0, c_max, 100)
+    q_r = np.interp(costs, ndch_router[:, 0], ndch_router[:, 1])
+    q_rand = np.interp(costs, ndch_random[:, 0], ndch_random[:, 1])
+    q_weak = ndch_random[0, 1]
+    q_strong = ndch_random[-1, 1]
+    pgr = (q_r - q_weak) / max(q_strong - q_weak, 1e-6)
+    return float(np.trapz(pgr, costs) / c_max)
+
+
+def build_frontier_report(policies: Dict[str, Tuple[float, float]]):
+    """Build full Pareto frontier report.
+    
+    Args:
+        policies: {name: (success_rate, cost_per_task)}
+    """
+    points = np.array([(c, q) for q, c in policies.values()])
+    names = list(policies.keys())
+    
+    ndch = compute_ndch(points)
+    c_min, c_max = points[:, 0].min(), points[:, 0].max()
+    aiq = compute_aiq(ndch, c_min, c_max)
+    
+    report = {
+        "policies": {n: {"success": q, "cost": c} for n, (q, c) in policies.items()},
+        "ndch_vertices": ndch.tolist(),
+        "aiq": round(aiq, 4),
+        "frontier_quality_levels": {},
+    }
+    
+    for q_target in [0.65, 0.70, 0.75, 0.78, 0.80, 0.85, 0.87]:
+        c = cost_at_quality(ndch, q_target)
+        if c is not None:
+            report["frontier_quality_levels"][f"q={q_target:.2f}"] = round(c, 4)
+    
+    return report