feat(report): 上线验收报告 go/no-go 结论
All checks were successful
CI / test (pull_request) Successful in 4m7s

新增纯函数模块 evaluation/go_no_go.py,基于 RunSummary 指标自动生成
go/no-go/conditional 三级结论。

- AcceptanceCriteria 模型:judged_pass_rate_min、pass_rate_min、avg_latency_max_ms
- GoNoGoVerdict 模型:decision、summary、criteria_results
- evaluate_go_no_go() 纯函数:接收 summary dict + criteria -> 返回 verdict
- 集成到 generate_report(),报告 dict 自动附带 go_no_go 字段
- 三级结论:go(全部达标)、no_go(核心未达标)、conditional(有风险)
- 新增 9 项单元测试(642 tests passed)

Closes #20
This commit is contained in:
sinohqb 2026-08-25 14:28:55 +08:00
parent 5827c3d3f5
commit 2314ebe3bc
3 changed files with 293 additions and 10 deletions

View File

@ -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,
)

View File

@ -12,6 +12,7 @@ from typing import Any, Optional
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts 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.metrics import aggregate_runs
from agenteval.evaluation.report_render import render_html, render_json, render_markdown from agenteval.evaluation.report_render import render_html, render_json, render_markdown
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary 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: if judged_pass_rate is None and judged_total > 0:
judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4) 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 { return {
"run_id": run.id, "run_id": run.id,
"target_id": run.target_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, "status": run.status.value,
"started_at": iso_utc(run.started_at), "started_at": iso_utc(run.started_at),
"completed_at": iso_utc(run.completed_at), "completed_at": iso_utc(run.completed_at),
"summary": { "summary": summary_dict,
"total_cases": total_cases, "go_no_go": verdict.model_dump(mode="json"),
"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,
},
"cases": cases, "cases": cases,
} }

138
tests/unit/test_go_no_go.py Normal file
View File

@ -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