diff --git a/templates/ti-cl4/llm-scenarios/config/alarm_explain.template.yaml b/templates/ti-cl4/llm-scenarios/config/alarm_explain.template.yaml new file mode 100644 index 0000000..970efae --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/config/alarm_explain.template.yaml @@ -0,0 +1,22 @@ +# -*- coding: utf-8 -*- +# 模板「报警解释」场景配置资产:ti-cl4(Template-Ti 一期,issue #74)。 +# +# 说明(父 Issue #11「④ LLM 报警解释 / 交接班 / NL 查询」子任务): +# - prompt_template:绑定 llm-gateway 提示词版本库(prompts.template.yaml) +# 中的 alarm_explain 模板(报警解释 + SOP 引用); +# - rag.categories:报警解释**只检索 sop 知识域**(异常处置 SOP), +# 不引入工艺规范/国标噪声,保证引用精准; +# - rag.min_score:检索阈值(TF 密度分),低于阈值视为未命中 → +# 提示"未检索到相关 SOP,请人工确认"(高利害场景宁缺毋滥,PRD 5.4); +# - high_stakes + confidence_threshold:低信度转人工确认(与 #11 验收一致)。 +template: ti-cl4 +version: 1.0.0 + +alarm_explain: + prompt_template: alarm_explain + rag: + categories: [sop] # 报警解释知识域:仅异常处置 SOP + top_k: 3 + min_score: 0.3 # 检索阈值(低于 = 未命中,降级提示人工) + high_stakes: true + confidence_threshold: 0.8 # 低信度转人工 diff --git a/templates/ti-cl4/llm-scenarios/scenarios.py b/templates/ti-cl4/llm-scenarios/scenarios.py index 48af168..9b960cd 100644 --- a/templates/ti-cl4/llm-scenarios/scenarios.py +++ b/templates/ti-cl4/llm-scenarios/scenarios.py @@ -19,7 +19,7 @@ from typing import Callable, Dict, List, Optional from llm_gateway.gateway import GatewayResult, LLMGateway from llm_gateway.prompts import PromptRegistry from llm_gateway.router import SensitivityRouter -from rag_kb import RagKnowledgeBase, load_kb_config +from rag_kb import KnowledgeSourceKind, RagKnowledgeBase, load_kb_config # --------------------------------------------------------------------------- # 场景配置资产路径 @@ -41,6 +41,12 @@ def scenarios_config_path() -> str: return os.path.join(_scenarios_dir(), "config", "scenarios.template.yaml") +def alarm_config_path() -> str: + """报警解释场景配置资产(alarm_explain.template.yaml)路径。""" + return os.path.join(_scenarios_dir(), "config", + "alarm_explain.template.yaml") + + def prompts_config_path() -> str: """复用 llm-gateway 提示词版本库资产路径。""" return os.path.join(_repo_root(), "core", "llm-gateway", "config", @@ -110,6 +116,7 @@ class TiScenarioRunner: prompts: Optional[PromptRegistry] = None, router: Optional[SensitivityRouter] = None, kb_loader: Callable[[str], str] = demo_kb_loader, + alarm_config: Optional[dict] = None, ) -> None: prompts = prompts or PromptRegistry.from_template_config(prompts_config_path()) kb = kb or RagKnowledgeBase.from_template_config( @@ -117,6 +124,8 @@ class TiScenarioRunner: ) router = router or SensitivityRouter.from_template_config(router_config_path()) self.kb = kb + # 报警解释场景配置(issue #74:sop 类目检索 + 检索阈值) + self.alarm_cfg = alarm_config or self._load_alarm_config() if gateway is not None: # 注入 gateway:由调用方负责 prompt_name / 高利害配置 self._gateways = { @@ -141,12 +150,35 @@ class TiScenarioRunner: # -- 场景 1:报警根因解释(高利害) ----------------------------------- + @staticmethod + def _load_alarm_config() -> dict: + """加载报警解释场景配置资产(issue #74)。""" + import yaml + + with open(alarm_config_path(), "r", encoding="utf-8") as fh: + return (yaml.safe_load(fh) or {}).get("alarm_explain", {}) + def explain_alarm(self, alarm: str, confidence: float = 1.0, - top_k: int = 3) -> GatewayResult: - """解释一条报警的可能原因与处置建议(RAG 检索异常处置 SOP)。""" - hits = self.kb.search(alarm, top_k=top_k, - categories=None) - sources = [h.source for h in hits] + top_k: Optional[int] = None) -> GatewayResult: + """解释一条报警的可能原因与处置建议。 + + RAG 接入(issue #74):只检索 **sop 知识域**(异常处置 SOP), + 并按 min_score 过滤——低于阈值视为未命中,降级提示人工确认 + (高利害场景宁缺毋滥,避免无据回答)。 + """ + rag = self.alarm_cfg.get("rag", {}) or {} + cat_names = rag.get("categories") + categories = ([KnowledgeSourceKind(c) for c in cat_names] + if cat_names else None) + top_k = top_k or int(rag.get("top_k", 3)) + min_score = float(rag.get("min_score", 0.0)) + hits = self.kb.search(alarm, top_k=top_k, categories=categories) + hits = [h for h in hits if h.score >= min_score] + if hits: + sources = [h.source for h in hits] + else: + # 未检索到 SOP:降级提示人工确认(来源占位,阻断无据回答) + sources = ["未检索到相关 SOP,请人工确认"] return self._gateways["alarm_explain"].ask( alarm, rag_context=sources, confidence=confidence, ) diff --git a/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py b/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py index f794389..b82ec07 100644 --- a/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py +++ b/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py @@ -105,6 +105,21 @@ class TestScenariosE2E(unittest.TestCase): self.assertIn("来源", result.answer) self.assertEqual(result.route.target, RouteTarget.LOCAL) + def test_explain_alarm_rag_sop_only(self): + """报警解释 RAG 接入(issue #74):只检索 sop 知识域并回显来源。""" + result = self.runner.explain_alarm("炉温超上限报警怎么处理") + # 演示库 sop 类目(异常处置SOP)应被检索到并随答案溯源 + self.assertIn("SOP", result.answer) + + def test_explain_alarm_min_score_fallback(self): + """检索阈值(issue #74):低分视为未命中 → 降级提示人工确认。""" + # 注入高 min_score 配置:正常演示库检索得分低于阈值 → 降级 + runner = TiScenarioRunner(alarm_config={ + "rag": {"categories": ["sop"], "top_k": 3, "min_score": 99.0}, + }) + result = runner.explain_alarm("炉温超上限报警怎么处理") + self.assertIn("人工确认", result.answer) + def test_shift_handover(self): result = self.runner.generate_handover("甲班:生产平稳,炉温正常,无异常事项") self.assertTrue(result.answer)