#!/usr/bin/env python3 """AI 管家金标准评测脚本。 支持三种模式： 1) lint-only：仅校验数据集结构与覆盖率。 2) online：调用 /rag/ask 实际评测。 3) offline：读取 answers 文件评测。 """ from __future__ import annotations import argparse import json import os 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=f"http://localhost:{os.getenv('BACKEND_PORT', '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())