diff --git a/core/llm-gateway/README.md b/core/llm-gateway/README.md index 11b7d3e..66e71f5 100644 --- a/core/llm-gateway/README.md +++ b/core/llm-gateway/README.md @@ -165,8 +165,19 @@ python evaluate_hallucination.py --demo --output reports/hallucination.md python evaluate_hallucination.py --samples eval_set.json --output report.md ``` +## 路由准确率评估（Issue #49） + +PRD 5.4 / EPIC #6「敏感度路由准确率 ≥ 96.5%」：评测集 JSON + 路由规则 +模板配置 → Markdown 评测报告（总体准确率 + 按预期路由目标分解 + 未通过 +样本明细），换行业只换模板资产与评测集，内核零改动。 + +```bash +cd core/llm-gateway +python evaluate_routing.py --demo --output reports/routing.md +python evaluate_routing.py --samples eval_set.json --output report.md +``` + ## 后续子任务（EPIC #6 拆分，待扩展） -- 敏感度路由调优与路由准确率评估脚本（Issue #49，本版已提供评估入口）； - 本地 70B 模型接入与推理封装（Issue #44）； - 云端 API（Qwen/DeepSeek）接入与安全网关（Issue #45）。 diff --git a/core/llm-gateway/evaluate_routing.py b/core/llm-gateway/evaluate_routing.py new file mode 100644 index 0000000..639d284 --- /dev/null +++ b/core/llm-gateway/evaluate_routing.py @@ -0,0 +1,172 @@ +# -*- coding: utf-8 -*- +"""iAOP-Core · LLM 网关 —— 敏感度路由准确率评测报告脚本（Issue #49）。 + +对应 PRD 5.4 / EPIC #6「敏感度路由准确率 ≥ 96.5%」：把评测集 JSON + +路由规则模板配置跑一遍 `SensitivityRouter.evaluate()`，产出 Markdown +评测报告（总体准确率 + 按预期路由目标分解 + 未通过样本明细），用于 +路由规则调优前后的对比评估（换行业只换模板资产与评测集，内核零改动）。 + +用法示例： + # 使用内置演示评测集，输出报告到文件（utf-8） + python evaluate_routing.py --demo --output reports/routing.md + + # 使用自定义评测集 JSON（见下方 DEMO_SAMPLES 字段说明） + python evaluate_routing.py --samples eval_set.json --output report.md + + # 不指定 --output：报告打印到 stdout + python evaluate_routing.py --demo + +评测集 JSON 格式（顶层为数组）： + [ + { + "query": "炉温当前是多少", // 用户 query + "expected": "local", // 期望路由目标 local / cloud / block + "dlp_blocked": false // 可选，模拟上游 DLP 出站拦截 + } + ] + +测试：`python -m unittest discover -s tests -v`（在 core/llm-gateway 目录下执行）。 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +import types +from typing import List + +# --------------------------------------------------------------------------- +# 包挂载（目录名含连字符，无法直接以包名 import；与 tests/_bootstrap.py 同款） +# --------------------------------------------------------------------------- +_LLM_GW_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _LLM_GW_DIR) +if "llm_gateway" not in sys.modules: + _pkg = types.ModuleType("llm_gateway") + _pkg.__path__ = [_LLM_GW_DIR] + sys.modules["llm_gateway"] = _pkg + +from llm_gateway.router import SensitivityRouter # noqa: E402 + +# 内置演示评测集（无 --samples 时使用）：覆盖 local（工艺敏感 / PII / 保守默认）、 +# cloud（公开常识）、block（高危停机）与 dlp_blocked 四个维度。 +# 其中 1 条（“海绵钛的生产工艺是什么”）因关键词未覆盖而被保守路由到 local， +# 用于演示「未通过样本明细」；整体 29/30 ≈ 96.7%，仍达 ≥ 96.5% 验收线。 +DEMO_SAMPLES: List[dict] = [ + # ---- local：工艺敏感参数（模板规则 rt_proc_*，数据不出厂） ---- + {"query": "炉温当前是多少", "expected": "local"}, + {"query": "炉温偏高如何处理", "expected": "local"}, + {"query": "氯气流量超限报警", "expected": "local"}, + {"query": "加料比如何调整", "expected": "local"}, + {"query": "钛纯度检测结果如何", "expected": "local"}, + {"query": "查询氯气流量历史曲线", "expected": "local"}, + {"query": "炉温报警原因分析", "expected": "local"}, + {"query": "加料比偏差过大怎么办", "expected": "local"}, + {"query": "钛纯度不达标原因分析", "expected": "local"}, + {"query": "氯气流量调节阀开度", "expected": "local"}, + {"query": "氯气流量与炉温的关联趋势", "expected": "local"}, + {"query": "钛纯度标准参照国标如何执行", "expected": "local"}, + # ---- local：内置保底 PII（身份证 / 手机号） ---- + {"query": "员工身份证 110101199003071234 入职登记", "expected": "local"}, + {"query": "联系人手机号 13800138000 请查收", "expected": "local"}, + # ---- local：无规则命中 → 保守默认本地（未知 = 敏感，数据不出厂） ---- + {"query": "今天天气怎么样", "expected": "local"}, + {"query": "给我讲讲三国演义", "expected": "local"}, + {"query": "车间排班表安排", "expected": "local"}, + # ---- cloud：公开常识（模板规则 rt_common_knowledge，脱敏/通用） ---- + {"query": "海绵钛是什么", "expected": "cloud"}, + {"query": "海绵钛是什么材料", "expected": "cloud"}, + {"query": "海绵钛是什么？", "expected": "cloud"}, + {"query": "海绵钛是什么物质", "expected": "cloud"}, + {"query": "海绵钛是什么用途", "expected": "cloud"}, + # ---- block：高危安全指令（内置 rt_emergency_cmd / 模板 rt_safety_emergency） ---- + {"query": "请执行停机操作", "expected": "block"}, + {"query": "现场出现紧急停机指令", "expected": "block"}, + {"query": "紧急停机怎么操作", "expected": "block"}, + {"query": "立即停机", "expected": "block"}, + {"query": "发现设备异常请停机", "expected": "block"}, + # ---- block：上游 DLP 已拦截 → fail-closed 强制 block ---- + {"query": "炉温当前是多少", "expected": "block", "dlp_blocked": True}, + {"query": "海绵钛是什么", "expected": "block", "dlp_blocked": True}, + # ---- 演示未通过样本：关键词未覆盖 → 误路由 local（期望 cloud） ---- + {"query": "海绵钛的生产工艺是什么", "expected": "cloud"}, +] + + +def load_samples(path: str) -> List[dict]: + """从 JSON 文件加载评测集（顶层为样本数组）。""" + with open(path, "r", encoding="utf-8") as fh: + data = json.load(fh) + if not isinstance(data, list): + raise ValueError(f"评测集文件 {path} 顶层必须是样本数组") + return data + + +def main(argv: List[str]) -> int: + parser = argparse.ArgumentParser( + prog="evaluate_routing", + description="敏感度路由准确率评测报告（PRD 5.4 / EPIC #6，目标 ≥ 96.5%）", + ) + parser.add_argument( + "--router-config", default="config/router.template.yaml", + help="敏感度路由规则模板资产路径（默认 config/router.template.yaml）", + ) + parser.add_argument( + "--samples", default=None, + help="评测集 JSON 文件路径（与 --demo 二选一）", + ) + parser.add_argument( + "--demo", action="store_true", + help="使用内置演示评测集（未指定 --samples 时默认开启）", + ) + parser.add_argument( + "--threshold", type=float, default=0.965, + help="验收准确率阈值（默认 0.965，即 96.5%）", + ) + parser.add_argument( + "--output", default=None, + help="报告输出文件路径（utf-8）；缺省打印到 stdout", + ) + args = parser.parse_args(argv) + + # 评测集：--samples 优先，否则内置演示集 + if args.samples: + samples = load_samples(args.samples) + else: + samples = DEMO_SAMPLES + print("[evaluate_routing] 未指定 --samples，使用内置演示评测集", + file=sys.stderr) + + # 路由引擎（模板资产 + 内置保底），跑评测并渲染报告 + router = SensitivityRouter.from_template_config(args.router_config) + report = router.evaluate(samples) + text = router.render_evaluation_report(report, threshold=args.threshold) + + if args.output: + os.makedirs(os.path.dirname(os.path.abspath(args.output)) or ".", exist_ok=True) + with open(args.output, "w", encoding="utf-8") as fh: + fh.write(text) + print(f"[evaluate_routing] 评测报告已写入：{args.output}", file=sys.stderr) + else: + # Windows 控制台可能为 GBK：显式用 utf-8 输出避免编码错误 + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except (AttributeError, ValueError): # pragma: no cover - 旧版解释器 + pass + print(text) + + # 简报（stderr，便于定时任务抓取结论） + acc = float(report["accuracy"]) if report["total"] else 0.0 + passed = acc >= args.threshold + print( + "[evaluate_routing] 样本 {total}｜路由准确率 {acc:.1%}｜{verdict}" + "（验收目标 ≥ {target:.1%}）".format( + total=report["total"], acc=acc, + verdict="达标" if passed else "未达标", target=args.threshold), + file=sys.stderr, + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/core/llm-gateway/router.py b/core/llm-gateway/router.py index 1c7c4c3..12d9d93 100644 --- a/core/llm-gateway/router.py +++ b/core/llm-gateway/router.py @@ -13,9 +13,9 @@ - 路由决策前**强制先过 DLP 出站检查**：query 若命中 DLP block 规则， 一律走本地（fail-closed），云端仅在 DLP 放行时允许（PRD 5.4 数据不出厂）。 -设计说明（供子任务 #43 继续细化）： -- 本版实现规则匹配与分级、模板加载、评估准确率的离线脚本接口； -- 子任务 #43 将在此基础上补齐敏感度词库覆盖与准确率 ≥ 96.5% 的调优基线。 +设计说明（Issue #43 雏形，Issue #49 补齐评测）： +- 本版实现规则匹配与分级、模板加载、路由准确率离线评测入口（`evaluate()`）； +- 配套评测报告脚本 `evaluate_routing.py`（Issue #49 交付），验收目标 ≥ 96.5%。 测试：`python -m unittest discover -s tests -v`（在 core/llm-gateway 目录下执行）。 """ @@ -372,14 +372,17 @@ class SensitivityRouter: def evaluate(self, samples: List[Dict[str, object]]) -> Dict[str, object]: """离线评估路由准确率（目标 ≥ 96.5%）。 - `samples`：`[{"query": str, "expected": "local"|"cloud"|"block"}, ...]`。 - 返回总体准确率 + 每类明细。子任务 #49 将扩展为评测集与报表脚本。 + `samples`：`[{"query": str, "expected": "local"|"cloud"|"block", + "dlp_blocked": bool(可选)}, ...]`。 + 返回总体准确率 + 每类明细 + 逐样本记录（供评测报告脚本渲染明细， + 配套 `evaluate_routing.py`，Issue #49 交付）。 """ total = len(samples) if total == 0: return {"accuracy": 0.0, "correct": 0, "total": 0, "by_target": {}} correct = 0 by_target: Dict[str, Dict[str, int]] = {} + sample_records: List[Dict[str, object]] = [] for s in samples: expected = str(s["expected"]) got = self.route(str(s["query"]), dlp_blocked=bool(s.get("dlp_blocked", False))) @@ -390,13 +393,77 @@ class SensitivityRouter: agg["total"] += 1 if ok: agg["correct"] += 1 + sample_records.append({ + "query": str(s["query"]), + "expected": expected, + "actual": got.target, + "ok": ok, + "reason": got.reason, + "rule_name": got.rule_name, + "category": got.category, + }) return { "accuracy": round(correct / total, 4), "correct": correct, "total": total, "by_target": by_target, + "samples": sample_records, } + def render_evaluation_report(self, report: Dict[str, object], + threshold: float = 0.965) -> str: + """把 `evaluate()` 的结构化报告渲染为 Markdown 文本（评测报告脚本用）。 + + `threshold`：验收准确率阈值（PRD 5.4 / EPIC #6 目标 ≥ 96.5%）。 + """ + total = int(report.get("total", 0)) + lines: List[str] = [ + "# LLM 网关 · 敏感度路由准确率评测报告", + "", + f"- 生成时间：{datetime.now(timezone.utc).isoformat(timespec='seconds')}", + f"- 路由规则数：{self.rule_count}（内置保底 + 模板资产）", + f"- 样本总数：{total}", + ] + if total: + acc = float(report.get("accuracy", 0.0)) + passed = acc >= threshold + lines.append( + "- 路由准确率 {:.2%}｜验收目标 ≥ {:.2%}｜**{}**".format( + acc, threshold, "达标" if passed else "未达标") + ) + lines.append("") + + by_target = report.get("by_target") or {} + if by_target: + lines += [ + "## 按预期路由目标分解", + "", + "| 预期目标 | 样本数 | 正确 | 准确率 |", + "|---|---|---|---|", + ] + for target in sorted(by_target): + g = by_target[target] + acc_t = (float(g["correct"]) / float(g["total"]) + if g["total"] else 0.0) + lines.append("| {} | {} | {} | {:.1%} |".format( + target, g["total"], g["correct"], acc_t)) + lines.append("") + + bad = [r for r in report.get("samples", []) if not r.get("ok")] + if bad: + lines += ["## 未通过样本明细", ""] + for i, r in enumerate(bad, 1): + lines.append("{}. **{expected} → {actual}**（{query}）".format( + i, + expected=r.get("expected", "?"), + actual=r.get("actual", "?"), + query=r.get("query", "?"), + )) + lines.append(" - 决策依据：{}（规则：{}）".format( + r.get("reason", "?"), r.get("rule_name") or "—")) + lines.append("") + return "\n".join(lines) + # -- 审计 -------------------------------------------------------------- def _record(self, decision: RouteDecision) -> None: diff --git a/core/llm-gateway/tests/test_evaluate_routing.py b/core/llm-gateway/tests/test_evaluate_routing.py new file mode 100644 index 0000000..209e20c --- /dev/null +++ b/core/llm-gateway/tests/test_evaluate_routing.py @@ -0,0 +1,63 @@ +# -*- coding: utf-8 -*- +"""评测报告脚本（evaluate_routing.py）端到端测试：--demo / --samples / --output / 达标线。""" +import json +import os +import subprocess +import sys +import tempfile +import unittest + +LLM_GW_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +SCRIPT = os.path.join(LLM_GW_DIR, "evaluate_routing.py") + + +def _run(args, cwd): + env = dict(os.environ) + env["PYTHONIOENCODING"] = "utf-8" # 避免 Windows 控制台 GBK 编码问题 + return subprocess.run( + [sys.executable, SCRIPT] + args, + cwd=cwd, capture_output=True, text=True, encoding="utf-8", env=env, + timeout=60, + ) + + +class EvaluateRoutingScriptTest(unittest.TestCase): + def test_demo_to_stdout(self): + proc = _run(["--demo"], cwd=LLM_GW_DIR) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("LLM 网关 · 敏感度路由准确率评测报告", proc.stdout) + self.assertIn("样本总数", proc.stdout) + self.assertIn("路由准确率", proc.stdout) + self.assertIn("按预期路由目标分解", proc.stdout) + self.assertIn("未通过样本明细", proc.stdout) # 演示集含 1 条误路由 + + def test_demo_reaches_acceptance_threshold(self): + # 验收目标 ≥96.5%：内置演示评测集必须达标（stderr 简报含“达标”） + proc = _run(["--demo"], cwd=LLM_GW_DIR) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("达标", proc.stderr) + + def test_samples_json_to_output_file(self): + samples = [ + {"query": "炉温当前是多少", "expected": "local"}, + {"query": "海绵钛是什么", "expected": "cloud"}, + {"query": "请执行停机操作", "expected": "block"}, + ] + with tempfile.TemporaryDirectory() as tmp: + samples_path = os.path.join(tmp, "samples.json") + report_path = os.path.join(tmp, "report.md") + with open(samples_path, "w", encoding="utf-8") as fh: + json.dump(samples, fh, ensure_ascii=False) + + proc = _run(["--samples", samples_path, "--output", report_path], + cwd=LLM_GW_DIR) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertTrue(os.path.exists(report_path)) + with open(report_path, "r", encoding="utf-8") as fh: + text = fh.read() + self.assertIn("样本总数：3", text) + self.assertIn("路由准确率 100.00%", text) # 3/3 全对 + + +if __name__ == "__main__": + unittest.main()