PMFlow/scripts/ai_steward_eval_compare.py

91 lines
3.0 KiB
Python
Executable File

#!/usr/bin/env python3
"""比较两份 AI 管家评测报告。"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
def _load(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def _eval_metric(report: dict, key: str) -> float:
return float(report.get("metrics", {}).get(key, 0.0) or 0.0)
def _resolve_path(payload: dict[str, Any], path: str) -> Any:
node: Any = payload
for part in [x for x in str(path or "").split(".") if x]:
if not isinstance(node, dict):
return None
node = node.get(part)
return node
def _pressure_metric(report: dict, key: str) -> float:
value = _resolve_path(report.get("summary") if isinstance(report.get("summary"), dict) else {}, key)
if value is None:
value = _resolve_path(report, key)
try:
return float(value or 0.0)
except Exception:
return 0.0
def _is_pressure_report(report: dict) -> bool:
return isinstance(report.get("summary"), dict) and "total_turns" in report.get("summary", {})
def _is_eval_report(report: dict) -> bool:
return isinstance(report.get("metrics"), dict) and "accuracy" in report.get("metrics", {})
def main() -> int:
parser = argparse.ArgumentParser(description="compare ai steward eval reports")
parser.add_argument("--base", required=True)
parser.add_argument("--target", required=True)
args = parser.parse_args()
base = _load(Path(args.base).resolve())
target = _load(Path(args.target).resolve())
print("====")
if _is_pressure_report(base) and _is_pressure_report(target):
keys = [
"success_rate",
"template_leak_rate",
"deep_hit_rate",
"question_requirement_hit_rate",
"endpoint_latency_ms.p95",
"backend_response_time_ms.p95",
]
for key in keys:
b = _pressure_metric(base, key)
t = _pressure_metric(target, key)
delta = round(t - b, 4)
print(f"{key}: base={b:.4f} target={t:.4f} delta={delta:+.4f}")
print(f"base_mode={str(base.get('summary', {}).get('mode') or '')}")
print(f"target_mode={str(target.get('summary', {}).get('mode') or '')}")
elif _is_eval_report(base) and _is_eval_report(target):
keys = ["accuracy", "hallucination_rate", "followup_hit_rate", "query_type_hit_rate"]
for key in keys:
b = _eval_metric(base, key)
t = _eval_metric(target, key)
delta = round(t - b, 4)
print(f"{key}: base={b:.4f} target={t:.4f} delta={delta:+.4f}")
print(f"base_pass={bool(base.get('metrics', {}).get('pass_gate', False))}")
print(f"target_pass={bool(target.get('metrics', {}).get('pass_gate', False))}")
else:
print("incompatible_report_type: expected both pressure or both eval report")
return 2
print(">>>> REPLACE")
return 0
if __name__ == "__main__":
raise SystemExit(main())