135 lines
6.0 KiB
Python
135 lines
6.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Template-Ti 一期 LLM 场景层单元测试（EPIC #11）。
|
||
|
||
覆盖：
|
||
1. 场景配置资产可解析且含验收指标（route_accuracy ≥ 96.5%）；
|
||
2. 三个业务场景端到端走通（报警解释 / 交接班摘要 / NL 查询驾驶舱），
|
||
答案带 RAG 引用溯源；
|
||
3. 路由准确率 ≥ 96.5%（Ti 场景查询集，验收指标）；
|
||
4. 幻觉校验：报警解释为高利害场景，低信度转人工（needs_human）。
|
||
"""
|
||
import os
|
||
import sys
|
||
import unittest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import _bootstrap # noqa: F401
|
||
|
||
from llm_gateway.router import RouteTarget, SensitivityRouter # noqa: E402
|
||
from ti_scenarios.scenarios import ( # noqa: E402
|
||
TiScenarioRunner,
|
||
router_config_path,
|
||
scenarios_config_path,
|
||
)
|
||
|
||
|
||
def _load_yaml(path):
|
||
import yaml
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return yaml.safe_load(fh) or {}
|
||
|
||
|
||
class TestScenarioConfig(unittest.TestCase):
|
||
"""场景配置资产：三场景齐备且验收指标达标。"""
|
||
|
||
def setUp(self):
|
||
self.cfg = _load_yaml(scenarios_config_path())
|
||
|
||
def test_three_scenarios_defined(self):
|
||
names = {s["name"] for s in self.cfg["scenarios"]}
|
||
self.assertEqual(names, {"alarm_explain", "shift_handover", "nl_query"})
|
||
|
||
def test_alarm_explain_acceptance_ge_96p5(self):
|
||
for s in self.cfg["scenarios"]:
|
||
if s["name"] == "alarm_explain":
|
||
self.assertGreaterEqual(s["acceptance"]["route_accuracy"], 0.965)
|
||
self.assertTrue(s["acceptance"]["hallucination_check"])
|
||
self.assertTrue(s["high_stakes"])
|
||
|
||
|
||
class TestRouteAccuracy(unittest.TestCase):
|
||
"""验收：路由准确率 ≥ 96.5%（Ti 场景行业规则 + 内核保底规则）。"""
|
||
|
||
ROUTE_CASES = [
|
||
# (query, expected_target, 说明)
|
||
("氯气流量是多少", RouteTarget.LOCAL, "工艺敏感参数（行业规则）"),
|
||
("炉温现在多少", RouteTarget.LOCAL, "工艺敏感参数（行业规则）"),
|
||
("加料比如何调整", RouteTarget.LOCAL, "工艺配比参数（行业规则）"),
|
||
("钛纯度合格标准", RouteTarget.LOCAL, "产品质量指标（行业规则）"),
|
||
("紧急停机", RouteTarget.BLOCK, "安全指令（默认保底规则）"),
|
||
("氯气泄漏立即停机", RouteTarget.BLOCK, "安全指令（默认保底规则）"),
|
||
("海绵钛是什么", RouteTarget.CLOUD, "公开常识（行业规则）"),
|
||
("身份证号 110101199001011234 是什么", RouteTarget.LOCAL, "PII（默认保底规则）"),
|
||
("电话 13800138000 查一下", RouteTarget.LOCAL, "PII（默认保底规则）"),
|
||
("今天车间排产如何安排", RouteTarget.LOCAL, "通用管理问题（默认本地）"),
|
||
("氯气流量偏低怎么处理", RouteTarget.LOCAL, "工艺敏感参数"),
|
||
("炉温超限怎么处置", RouteTarget.LOCAL, "工艺敏感参数"),
|
||
("海绵钛纯度检验标准", RouteTarget.LOCAL, "产品质量指标"),
|
||
("交接班注意事项", RouteTarget.LOCAL, "管理流程（默认本地）"),
|
||
("驾驶舱能看到哪些指标", RouteTarget.LOCAL, "驾驶舱查询（默认本地）"),
|
||
("钛锭强度如何", RouteTarget.LOCAL, "工艺/产品参数"),
|
||
("氯化炉操作手册要点", RouteTarget.LOCAL, "工艺文档查询"),
|
||
("国标氯气安全要求", RouteTarget.LOCAL, "标准文档查询"),
|
||
("停机按钮在哪", RouteTarget.BLOCK, "安全指令关键字"),
|
||
("海绵钛和钛合金区别", RouteTarget.LOCAL, "工艺/产品对比（保守默认本地）"),
|
||
]
|
||
|
||
def setUp(self):
|
||
self.router = SensitivityRouter.from_template_config(router_config_path())
|
||
|
||
def test_route_accuracy_ge_96p5(self):
|
||
total = len(self.ROUTE_CASES)
|
||
hit = 0
|
||
for query, expected, desc in self.ROUTE_CASES:
|
||
decision = self.router.route(query)
|
||
if decision.target == expected:
|
||
hit += 1
|
||
else:
|
||
print(f"[route 偏差] {desc} | {query!r} -> {decision.target}（期望 {expected}）")
|
||
accuracy = hit / total
|
||
self.assertGreaterEqual(accuracy, 0.965,
|
||
f"路由准确率 {accuracy:.1%} < 96.5%（{hit}/{total}）")
|
||
|
||
|
||
class TestScenariosE2E(unittest.TestCase):
|
||
"""三个场景端到端（复用 llm-gateway 主编排 + 领域 RAG + 演示文档）。"""
|
||
|
||
def setUp(self):
|
||
self.runner = TiScenarioRunner()
|
||
|
||
def test_explain_alarm_with_sources(self):
|
||
result = self.runner.explain_alarm("炉温超上限报警怎么处理")
|
||
self.assertIn("报警", result.query)
|
||
self.assertTrue(result.answer)
|
||
# 高利害场景应带 SOP 引用溯源（答案回显来源）
|
||
self.assertIn("来源", result.answer)
|
||
self.assertEqual(result.route.target, RouteTarget.LOCAL)
|
||
|
||
def test_shift_handover(self):
|
||
result = self.runner.generate_handover("甲班：生产平稳，炉温正常，无异常事项")
|
||
self.assertTrue(result.answer)
|
||
self.assertTrue(result.answer_id)
|
||
|
||
def test_nl_query_cockpit(self):
|
||
result = self.runner.query_cockpit("查询最近一小时的氯气流量趋势")
|
||
self.assertTrue(result.answer)
|
||
self.assertEqual(result.route.target, RouteTarget.LOCAL)
|
||
|
||
def test_high_stakes_alarm_low_confidence_needs_human(self):
|
||
"""幻觉校验：报警解释低信度 → 转人工（needs_human=True）。"""
|
||
result = self.runner.explain_alarm("炉温超上限报警", confidence=0.2)
|
||
self.assertTrue(result.needs_human,
|
||
"高利害场景低信度应转人工确认")
|
||
# 高信度正常通过
|
||
ok = self.runner.explain_alarm("炉温超上限报警", confidence=0.95)
|
||
self.assertFalse(ok.needs_human)
|
||
|
||
def test_drain_audits_available(self):
|
||
self.runner.explain_alarm("氯气流量报警")
|
||
audits = self.runner.drain_audits()
|
||
self.assertEqual(set(audits), {"dlp", "router", "prompts", "guard"})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|