feat: 完成 issue #49 ④ 路由准确率评估脚本(≥96.5%)

This commit is contained in:
2026-08-05 02:45:00 +08:00
parent c5a7a0ee52
commit 437771da34
4 changed files with 319 additions and 6 deletions
+72 -5
View File
@@ -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: