From 8b8fa5402427dfac7aca3fcae4f1ce832ec38b72 Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Tue, 4 Aug 2026 22:12:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#11=20[Tem?= =?UTF-8?q?plate-Ti=20=E4=B8=80=E6=9C=9F]=20=E2=91=A3=20LLM=20=E6=8A=A5?= =?UTF-8?q?=E8=AD=A6=E8=A7=A3=E9=87=8A=20/=20=E4=BA=A4=E6=8E=A5=E7=8F=AD?= =?UTF-8?q?=20/=20NL=20=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- templates/ti-cl4/llm-scenarios/README.md | 57 ++++++ templates/ti-cl4/llm-scenarios/__init__.py | 15 ++ .../config/scenarios.template.yaml | 33 +++ templates/ti-cl4/llm-scenarios/scenarios.py | 192 ++++++++++++++++++ .../ti-cl4/llm-scenarios/tests/_bootstrap.py | 34 ++++ .../llm-scenarios/tests/test_scenarios.py | 134 ++++++++++++ 6 files changed, 465 insertions(+) create mode 100644 templates/ti-cl4/llm-scenarios/README.md create mode 100644 templates/ti-cl4/llm-scenarios/__init__.py create mode 100644 templates/ti-cl4/llm-scenarios/config/scenarios.template.yaml create mode 100644 templates/ti-cl4/llm-scenarios/scenarios.py create mode 100644 templates/ti-cl4/llm-scenarios/tests/_bootstrap.py create mode 100644 templates/ti-cl4/llm-scenarios/tests/test_scenarios.py diff --git a/templates/ti-cl4/llm-scenarios/README.md b/templates/ti-cl4/llm-scenarios/README.md new file mode 100644 index 0000000..195f747 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/README.md @@ -0,0 +1,57 @@ +# iAOP-Template-Ti 一期 · LLM 场景层（LLM Scenarios） + +对应 PRD 5.4 / 5.5 与 EPIC #11「[Template-Ti 一期] ④ LLM 报警解释 / 交接班 / NL 查询」： +复用 ④ LLM 网关（core/llm-gateway）与领域 RAG（core/rag-kb），针对 Ti 场景 +配置 RAG 知识域与提示词模板，落地三个业务场景： +- **报警根因解释**（`alarm_explain`）：高利害，启用幻觉校验信度阈值，低信度转人工； +- **交接班自动摘要**（`shift_handover`）：生产概况 / 异常事项 / 安全注意事项； +- **自然语言查询驾驶舱**（`nl_query`）：NL → 指标 / 查询意图。 + +验收（Issue #11）：路由准确率 ≥ 96.5%、幻觉校验通过。 + +## 模块结构 + +``` +templates/ti-cl4/llm-scenarios/ +├── __init__.py 场景包入口（导出 TiScenarioRunner） +├── scenarios.py Ti 场景编排（三场景 + 演示知识文档集 + 组件装配） +├── config/ +│ └── scenarios.template.yaml 场景配置资产（提示词绑定 / RAG 知识域 / 验收指标） +└── tests/ + ├── _bootstrap.py 测试引导（挂载 ti_scenarios / llm_gateway / rag_kb） + └── test_scenarios.py 场景端到端 + 路由准确率 + 幻觉校验测试 +``` + +## 用法 + +```python +from ti_scenarios import TiScenarioRunner + +runner = TiScenarioRunner() + +# 1) 报警根因解释（高利害；低信度自动转人工） +r1 = runner.explain_alarm("炉温超上限报警怎么处理", confidence=0.95) +print(r1.answer) # 含 SOP 引用溯源 + +# 2) 交接班自动摘要 +r2 = runner.generate_handover("甲班：生产平稳，炉温正常，无异常事项") + +# 3) 自然语言查询驾驶舱 +r3 = runner.query_cockpit("查询最近一小时的氯气流量趋势") +``` + +## 设计说明 + +- **内核零改动**：场景层只复用 `LLMGateway` / `PromptRegistry` / `SensitivityRouter` / + `RagKnowledgeBase`，换行业只改模板资产（prompts / router / kb / scenarios）； +- **RAG 演示数据**：`DEMO_KB_DOCS` 内置工艺规范 / 异常处置 SOP / 交接班规范等 + 演示文档；接入真实客户数据时替换 `kb_loader`（`TiScenarioRunner(kb_loader=...)`）； +- **验收自检**：`tests/test_scenarios.py::TestRouteAccuracy` 用 Ti 场景查询集 + 断言路由准确率 ≥ 96.5%；`TestScenariosE2E` 覆盖三场景端到端与幻觉校验。 + +## 运行测试 + +```bash +cd templates/ti-cl4/llm-scenarios/tests +python -m unittest discover -s . -p "test_*.py" +``` diff --git a/templates/ti-cl4/llm-scenarios/__init__.py b/templates/ti-cl4/llm-scenarios/__init__.py new file mode 100644 index 0000000..2376056 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/__init__.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +"""iAOP-Template-Ti 一期 · LLM 场景层（LLM Scenarios）。 + +对应 PRD 5.4 / 5.5 与 EPIC #11「[Template-Ti 一期] ④ LLM 报警解释 / 交接班 / NL 查询」： +复用 ④ LLM 网关（core/llm-gateway）+ 领域 RAG（core/rag-kb），针对 Ti 场景配置 +RAG 知识域与提示词模板，落地三个业务场景： +- 报警根因解释（alarm_explain，高利害，启用信度阈值）； +- 交接班自动摘要（shift_handover）； +- 自然语言查询驾驶舱（nl_query，NL → 查询意图/指标）。 + +验收（Issue #11）：路由准确率 ≥ 96.5%、幻觉校验通过。 +""" +from .scenarios import TiScenarioRunner + +__all__ = ["TiScenarioRunner"] diff --git a/templates/ti-cl4/llm-scenarios/config/scenarios.template.yaml b/templates/ti-cl4/llm-scenarios/config/scenarios.template.yaml new file mode 100644 index 0000000..ce9b82b --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/config/scenarios.template.yaml @@ -0,0 +1,33 @@ +# -*- coding: utf-8 -*- +# 模板「Ti 场景配置」资产示例：ti-cl4（氯化车间/海绵钛，Template-Ti 一期）。 +# +# 说明： +# - 这是「LLM 业务场景」配置点（EPIC #11）：报警解释 / 交接班摘要 / NL 查询； +# - prompt_template 绑定 llm-gateway 提示词版本库中的模板名（prompts.template.yaml）； +# - rag_categories 限定该场景 RAG 检索的知识域（process/sop/standard，见 kb.template.yaml）； +# - high_stakes: true 的场景启用幻觉校验信度阈值（低信度转人工）； +# - acceptance.route_accuracy 为场景验收指标（EPIC #11：路由准确率 ≥ 96.5%）。 +template: ti-cl4 +version: 1.0.0 + +scenarios: + - name: alarm_explain + description: 报警根因解释（高利害，低信度转人工确认） + prompt_template: alarm_explain + high_stakes: true + rag_categories: [sop] + acceptance: + route_accuracy: 0.965 + hallucination_check: true + + - name: shift_handover + description: 交接班自动摘要（生产概况/异常事项/安全注意事项） + prompt_template: shift_handover + high_stakes: false + rag_categories: [sop] + + - name: nl_query + description: 自然语言查询驾驶舱（NL → 指标/查询意图） + prompt_template: qa + high_stakes: false + rag_categories: [process] diff --git a/templates/ti-cl4/llm-scenarios/scenarios.py b/templates/ti-cl4/llm-scenarios/scenarios.py new file mode 100644 index 0000000..48af168 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/scenarios.py @@ -0,0 +1,192 @@ +# -*- coding: utf-8 -*- +"""Ti 场景编排：报警解释 / 交接班摘要 / NL 查询驾驶舱（EPIC #11）。 + +复用内核组件（零内核改动，换行业只改模板资产）： +- ``llm_gateway.LLMGateway``：混合网关主编排（敏感度路由 → 生成 → 幻觉校验 → DLP）； +- ``llm_gateway.prompts.PromptRegistry``：提示词版本库（alarm_explain / shift_handover / qa）； +- ``llm_gateway.router.SensitivityRouter``：敏感度路由（行业规则 + 内置保底）； +- ``rag_kb.RagKnowledgeBase``：领域 RAG（工艺规范 / SOP / 国标三类知识源）。 + +三个场景均通过 ``TiScenarioRunner`` 暴露，业务方只需一行调用： + runner = TiScenarioRunner() + result = runner.explain_alarm("氯化炉炉温异常") +""" +from __future__ import annotations + +import os +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 + +# --------------------------------------------------------------------------- +# 场景配置资产路径 +# --------------------------------------------------------------------------- + + +def _scenarios_dir() -> str: + return os.path.dirname(os.path.abspath(__file__)) + + +def _repo_root() -> str: + """仓库根目录（scenarios.py 向上 4 层：llm-scenarios → ti-cl4 → templates → 根）。""" + return os.path.dirname(os.path.dirname(os.path.dirname( + os.path.dirname(os.path.abspath(__file__))))) + + +def scenarios_config_path() -> str: + """场景配置资产（scenarios.template.yaml）路径。""" + return os.path.join(_scenarios_dir(), "config", "scenarios.template.yaml") + + +def prompts_config_path() -> str: + """复用 llm-gateway 提示词版本库资产路径。""" + return os.path.join(_repo_root(), "core", "llm-gateway", "config", + "prompts.template.yaml") + + +def kb_config_path() -> str: + """复用 rag-kb 知识库模板资产路径。""" + return os.path.join(_repo_root(), "core", "rag-kb", "config", + "kb.template.yaml") + + +# --------------------------------------------------------------------------- +# 演示知识文档集（Ti 场景开箱即用；接入真实客户数据时替换 loader 即可） +# --------------------------------------------------------------------------- + +DEMO_KB_DOCS: Dict[str, str] = { + "沸腾氯化工艺规范": ( + "沸腾氯化炉采用流态化氯化工艺，炉温控制在 850±50℃，" + "氯气流量按加料量比例调节，加料比保持 1:2.4~1:2.8。" + "炉温异常时优先检查氯气流量与加料系统。" + ), + "沸腾氯化炉操作手册": ( + "开机前确认氯气缓冲罐压力、炉体密封与尾气处理系统正常；" + "运行中每 30 分钟记录一次炉温、氯气流量与出料量。" + ), + "沸腾氯化炉异常处置SOP": ( + "SOP-CL-001：炉温超上限（>900℃）时立即降低氯气流量并减少加料，" + "若 10 分钟内未回落则按紧急停机流程处理，并通知当班班长。" + ), + "交接班报告生成规范": ( + "交接班报告须包含：当班生产概况、设备运行状态、异常与处置记录、" + "安全注意事项、待办事项。异常事项必须标注发生时间与处理人。" + ), + "GB/T 氯气安全使用标准": ( + "氯气属于剧毒气体，作业场所应配备气体泄漏检测与报警装置，" + "作业人员须佩戴防护用品，泄漏时启动应急程序并疏散无关人员。" + ), + "GB/T 钛及钛合金加工标准": ( + "海绵钛产品纯度按 GB/T 标准分级，钛纯度 ≥99.5% 为一级品；" + "氯化产物杂质含量影响最终钛纯度，须按批次检验并留样。" + ), +} + + +def demo_kb_loader(title: str) -> str: + """演示文档加载器：按标题返回内置演示文本（换数据源时替换本函数）。""" + return DEMO_KB_DOCS.get(title, "") + + +# --------------------------------------------------------------------------- +# Ti 场景编排 +# --------------------------------------------------------------------------- + + +class TiScenarioRunner: + """Template-Ti 一期 LLM 场景统一入口（EPIC #11 主体交付）。 + + 构造参数均可注入（便于测试与替换真实组件）；缺省使用仓库模板资产 + （提示词版本库 + 领域 RAG + 行业路由规则）构建完整场景编排。 + """ + + def __init__( + self, + gateway: Optional[LLMGateway] = None, + kb: Optional[RagKnowledgeBase] = None, + prompts: Optional[PromptRegistry] = None, + router: Optional[SensitivityRouter] = None, + kb_loader: Callable[[str], str] = demo_kb_loader, + ) -> None: + prompts = prompts or PromptRegistry.from_template_config(prompts_config_path()) + kb = kb or RagKnowledgeBase.from_template_config( + load_kb_config(kb_config_path()), loader=kb_loader, + ) + router = router or SensitivityRouter.from_template_config(router_config_path()) + self.kb = kb + if gateway is not None: + # 注入 gateway：由调用方负责 prompt_name / 高利害配置 + self._gateways = { + "alarm_explain": gateway, + "shift_handover": gateway, + "nl_query": gateway, + } + else: + # 每个场景绑定各自的提示词模板版本（可复现） + base = dict(prompts=prompts, router=router) + self._gateways = { + # 报警解释：高利害，命中即启用幻觉校验信度阈值 → 低信度转人工 + "alarm_explain": LLMGateway( + prompt_name="alarm_explain", + high_stakes_names={"alarm_explain"}, **base, + ), + "shift_handover": LLMGateway( + prompt_name="shift_handover", **base, + ), + "nl_query": LLMGateway(prompt_name="qa", **base), + } + + # -- 场景 1：报警根因解释（高利害） ----------------------------------- + + 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] + return self._gateways["alarm_explain"].ask( + alarm, rag_context=sources, confidence=confidence, + ) + + # -- 场景 2：交接班自动摘要 ------------------------------------------- + + def generate_handover(self, shift_desc: str, confidence: float = 1.0, + top_k: int = 3) -> GatewayResult: + """按班次情况生成交接班摘要（RAG 检索交接班规范）。""" + hits = self.kb.search(shift_desc, top_k=top_k, categories=None) + sources = [h.source for h in hits] + return self._gateways["shift_handover"].ask( + shift_desc, rag_context=sources, confidence=confidence, + ) + + # -- 场景 3：自然语言查询驾驶舱（NL → 查询意图/指标） ----------------- + + def query_cockpit(self, question: str, confidence: float = 1.0, + top_k: int = 3) -> GatewayResult: + """把自然语言问题映射为驾驶舱查询意图（RAG 检索工艺指标定义）。""" + hits = self.kb.search(question, top_k=top_k, categories=None) + sources = [h.source for h in hits] + return self._gateways["nl_query"].ask( + question, rag_context=sources, confidence=confidence, + ) + + # -- 审计 -------------------------------------------------------------- + + def drain_audits(self) -> Dict[str, List[Dict[str, object]]]: + """取走网关各组件审计记录（DLP/路由/Prompt/幻觉校验）。""" + audits: Dict[str, List[Dict[str, object]]] = { + "dlp": [], "router": [], "prompts": [], "guard": [], + } + for gw in set(self._gateways.values()): + for key, rows in gw.drain_audits().items(): + audits.setdefault(key, []).extend(rows) + return audits + + +def router_config_path() -> str: + """复用 llm-gateway 行业路由规则资产路径。""" + return os.path.join(_repo_root(), "core", "llm-gateway", "config", + "router.template.yaml") diff --git a/templates/ti-cl4/llm-scenarios/tests/_bootstrap.py b/templates/ti-cl4/llm-scenarios/tests/_bootstrap.py new file mode 100644 index 0000000..6e65b98 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/tests/_bootstrap.py @@ -0,0 +1,34 @@ +# -*- coding: utf-8 -*- +"""测试引导：加载连字符目录为可导入包，使场景模块可复用内核组件。 + +- `templates/ti-cl4/llm-scenarios` → 包名 ``ti_scenarios``； +- `core/llm-gateway` → 包名 ``llm_gateway``（场景依赖）； +- `core/rag-kb` → 包名 ``rag_kb``（场景依赖）。 + +与仓库内各 core 模块的测试引导同款模式；这里用 importlib 完整加载包 +（执行 __init__.py），保持 ``from rag_kb import ...`` 顶层导出可用。 +""" +import importlib.util +import os +import sys + +SCEN_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) + + +def _load_package(name: str, path: str) -> None: + """按文件路径完整加载一个包（执行其 __init__.py）。""" + if name in sys.modules: + return + init_py = os.path.join(path, "__init__.py") + spec = importlib.util.spec_from_file_location( + name, init_py, submodule_search_locations=[path]) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + + +_load_package("rag_kb", os.path.join(REPO_ROOT, "core", "rag-kb")) +_load_package("llm_gateway", os.path.join(REPO_ROOT, "core", "llm-gateway")) +_load_package("ti_scenarios", SCEN_DIR) diff --git a/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py b/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py new file mode 100644 index 0000000..f794389 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/tests/test_scenarios.py @@ -0,0 +1,134 @@ +# -*- 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()