diff --git a/backend/agenteval/evaluation/go_no_go.py b/backend/agenteval/evaluation/go_no_go.py new file mode 100644 index 0000000..b0afb2f --- /dev/null +++ b/backend/agenteval/evaluation/go_no_go.py @@ -0,0 +1,137 @@ +"""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, + ) diff --git a/backend/agenteval/evaluation/report.py b/backend/agenteval/evaluation/report.py index 255e7a7..8e3ba82 100644 --- a/backend/agenteval/evaluation/report.py +++ b/backend/agenteval/evaluation/report.py @@ -12,6 +12,7 @@ from typing import Any, Optional from sqlmodel import Session from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts +from agenteval.evaluation.go_no_go import evaluate_go_no_go from agenteval.evaluation.metrics import aggregate_runs from agenteval.evaluation.report_render import render_html, render_json, render_markdown from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary @@ -103,6 +104,21 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]: if judged_pass_rate is None and judged_total > 0: judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4) + summary_dict = { + "total_cases": total_cases, + "passed_cases": passed_cases, + "failed_cases": summary.failed_cases, + "total_rules": summary.total_rules, + "passed_rules": summary.passed_rules, + "pass_rate": summary.pass_rate if summary.pass_rate is not None else 0.0, + "connectivity_cases": connectivity_count, + "judged_pass_rate": judged_pass_rate, + "avg_latency_ms": summary.avg_latency_ms, + } + + # Generate go/no-go verdict + verdict = evaluate_go_no_go(summary_dict) + return { "run_id": run.id, "target_id": run.target_id, @@ -114,16 +130,8 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]: "status": run.status.value, "started_at": iso_utc(run.started_at), "completed_at": iso_utc(run.completed_at), - "summary": { - "total_cases": total_cases, - "passed_cases": passed_cases, - "failed_cases": summary.failed_cases, - "total_rules": summary.total_rules, - "passed_rules": summary.passed_rules, - "pass_rate": summary.pass_rate if summary.pass_rate is not None else 0.0, - "connectivity_cases": connectivity_count, - "judged_pass_rate": judged_pass_rate, - }, + "summary": summary_dict, + "go_no_go": verdict.model_dump(mode="json"), "cases": cases, } diff --git a/tests/unit/test_go_no_go.py b/tests/unit/test_go_no_go.py new file mode 100644 index 0000000..fd10f67 --- /dev/null +++ b/tests/unit/test_go_no_go.py @@ -0,0 +1,138 @@ +"""Tests for go/no-go acceptance verdict.""" + +import pytest + +from agenteval.evaluation.go_no_go import ( + AcceptanceCriteria, + CriterionResult, + GoNoGoVerdict, + evaluate_go_no_go, +) + + +def test_go_verdict_all_pass(): + """All criteria met -> go.""" + summary = { + "judged_pass_rate": 0.96, + "pass_rate": 0.95, + "avg_latency_ms": 2000, + } + criteria = AcceptanceCriteria( + judged_pass_rate_min=0.95, + pass_rate_min=0.90, + avg_latency_max_ms=5000, + ) + verdict = evaluate_go_no_go(summary, criteria) + assert verdict.decision == "go" + assert "达标" in verdict.summary + assert all(r.passed for r in verdict.criteria_results) + + +def test_no_go_verdict_core_failed(): + """Core metric (pass rate) failed -> no_go.""" + summary = { + "judged_pass_rate": 0.80, + "pass_rate": 0.75, + } + criteria = AcceptanceCriteria( + judged_pass_rate_min=0.95, + pass_rate_min=0.90, + ) + verdict = evaluate_go_no_go(summary, criteria) + assert verdict.decision == "no_go" + assert "不建议上线" in verdict.summary + assert any(not r.passed for r in verdict.criteria_results) + + +def test_conditional_verdict_non_core_risk(): + """Core passed but non-core (latency) failed -> conditional.""" + summary = { + "judged_pass_rate": 0.96, + "pass_rate": 0.95, + "avg_latency_ms": 8000, + } + criteria = AcceptanceCriteria( + judged_pass_rate_min=0.95, + pass_rate_min=0.90, + avg_latency_max_ms=5000, + ) + verdict = evaluate_go_no_go(summary, criteria) + assert verdict.decision == "conditional" + assert "风险" in verdict.summary + + +def test_default_criteria(): + """Default criteria should be applied when None.""" + summary = { + "judged_pass_rate": 0.96, + "pass_rate": 0.95, + } + verdict = evaluate_go_no_go(summary, None) + assert verdict.decision == "go" + assert len(verdict.criteria_results) >= 1 + + +def test_empty_summary(): + """Empty summary -> conditional with no results.""" + verdict = evaluate_go_no_go({}, None) + assert verdict.decision == "conditional" + assert "无可用指标" in verdict.summary + assert len(verdict.criteria_results) == 0 + + +def test_criterion_result_model(): + """CriterionResult model should work correctly.""" + result = CriterionResult( + criterion="pass_rate", + threshold=0.95, + actual=0.96, + passed=True, + detail="通过率 96% >= 95%", + ) + assert result.criterion == "pass_rate" + assert result.passed is True + + +def test_verdict_model(): + """GoNoGoVerdict model should work correctly.""" + verdict = GoNoGoVerdict( + decision="go", + summary="测试通过", + criteria_results=[], + ) + assert verdict.decision == "go" + assert verdict.generated_at is not None + + +def test_latency_only_when_configured(): + """Latency should only be checked when avg_latency_max_ms is set.""" + summary = { + "judged_pass_rate": 0.96, + "avg_latency_ms": 10000, # High latency + } + # Without latency threshold + criteria_no_latency = AcceptanceCriteria(judged_pass_rate_min=0.95) + verdict1 = evaluate_go_no_go(summary, criteria_no_latency) + assert verdict1.decision == "go" + assert all(r.criterion != "avg_latency_ms" for r in verdict1.criteria_results) + + # With latency threshold + criteria_with_latency = AcceptanceCriteria( + judged_pass_rate_min=0.95, + avg_latency_max_ms=5000, + ) + verdict2 = evaluate_go_no_go(summary, criteria_with_latency) + assert verdict2.decision == "conditional" + assert any(r.criterion == "avg_latency_ms" for r in verdict2.criteria_results) + + +def test_judged_pass_rate_fallback_to_pass_rate(): + """If judged_pass_rate is missing, fall back to pass_rate.""" + summary = {"pass_rate": 0.96} + criteria = AcceptanceCriteria(judged_pass_rate_min=0.95) + verdict = evaluate_go_no_go(summary, criteria) + assert verdict.decision == "go" + # Should have one result using pass_rate as judged_pass_rate + assert len(verdict.criteria_results) == 1 + assert verdict.criteria_results[0].criterion == "judged_pass_rate" + assert verdict.criteria_results[0].actual == 0.96