AgentEvalTool/backend/agenteval/evaluation/go_no_go.py

"""Go/No-Go acceptance verdict for evaluation runs."""

from datetime import datetime, timezone
from typing import Any, Optional

from pydantic import BaseModel, Field


class AcceptanceCriteria(BaseModel):
    """Acceptance criteria for go/no-go verdict."""

    judged_pass_rate_min: float = Field(default=0.95, description="Minimum judged pass rate (0-1)")
    pass_rate_min: float = Field(default=0.90, description="Minimum overall pass rate (0-1)")
    avg_latency_max_ms: Optional[float] = Field(default=None, description="Maximum average latency in ms")
    availability_min: Optional[float] = Field(default=None, description="Minimum availability (0-1), for campaign level")


class CriterionResult(BaseModel):
    """Result of checking a single criterion."""

    criterion: str = Field(description="Criterion name")
    threshold: float = Field(description="Threshold value")
    actual: float = Field(description="Actual value")
    passed: bool = Field(description="Whether the criterion passed")
    detail: str = Field(default="", description="Human-readable detail")


class GoNoGoVerdict(BaseModel):
    """Go/No-Go verdict for an evaluation run."""

    decision: str = Field(description="go | no_go | conditional")
    summary: str = Field(description="Human-readable summary")
    criteria_results: list[CriterionResult] = Field(default_factory=list)
    generated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


def evaluate_go_no_go(
    summary: dict[str, Any],
    criteria: AcceptanceCriteria | None = None,
) -> GoNoGoVerdict:
    """Evaluate go/no-go verdict based on run summary and acceptance criteria.

    Args:
        summary: Run summary dict with keys like judged_pass_rate, pass_rate, avg_latency_ms
        criteria: Acceptance criteria. If None, uses defaults.

    Returns:
        GoNoGoVerdict with decision, summary, and criteria results.
    """
    if criteria is None:
        criteria = AcceptanceCriteria()

    results: list[CriterionResult] = []

    # Check judged pass rate (only if present in summary)
    judged_pass_rate = summary.get("judged_pass_rate")
    if judged_pass_rate is None:
        judged_pass_rate = summary.get("pass_rate")
    if judged_pass_rate is not None:
        passed = judged_pass_rate >= criteria.judged_pass_rate_min
        results.append(CriterionResult(
            criterion="judged_pass_rate",
            threshold=criteria.judged_pass_rate_min,
            actual=judged_pass_rate,
            passed=passed,
            detail=f"判定型通过率 {judged_pass_rate*100:.1f}% {'>=' if passed else '<'} {criteria.judged_pass_rate_min*100:.0f}%",
        ))

    # Check overall pass rate (only if different from judged and present)
    pass_rate = summary.get("pass_rate")
    if pass_rate is not None and pass_rate != judged_pass_rate:
        passed = pass_rate >= criteria.pass_rate_min
        results.append(CriterionResult(
            criterion="pass_rate",
            threshold=criteria.pass_rate_min,
            actual=pass_rate,
            passed=passed,
            detail=f"全量通过率 {pass_rate*100:.1f}% {'>=' if passed else '<'} {criteria.pass_rate_min*100:.0f}%",
        ))

    # Check average latency
    avg_latency_ms = summary.get("avg_latency_ms")
    if avg_latency_ms is not None and criteria.avg_latency_max_ms is not None:
        passed = avg_latency_ms <= criteria.avg_latency_max_ms
        results.append(CriterionResult(
            criterion="avg_latency_ms",
            threshold=criteria.avg_latency_max_ms,
            actual=avg_latency_ms,
            passed=passed,
            detail=f"平均延迟 {avg_latency_ms:.0f}ms {'<=' if passed else '>'} {criteria.avg_latency_max_ms:.0f}ms",
        ))

    # Check availability (for campaign level)
    availability = summary.get("overall_availability") or summary.get("availability")
    if availability is not None and criteria.availability_min is not None:
        passed = availability >= criteria.availability_min
        results.append(CriterionResult(
            criterion="availability",
            threshold=criteria.availability_min,
            actual=availability,
            passed=passed,
            detail=f"可用性 {availability*100:.1f}% {'>=' if passed else '<'} {criteria.availability_min*100:.0f}%",
        ))

    # Determine overall decision
    if not results:
        return GoNoGoVerdict(
            decision="conditional",
            summary="无可用指标进行评估",
            criteria_results=[],
        )

    all_passed = all(r.passed for r in results)
    if all_passed:
        decision = "go"
        main_metric = results[0]
        summary_text = f"通过率 {main_metric.actual*100:.0f}%，达标，建议上线"
    else:
        # Check if core metrics (pass rate) failed
        core_failed = any(
            not r.passed and r.criterion in ("judged_pass_rate", "pass_rate", "availability")
            for r in results
        )
        if core_failed:
            decision = "no_go"
            failed_metrics = [r for r in results if not r.passed]
            summary_text = f"核心指标未达标（{', '.join(r.criterion for r in failed_metrics)}），不建议上线"
        else:
            decision = "conditional"
            risky_metrics = [r for r in results if not r.passed]
            summary_text = f"部分指标达标，存在风险项（{', '.join(r.criterion for r in risky_metrics)}），建议修复后复测"

    return GoNoGoVerdict(
        decision=decision,
        summary=summary_text,
        criteria_results=results,
    )