#!/usr/bin/env python3 """AI 管家金标准评测脚本。 支持三种模式： 1) lint-only：仅校验数据集结构与覆盖率。 2) online：调用 /rag/ask 实际评测。 3) offline：读取 answers 文件评测。 """ from __future__ import annotations import argparse import json import sys from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any from backend.app.utils.timezone import now_cn_aware try: import httpx except Exception: # pragma: no cover httpx = None @dataclass class EvalCase: id: str question: str expected_any: list[str] expected_all: list[str] forbidden_any: list[str] expected_query_type: str | None is_followup: bool group: str | None @dataclass class EvalResult: case_id: str passed: bool answer: str query_type: str score: float hit_any: bool hit_all: bool hit_query_type: bool forbidden_hit: bool hallucination_hit: bool expected_any: list[str] expected_all: list[str] forbidden_any: list[str] def _load_jsonl_cases(path: Path) -> list[EvalCase]: rows: list[EvalCase] = [] with path.open("r", encoding="utf-8") as f: for idx, raw in enumerate(f, start=1): line = raw.strip() if not line: continue data = json.loads(line) case_id = str(data.get("id") or f"L{idx:03d}") question = str(data.get("question") or "").strip() if not question: raise ValueError(f"line {idx} missing question") rows.append( EvalCase( id=case_id, question=question, expected_any=[str(x) for x in (data.get("expected_any") or []) if str(x).strip()], expected_all=[str(x) for x in (data.get("expected_all") or []) if str(x).strip()], forbidden_any=[str(x) for x in (data.get("forbidden_any") or []) if str(x).strip()], expected_query_type=(str(data.get("expected_query_type")).strip() if data.get("expected_query_type") else None), is_followup=bool(data.get("is_followup", False)), group=(str(data.get("group")).strip() if data.get("group") else None), ) ) return rows def _lint_cases(cases: list[EvalCase]) -> dict[str, Any]: unique_ids = len({c.id for c in cases}) followups = sum(1 for c in cases if c.is_followup) with_query_type = sum(1 for c in cases if c.expected_query_type) question_lens = [len(c.question) for c in cases] avg_len = round(sum(question_lens) / len(question_lens), 2) if question_lens else 0 return { "count": len(cases), "unique_ids": unique_ids, "followups": followups, "with_query_type": with_query_type, "avg_question_len": avg_len, "valid": len(cases) >= 50 and unique_ids == len(cases) and followups > 0, } def _load_offline_answers(path: Path) -> dict[str, dict[str, Any]]: payload = json.loads(path.read_text(encoding="utf-8")) if isinstance(payload, dict) and "answers" in payload and isinstance(payload["answers"], list): rows = payload["answers"] elif isinstance(payload, list): rows = payload else: raise ValueError("offline answers 必须是 list 或包含 answers 字段") result: dict[str, dict[str, Any]] = {} for row in rows: if not isinstance(row, dict): continue rid = str(row.get("id") or "").strip() if not rid: continue result[rid] = { "answer": str(row.get("answer") or ""), "query_type": str(row.get("query_type") or "unknown"), "confidence": float(row.get("confidence") or 0.0), } return result async def _ask_online( *, api_url: str, token: str | None, question: str, project_id: int | None, timeout_seconds: float, ) -> dict[str, Any]: if httpx is None: raise RuntimeError("缺少 httpx 依赖，无法执行在线评测") headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" body: dict[str, Any] = { "question": question, "asked_via": "pmflow", } if project_id is not None: body["project_id"] = int(project_id) async with httpx.AsyncClient(timeout=timeout_seconds, trust_env=False) as client: resp = await client.post(api_url, headers=headers, json=body) resp.raise_for_status() data = resp.json() if isinstance(data, dict) and "data" in data and isinstance(data.get("data"), dict): return data["data"] if isinstance(data, dict): return data return {} def _contains_any(text: str, keywords: list[str]) -> bool: if not keywords: return True lowered = text.lower() return any(str(k).lower() in lowered for k in keywords) def _contains_all(text: str, keywords: list[str]) -> bool: if not keywords: return True lowered = text.lower() return all(str(k).lower() in lowered for k in keywords) def _find_forbidden(text: str, keywords: list[str]) -> bool: if not keywords: return False lowered = text.lower() return any(str(k).lower() in lowered for k in keywords) def _is_hallucination(answer: str, forbidden: list[str]) -> bool: lowered = str(answer or "").lower() if _find_forbidden(lowered, forbidden): return True return not lowered.strip() def _evaluate_case(case: EvalCase, payload: dict[str, Any]) -> EvalResult: answer = str(payload.get("answer") or "").strip() query_type = str(payload.get("query_type") or "unknown").strip().lower() confidence = float(payload.get("confidence") or 0.0) hit_any = _contains_any(answer, case.expected_any) hit_all = _contains_all(answer, case.expected_all) forbidden_hit = _find_forbidden(answer, case.forbidden_any) hallucination_hit = _is_hallucination(answer, case.forbidden_any) hit_query_type = True if case.expected_query_type: hit_query_type = query_type == str(case.expected_query_type).strip().lower() passed = bool(hit_any and hit_all and (not forbidden_hit) and hit_query_type) score = 1.0 if passed else max(0.0, min(0.99, confidence * 0.6)) return EvalResult( case_id=case.id, passed=passed, answer=answer, query_type=query_type, score=round(score, 4), hit_any=hit_any, hit_all=hit_all, hit_query_type=hit_query_type, forbidden_hit=forbidden_hit, hallucination_hit=hallucination_hit, expected_any=case.expected_any, expected_all=case.expected_all, forbidden_any=case.forbidden_any, ) def _build_report( *, cases: list[EvalCase], results: list[EvalResult], thresholds: dict[str, float], mode: str, api_url: str | None, ) -> dict[str, Any]: total = len(results) passed = sum(1 for r in results if r.passed) hallucinations = sum(1 for r in results if r.hallucination_hit) follow_cases = [c for c in cases if c.is_followup] follow_ids = {c.id for c in follow_cases} follow_results = [r for r in results if r.case_id in follow_ids] follow_passed = sum(1 for r in follow_results if r.passed) qt_cases = [c for c in cases if c.expected_query_type] qt_ids = {c.id for c in qt_cases} qt_results = [r for r in results if r.case_id in qt_ids] qt_passed = sum(1 for r in qt_results if r.hit_query_type) accuracy = round((passed / total) if total else 0.0, 4) hallucination_rate = round((hallucinations / total) if total else 0.0, 4) followup_hit_rate = round((follow_passed / len(follow_results)) if follow_results else 0.0, 4) query_type_hit_rate = round((qt_passed / len(qt_results)) if qt_results else 0.0, 4) pass_gate = ( accuracy >= thresholds["accuracy"] and hallucination_rate <= thresholds["hallucination_rate"] and followup_hit_rate >= thresholds["followup_hit_rate"] and query_type_hit_rate >= thresholds["query_type_hit_rate"] ) return { "version": "v1", "generated_at": now_cn_aware().isoformat(), "mode": mode, "api_url": api_url, "thresholds": thresholds, "metrics": { "total": total, "passed": passed, "accuracy": accuracy, "hallucination_rate": hallucination_rate, "followup_total": len(follow_results), "followup_hit_rate": followup_hit_rate, "query_type_total": len(qt_results), "query_type_hit_rate": query_type_hit_rate, "pass_gate": pass_gate, }, "cases": [ { "id": r.case_id, "passed": r.passed, "score": r.score, "query_type": r.query_type, "hit_any": r.hit_any, "hit_all": r.hit_all, "hit_query_type": r.hit_query_type, "forbidden_hit": r.forbidden_hit, "hallucination_hit": r.hallucination_hit, "answer": r.answer, "expected_any": r.expected_any, "expected_all": r.expected_all, "forbidden_any": r.forbidden_any, } for r in results ], } def _write_report(report: dict[str, Any], output_dir: Path) -> Path: output_dir.mkdir(parents=True, exist_ok=True) ts = now_cn_aware().strftime("%Y%m%dT%H%M%SZ") report_path = output_dir / f"ai_steward_eval_{ts}.json" report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") latest = output_dir / "latest.json" latest.write_text(report_path.read_text(encoding="utf-8"), encoding="utf-8") return report_path def _load_latest_report(output_dir: Path) -> dict[str, Any] | None: latest = output_dir / "latest.json" if not latest.exists(): return None try: return json.loads(latest.read_text(encoding="utf-8")) except Exception: return None def _print_summary(report: dict[str, Any], previous: dict[str, Any] | None) -> None: metrics = report.get("metrics", {}) print("====") print(f"total={metrics.get('total')} passed={metrics.get('passed')}") print(f"accuracy={metrics.get('accuracy')}") print(f"hallucination_rate={metrics.get('hallucination_rate')}") print(f"followup_hit_rate={metrics.get('followup_hit_rate')}") print(f"query_type_hit_rate={metrics.get('query_type_hit_rate')}") print(f"pass_gate={metrics.get('pass_gate')}") if previous: prev_m = previous.get("metrics", {}) print("----") print(f"prev_accuracy={prev_m.get('accuracy')}") print(f"prev_hallucination_rate={prev_m.get('hallucination_rate')}") print(f"prev_followup_hit_rate={prev_m.get('followup_hit_rate')}") print(">>>> REPLACE") async def _run(args: argparse.Namespace) -> int: input_path = Path(args.input).resolve() if not input_path.exists(): print(f"[AUDIT-WARN] golden set not found: {input_path}") return 2 cases = _load_jsonl_cases(input_path) lint = _lint_cases(cases) if args.lint_only: print(json.dumps({"lint": lint}, ensure_ascii=False, indent=2)) return 0 if lint.get("valid") else 2 if not lint.get("valid"): print(json.dumps({"lint": lint, "error": "golden set invalid"}, ensure_ascii=False, indent=2)) return 2 offline_answers: dict[str, dict[str, Any]] = {} mode = "online" api_url = args.api_url if args.answers_file: mode = "offline" offline_answers = _load_offline_answers(Path(args.answers_file).resolve()) api_url = None results: list[EvalResult] = [] for case in cases: if mode == "offline": payload = offline_answers.get(case.id, {"answer": "", "query_type": "unknown", "confidence": 0.0}) else: payload = await _ask_online( api_url=args.api_url, token=args.token, question=case.question, project_id=args.project_id, timeout_seconds=args.timeout, ) results.append(_evaluate_case(case, payload)) thresholds = { "accuracy": float(args.threshold_accuracy), "hallucination_rate": float(args.threshold_hallucination), "followup_hit_rate": float(args.threshold_followup), "query_type_hit_rate": float(args.threshold_query_type), } output_dir = Path(args.output_dir).resolve() previous = _load_latest_report(output_dir) report = _build_report(cases=cases, results=results, thresholds=thresholds, mode=mode, api_url=api_url) report_path = _write_report(report, output_dir) print(f"report={report_path}") _print_summary(report, previous) if args.fail_on_threshold and not bool(report.get("metrics", {}).get("pass_gate", False)): return 2 return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="AI 管家评测与门禁") parser.add_argument("--input", default="backend/evals/golden_qa_v1.jsonl") parser.add_argument("--api-url", default="http://127.0.0.1:8000/rag/ask") parser.add_argument("--token", default=None) parser.add_argument("--project-id", type=int, default=None) parser.add_argument("--timeout", type=float, default=45.0) parser.add_argument("--answers-file", default=None) parser.add_argument("--output-dir", default="backend/evals/reports") parser.add_argument("--lint-only", action="store_true") parser.add_argument("--fail-on-threshold", action="store_true") parser.add_argument("--threshold-accuracy", type=float, default=0.80) parser.add_argument("--threshold-hallucination", type=float, default=0.12) parser.add_argument("--threshold-followup", type=float, default=0.75) parser.add_argument("--threshold-query-type", type=float, default=0.80) return parser def main() -> int: parser = build_parser() args = parser.parse_args() try: import asyncio return asyncio.run(_run(args)) except KeyboardInterrupt: return 130 except Exception as exc: print(f"[AUDIT-WARN] eval failed: {exc}") return 2 if __name__ == "__main__": raise SystemExit(main())