Agentswarm/benchmark/metrics.py

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