feat: 完成 issue #74 [Ti-1] 报警解释 Prompt 模板与 RAG 接入

This commit is contained in:
2026-08-05 03:39:03 +08:00
parent f6cdc84860
commit fc4ac42007
3 changed files with 75 additions and 6 deletions
@@ -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 # 低信度转人工
+37 -5
View File
@@ -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)
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,
)
@@ -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)