"""Swarm benchmark metric formulas + SwarmMetrics schema. Aligned to **Agent 蜂群指标量化与标准 v2.0**. Pure formulas only (given inputs → value, unit-testable); data collection (turning runtime signals into inputs) is partial — see collectors/run_collector.py and docs/benchmark/metric-coverage-gaps.md. Recommended weights/hyperparameters from v2.0 are provided as defaults below. """ from __future__ import annotations import math from dataclasses import dataclass from typing import Optional # --- v2.0 recommended weights / hyperparameters --- TAU_WEIGHTS = {"success": 0.25, "quality": 0.20, "acceptance": 0.20, "cost": 0.10, "time": 0.10, "risk": 0.08, "rollback": 0.07} ETA_WEIGHTS = {"match": 0.25, "urgency": 0.15, "dependency": 0.15, "resource": 0.15, "confidence": 0.10, "risk": 0.10, "budget_pressure": 0.10} REWARD_WEIGHTS = {"s_task": 0.20, "q_quality": 0.20, "v_speed": 0.12, "e_cost": 0.13, "r_robust": 0.13, "g_gov": 0.10, "p_risk": 0.07, "p_rework": 0.05} # Sub-weights inside Q_quality (v2.0 §4.1). Masked & renormalized per task — see quality_score(). QUALITY_WEIGHTS = {"test_pass_rate": 0.4, "code_review_score": 0.3, "user_acceptance": 0.3} SWARM_WEIGHTS = {"completion": 0.25, "gain": 0.20, "collaboration": 0.15, "communication": 0.10, "cost": 0.10, "robustness": 0.10, "governance": 0.10} LAMBDA_WEIGHTS = {"lambda1": 0.30, "lambda2": 0.20, "lambda3": 0.25, "lambda4": 0.15, "lambda5": 0.10} # Σ = 1.0 THETA_DEFAULTS = {"alpha": 1.0, "beta": 2.0, "rho": 0.10, "n_agent": 5, "epsilon": 0.10} # Q_base reference coefficients (standard §6.1), by baseline type. # ⚠️ PROVISIONAL / METADATA ONLY: these are assumed constants — the standard gives NO derivation. # They are reported as `base_coefficient` but used in NO formula (G_E/G_E,c use raw Q_base). # TODO(benchmark): quantify empirically later, e.g. coefficient = Q_baseline / Q_reference on the # shared task set (reference = strongest baseline or an oracle/acceptance ceiling). Until then, # do not wire into scoring. See docs/benchmark/baseline-comparison.md §1.1. BASE_COEFFICIENTS = {"single": 0.75, "chain": 0.85, "sub": 0.90, "strong": 0.95} @dataclass class SwarmMetrics: """v2.0 §8.1 collection interface.""" tau: float # 信息素得分 eta: float # 启发式得分 p_decision: float # 决策概率 reward: float # 执行奖励 s_completion: float s_gain: float s_collaboration: float s_communication: float s_cost: float s_robustness: float s_governance: float s_swarm: float # 蜂群总分 g_e: float # 涌现增益 g_e_cost: float # 成本归一化增益 benchmark: float # 综合评分 def _ratio_pct(numerator: float, denominator: float, *, empty: float = 0.0) -> float: if not denominator: return empty return 100.0 * numerator / denominator # --- 决策层（v2.0 §3） --- def pheromone(*, success: float, quality: float, acceptance: float, cost: float, time: float, risk: float, rollback: float, weights: dict = TAU_WEIGHTS) -> float: w = weights return (w["success"] * success + w["quality"] * quality + w["acceptance"] * acceptance - w["cost"] * cost - w["time"] * time - w["risk"] * risk - w["rollback"] * rollback) def heuristic(*, match: float, urgency: float, dependency: float, resource: float, confidence: float, risk: float, budget_pressure: float, weights: dict = ETA_WEIGHTS) -> float: g = weights return (g["match"] * match + g["urgency"] * urgency + g["dependency"] * dependency + g["resource"] * resource + g["confidence"] * confidence - g["risk"] * risk - g["budget_pressure"] * budget_pressure) def action_probability(tau: float, eta: float, alpha: float, beta: float, candidates: list[tuple[float, float]]) -> float: # P(s,a,r) = τ^α·η^β / Σ τ_i^α·η_i^β denom = sum((t ** alpha) * (e ** beta) for t, e in candidates) if not denom: return 0.0 return ((tau ** alpha) * (eta ** beta)) / denom def p_decision(tau: float, eta: float, alpha: float = THETA_DEFAULTS["alpha"], beta: float = THETA_DEFAULTS["beta"]) -> float: # v2.0 §3.3: P_decision = τ^α · η^β · 100 return (tau ** alpha) * (eta ** beta) * 100.0 # --- 执行层（v2.0 §4） --- def quality_score( test_pass_rate: Optional[float] = None, code_review_score: Optional[float] = None, user_acceptance: Optional[float] = None, *, weights: dict = QUALITY_WEIGHTS, ) -> float: """Q_quality as a MASKED, RENORMALIZED weighted mean over the inputs that apply (v2.1 ruling). An input that does not apply to the task is passed as ``None`` and drops out of BOTH the weighted sum and the weight normalizer, so the score stays on [0,100] and remains comparable across task types. Canonical case: a non-code prompt has no tests → ``test_pass_rate=None`` → Q_quality is renormalized over {code_review_score, user_acceptance}. The same masking applies to any absent input, not just test_pass_rate. Q_quality = Σ_{i∈present} wᵢ·xᵢ / Σ_{i∈present} wᵢ All inputs absent → NaN (org rule #9: never fabricate a 0/100 for an uncollected quantity). With all three present and the default weights this reduces to the v2.0 0.4/0.3/0.3 blend. """ inputs = { "test_pass_rate": test_pass_rate, "code_review_score": code_review_score, "user_acceptance": user_acceptance, } num = sum(weights[k] * x for k, x in inputs.items() if x is not None) denom = sum(weights[k] for k, x in inputs.items() if x is not None) if not denom: return math.nan return num / denom def speed_score(target_time: float, actual_time: float) -> float: return _ratio_pct(target_time, actual_time) def cost_efficiency_score(expected_cost: float, actual_cost: float) -> float: # Owner ruling (v2.1): E_cost (§4.1) == S_cost (§5.1) == CostEfficiency (§6.2) — one quantity, # 100×Budget/ActualCost (Budget≡ExpectedCost, ActualUsage≡ActualCost). Same math as cost_score(). return _ratio_pct(expected_cost, actual_cost) def rework_penalty(rework_count: int, total_tasks: int) -> float: return _ratio_pct(rework_count, total_tasks) def reward(*, s_task: float, q_quality: float, v_speed: float, e_cost: float, r_robust: float, g_gov: float, p_risk: float, p_rework: float, weights: dict = REWARD_WEIGHTS) -> float: w = weights return (w["s_task"] * s_task + w["q_quality"] * q_quality + w["v_speed"] * v_speed + w["e_cost"] * e_cost + w["r_robust"] * r_robust + w["g_gov"] * g_gov - w["p_risk"] * p_risk - w["p_rework"] * p_rework) # --- 蜂群层（v2.0 §5） --- def completion_score(completed_tasks: int, total_tasks: int) -> float: return _ratio_pct(completed_tasks, total_tasks) def collaboration_score(handoff_success_rate: float, dependency_resolution_rate: float, workload_balance_score: float) -> float: return (0.5 * handoff_success_rate + 0.3 * dependency_resolution_rate + 0.2 * workload_balance_score) def communication_score(successful_messages: int, total_messages: int) -> float: return _ratio_pct(successful_messages, total_messages) def cost_score(budget: float, actual_usage: float) -> float: return _ratio_pct(budget, actual_usage) def robustness_score(recovered_failures: int, total_failures: int) -> float: return _ratio_pct(recovered_failures, total_failures, empty=100.0) def governance_score(compliant_operations: int, total_operations: int) -> float: return _ratio_pct(compliant_operations, total_operations, empty=100.0) def swarm_score(*, completion: float, gain: float, collaboration: float, communication: float, cost: float, robustness: float, governance: float, weights: dict = SWARM_WEIGHTS) -> float: w = weights return (w["completion"] * completion + w["gain"] * gain + w["collaboration"] * collaboration + w["communication"] * communication + w["cost"] * cost + w["robustness"] * robustness + w["governance"] * governance) # --- 涌现增益（v2.0 §6） --- def emergence_gain(q_swarm: float, q_base: float) -> float: # G_E = Q_swarm − Q_base return q_swarm - q_base def swarm_cost(n_agent: int, cost_efficiency: float) -> float: # v2.0 §6.2: C_swarm = N_agent × (CostEfficiency/100 + 0.5) return n_agent * (cost_efficiency / 100.0 + 0.5) def cost_normalized_gain(q_swarm: float, c_swarm: float, q_base: float, c_base: float = 1.0) -> float: # v2.0 §6.2 (CHANGED from v1 ratio-of-ratios to a DIFFERENCE): # G_E,c = (Q_swarm / C_swarm) − (Q_base / C_base) if not c_swarm or not c_base: raise ValueError("cost_normalized_gain needs non-zero C_swarm / C_base") return (q_swarm / c_swarm) - (q_base / c_base) # --- 综合 Benchmark（v2.0 §7） --- def benchmark_agent(*, s_swarm: float, g_e: float, reward: float, observability: float, governance: float, weights: dict = LAMBDA_WEIGHTS, validate: bool = True) -> float: """Benchmark_Agent = λ1·S_swarm + λ2·G_E + λ3·R + λ4·O + λ5·Gov, with Σλ = 1.0 (v2.0 §7.1).""" if validate and abs(sum(weights.values()) - 1.0) > 1e-6: raise ValueError(f"λ weights must sum to 1.0 (got {sum(weights.values())})") return (weights["lambda1"] * s_swarm + weights["lambda2"] * g_e + weights["lambda3"] * reward + weights["lambda4"] * observability + weights["lambda5"] * governance)