From e2d24f18393048ce3810a806bc07c8c15e9e25eb Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Wed, 5 Aug 2026 04:19:36 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#75=20[Ti-?= =?UTF-8?q?1]=20=E4=BA=A4=E6=8E=A5=E7=8F=AD=E6=91=98=E8=A6=81=E7=94=9F?= =?UTF-8?q?=E6=88=90=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/handover.template.yaml | 21 +++ templates/ti-cl4/llm-scenarios/handover.py | 166 ++++++++++++++++++ .../llm-scenarios/tests/test_handover.py | 109 ++++++++++++ 3 files changed, 296 insertions(+) create mode 100644 templates/ti-cl4/llm-scenarios/config/handover.template.yaml create mode 100644 templates/ti-cl4/llm-scenarios/handover.py create mode 100644 templates/ti-cl4/llm-scenarios/tests/test_handover.py diff --git a/templates/ti-cl4/llm-scenarios/config/handover.template.yaml b/templates/ti-cl4/llm-scenarios/config/handover.template.yaml new file mode 100644 index 0000000..e888a2f --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/config/handover.template.yaml @@ -0,0 +1,21 @@ +# -*- coding: utf-8 -*- +# 模板「交接班摘要」配置资产：ti-cl4（Template-Ti 一期）。 +# +# 说明（issue #75 / 父 Issue #11）： +# - sections：交接班摘要章节（固定五节，可增删改标题）； +# - alarm_requires_disposal：异常事件必须含处置记录（缺失 → 待跟进标记， +# 对齐交接班规范「异常事项须标注发生时间与处理人」）； +# - default_shift：默认班次； +# - 换行业/调章节只改本文件，生成器内核零改动。 +template: ti-cl4 +version: 1.0.0 + +handover: + default_shift: 甲班 + alarm_requires_disposal: true + sections: + - {key: production, title: 生产概况} + - {key: equipment, title: 设备运行状态} + - {key: alarm, title: 异常与处置} + - {key: safety, title: 安全注意事项} + - {key: todo, title: 待办事项} diff --git a/templates/ti-cl4/llm-scenarios/handover.py b/templates/ti-cl4/llm-scenarios/handover.py new file mode 100644 index 0000000..44d29cb --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/handover.py @@ -0,0 +1,166 @@ +# -*- coding: utf-8 -*- +"""Template-Ti 一期 · 交接班摘要生成功能 —— issue #75。 + +父 Issue #11「④ LLM 报警解释 / 交接班 / NL 查询」子任务： +把班次事件记录生成为**结构化交接班摘要**（本地规则，可离线测试）： + +- 事件归类：production（生产概况）/ equipment（设备运行）/ alarm（异常与处置）/ + safety（安全注意事项）/ todo（待办事项）； +- 异常处置校验：alarm 事件必须含处置记录，缺失 → `needs_followup=True` + （对齐交接班规范：异常事项须标注发生时间与处理人，PRD 5.4）； +- 章节输出：固定五节 Markdown 摘要（to_text），供交接班对话/报表使用； +- 与 `TiScenarioRunner.generate_handover`（LLM + RAG 润色）互补： + 本模块产出结构化底稿，LLM 层可再润色/补充。 + +纯本地规则实现（无 LLM 依赖）；事件来源：DCS 告警、人工录入、巡检记录。 +""" +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from datetime import date as _date +from typing import Dict, List, Optional + +#: 默认配置资产路径（相对本模块） +DEFAULT_CONFIG_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "config", + "handover.template.yaml") + +#: 默认章节定义（key → 标题） +DEFAULT_SECTIONS: List[Dict[str, str]] = [ + {"key": "production", "title": "生产概况"}, + {"key": "equipment", "title": "设备运行状态"}, + {"key": "alarm", "title": "异常与处置"}, + {"key": "safety", "title": "安全注意事项"}, + {"key": "todo", "title": "待办事项"}, +] + + +@dataclass +class HandoverReport: + """结构化交接班摘要。""" + + shift: str + date: str + sections: Dict[str, List[str]] = field(default_factory=dict) + needs_followup: bool = False + followup_items: List[str] = field(default_factory=list) + + def to_text(self, section_titles: Optional[Dict[str, str]] = None) -> str: + """输出 Markdown 交接班摘要文本。""" + titles = section_titles or {s["key"]: s["title"] + for s in DEFAULT_SECTIONS} + lines = [f"# 交接班摘要（{self.date} {self.shift}）"] + for key, title in titles.items(): + items = self.sections.get(key, []) + lines.append(f"\n## {title}") + lines.append("\n".join(f"- {it}" for it in items) + if items else "- （无记录）") + if self.needs_followup: + lines.append("\n> ⚠ 待跟进：异常事项缺少处置记录，请人工确认。") + return "\n".join(lines) + + def to_dict(self) -> dict: + return { + "shift": self.shift, "date": self.date, + "sections": self.sections, + "needs_followup": self.needs_followup, + "followup_items": self.followup_items, + } + + +class HandoverGenerator: + """交接班摘要生成器（事件 → 结构化报告，本地规则）。""" + + #: 事件类型 → 章节 key（未知类型归入待办） + _TYPE_TO_SECTION = { + "production": "production", "设备": "production", + "equipment": "equipment", "设备运行": "equipment", + "alarm": "alarm", "异常": "alarm", "报警": "alarm", + "safety": "safety", "安全": "safety", + "todo": "todo", "待办": "todo", + } + + def __init__( + self, + sections: Optional[List[Dict[str, str]]] = None, + alarm_requires_disposal: bool = True, + default_shift: str = "甲班", + ) -> None: + self.sections = sections or DEFAULT_SECTIONS + self.alarm_requires_disposal = alarm_requires_disposal + self.default_shift = default_shift + + # ------------------------------------------------------------------ + @classmethod + def from_template_config(cls, path: str = DEFAULT_CONFIG_PATH) -> "HandoverGenerator": + """从模板配置资产加载（config/handover.template.yaml）。""" + import yaml + + with open(path, "r", encoding="utf-8") as fh: + raw = yaml.safe_load(fh) or {} + cfg = raw.get("handover", {}) or {} + return cls( + sections=cfg.get("sections") or DEFAULT_SECTIONS, + alarm_requires_disposal=bool( + cfg.get("alarm_requires_disposal", True)), + default_shift=cfg.get("default_shift", "甲班"), + ) + + # ------------------------------------------------------------------ + def generate( + self, + events: List[dict], + shift: Optional[str] = None, + date: Optional[str] = None, + ) -> HandoverReport: + """按事件记录生成交接班摘要。 + + Args: + events: `[{type, text, time?, operator?}]`；type 见 `_TYPE_TO_SECTION`； + shift: 班次（缺省 = 配置默认）； + date: 日期（缺省 = 今天）。 + """ + sections: Dict[str, List[str]] = {s["key"]: [] + for s in self.sections} + report = HandoverReport( + shift=shift or self.default_shift, + date=date or _date.today().isoformat(), + sections=sections, + ) + for ev in events: + section = self._section_for(ev.get("type", "")) + text = self._format_event(ev) + sections.setdefault(section, []).append(text) + if section == "alarm": + self._check_disposal(ev, report) + if not sections.get("alarm"): + sections["alarm"] = ["本班无异常事项"] + return report + + # ------------------------------------------------------------------ + def _section_for(self, event_type: str) -> str: + return self._TYPE_TO_SECTION.get(event_type.strip(), "todo") + + @staticmethod + def _format_event(ev: dict) -> str: + time_ = ev.get("time") + operator = ev.get("operator") + text = str(ev.get("text", "")).strip() + parts = [] + if time_: + parts.append(f"[{time_}]") + parts.append(text or "（无描述）") + if operator: + parts.append(f"（处理人: {operator}）") + return " ".join(parts) + + def _check_disposal(self, ev: dict, report: HandoverReport) -> None: + """异常事件须含处置记录（disposal 或 operator），否则标记待跟进。""" + if not self.alarm_requires_disposal: + return + disposal = ev.get("disposal") or ev.get("operator") + if not disposal: + report.needs_followup = True + report.followup_items.append( + f"异常事件缺处置记录: {ev.get('text', '')[:40]}") diff --git a/templates/ti-cl4/llm-scenarios/tests/test_handover.py b/templates/ti-cl4/llm-scenarios/tests/test_handover.py new file mode 100644 index 0000000..1807f08 --- /dev/null +++ b/templates/ti-cl4/llm-scenarios/tests/test_handover.py @@ -0,0 +1,109 @@ +# -*- coding: utf-8 -*- +"""交接班摘要生成器测试（issue #75）。 + +覆盖： +1. 配置资产加载（章节/异常处置要求/默认班次）； +2. 事件归类（production/equipment/alarm/safety/todo + 未知 → 待办）； +3. 异常处置校验：缺失处置 → needs_followup + followup_items； +4. 无异常事件 → 「本班无异常事项」占位； +5. to_text Markdown 输出（标题/章节/待跟进标注）。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from ti_scenarios.handover import ( # noqa: E402 + DEFAULT_SECTIONS, + HandoverGenerator, +) + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "handover.template.yaml", +) + + +class TestConfigLoad(unittest.TestCase): + """配置资产加载。""" + + def setUp(self): + self.g = HandoverGenerator.from_template_config(CONFIG) + + def test_sections_and_defaults(self): + self.assertEqual([s["key"] for s in self.g.sections], + [s["key"] for s in DEFAULT_SECTIONS]) + self.assertTrue(self.g.alarm_requires_disposal) + self.assertEqual(self.g.default_shift, "甲班") + + +class TestGenerate(unittest.TestCase): + """事件归类与摘要生成。""" + + def setUp(self): + self.g = HandoverGenerator.from_template_config(CONFIG) + + def test_event_classification(self): + events = [ + {"type": "production", "text": "生产平稳，产出 12 批", "time": "08:00"}, + {"type": "equipment", "text": "洗涤泵 A 运行正常"}, + {"type": "safety", "text": "巡检未发现隐患"}, + {"type": "todo", "text": "明日更换滤芯"}, + {"type": "未知类型", "text": "临时事项"}, + ] + report = self.g.generate(events, shift="甲班", date="2026-09-01") + self.assertEqual(len(report.sections["production"]), 1) + self.assertEqual(len(report.sections["equipment"]), 1) + self.assertEqual(len(report.sections["safety"]), 1) + self.assertEqual(len(report.sections["todo"]), 2) # 已知 + 未知类型 + self.assertFalse(report.needs_followup) + + def test_alarm_with_disposal_ok(self): + events = [{"type": "alarm", "text": "炉温超上限", + "time": "09:12", "operator": "李工", + "disposal": "已降低氯气流量，10 分钟内回落"}] + report = self.g.generate(events, shift="乙班") + self.assertEqual(len(report.sections["alarm"]), 1) + self.assertFalse(report.needs_followup) + + def test_alarm_missing_disposal_marks_followup(self): + events = [{"type": "alarm", "text": "炉温超上限", "time": "09:12"}] + report = self.g.generate(events) + self.assertTrue(report.needs_followup) + self.assertTrue(any("炉温超上限" in it for it in report.followup_items)) + + def test_no_alarm_placeholder(self): + report = self.g.generate([{"type": "production", "text": "正常"}]) + self.assertEqual(report.sections["alarm"], ["本班无异常事项"]) + self.assertFalse(report.needs_followup) + + def test_default_shift(self): + report = self.g.generate([]) + self.assertEqual(report.shift, "甲班") + + +class TestOutput(unittest.TestCase): + """Markdown 输出。""" + + def test_to_text_sections(self): + g = HandoverGenerator.from_template_config(CONFIG) + report = g.generate([{"type": "alarm", "text": "炉温报警"}], + shift="甲班", date="2026-09-01") + text = report.to_text() + self.assertIn("# 交接班摘要（2026-09-01 甲班）", text) + self.assertIn("## 生产概况", text) + self.assertIn("## 异常与处置", text) + self.assertIn("待跟进", text) # 缺处置记录 → 待跟进标注 + self.assertIn("- 炉温报警", text) + + def test_to_dict(self): + report = HandoverGenerator().generate([]) + d = report.to_dict() + self.assertEqual(set(d), {"shift", "date", "sections", + "needs_followup", "followup_items"}) + + +if __name__ == "__main__": + unittest.main()