feat(#53): 移动端交接班摘要生成(NL,PRD 场景C ⑤ 移动端交接班摘要)

对应 PRD 场景C(辅助):交接班 → LLM 汇总本班关键事件/能耗/待办 → 生成交接班
报告 → 推送下一班(line 84/350-351)。把本班原始数据 → 移动端交接班摘要这条
链路模板化、可配置、可测试。

新增 core/shift-handover 内核模块:
- handover.py:班次数据归一化(ShiftRecord) + 摘要配置(HandoverBriefConfig) +
  validate/load 校验对 + build_llm_input(填 shift_handover v1.0.1 提示词占位符) +
  generate_handover_brief(LLM 注入生成, 离线/故障自动降级为确定性摘要) +
  render_handover_brief(编译移动端只读卡片 props)
- 零运行时依赖(仅标准库); LLM 以依赖注入传入, 内核不绑定云端 SDK
- 离线/LLM故障降级保证可用性≥99.8%(PRD 模型服务故障自动降级)
- 含 P0/安全事件时强制 requireConfirm=true(PRD 高利害人工确认)

新增资产/测试/验收:
- templates/ti-cl4/dashboard/handover_brief.ti.yaml(Ti 模板配置资产)
- tests/test_handover.py(32 用例全通过) + tests/_bootstrap.py(连字符目录挂载)
- scripts/verify_handover_brief.py(4 能力点全通过: 配置合法/LLM+降级/
  配置点驱动展示/≤2分钟验收线)
- README.md

验证: python -m unittest discover -s tests (32 OK) +
      python scripts/verify_handover_brief.py (全部通过)
This commit is contained in:
2026-08-05 03:38:07 +08:00
parent f6cdc84860
commit 5bc7ff957e
7 changed files with 1494 additions and 0 deletions
+79
View File
@@ -0,0 +1,79 @@
# 移动端交接班摘要模块(iAOP-Core / shift-handover)
对齐 PRD **场景C(辅助)** 与 **5.4「④ LLM 网关」**:
> 交接班 → LLM 汇总本班关键事件/能耗/待办 → 生成交接班报告 → 推送下一班。
以及 PRD 验收口径:交接班报告生成 ≤ 2 分钟(line 350)、移动端交接班摘要可用(line 284/351)。
## 模块状态
| 能力 | Issue | 状态 |
| --- | --- | --- |
| **移动端交接班摘要生成(NL)** | **#53** | **本分支 feature/issue-53** |
`#53` 把"本班原始数据 → 移动端交接班摘要"这条链路**模板化、可配置、可测试**:
班次数据归一化(`ShiftRecord`)→ 按 `HandoverBriefConfig` 经 LLM(或离线确定性降级)
生成摘要 → 编译成移动端只读卡片可直接消费的 props。
## 设计要点
- **零运行时依赖**:与 cockpit / data-bus / llm-gateway 一致,只用标准库;LLM 调用以
**依赖注入**形式传入(`llm_generate: Callable[[str], str]`),内核不绑定云端 SDK,
便于本地闭环与离线降级。
- **离线/故障自动降级**:未注入 LLM 或调用失败时回落到 `render_deterministic_brief`
(PRD「模型服务故障自动降级」「可用性 ≥ 99.8%」),保证交接班链路不中断。
- **声明式 + 强校验**:`validate_handover_config` 收集全部字段级错误(沿用 #50/#52
风格),`load_handover_config` 校验失败抛 `HandoverConfigError` 并携带错误清单。
- **与未合并分支解耦**:不 import `llm-gateway` 的 `PromptRegistry`;仅按约定的
`prompt_template` 名字产出 LLM 输入文本,待 llm-gateway 合入后由调用方注入真正的
LLM 生成函数。
## 配置点(外置为模板资产,切换模板零改码)
| 配置点 | 字段 | 说明 |
| --- | --- | --- |
| 提示词模板 | `promptTemplate` / `promptVersion` | 绑定 `shift_handover` v1.0.1(current) |
| 章节 | `sections` | overview/abnormal/safety/energy/todos |
| 展示策略 | `maxEvents` / `maxTodos` / `collapseThreshold` | 移动端折叠与截断 |
| 字号 | `fontSize` | sm/md/lg |
| 高利害确认 | `requireConfirm` | 默认 false;含 P0/安全事件时内核强制 true |
## 用法
```python
import sys, types, os
# 含连字符目录挂载为 shift_handover 包(同 core/data-bus 的做法)
_d = os.path.join("core", "shift-handover")
if "shift_handover" not in sys.modules:
_p = types.ModuleType("shift_handover"); _p.__path__ = [_d]; sys.modules["shift_handover"] = _p
from shift_handover.handover import (
normalize_shift_record, render_handover_brief, generate_handover_brief,
)
# 1) 班次数据(来自驾驶舱/总线/告警面板聚合)
record = normalize_shift_record({
"shift": "夜班 2026-08-05 00:00~08:00", "operator": "张工",
"overview": "TiCl4 产量 36.2t。",
"events": [{"time": "01:20", "title": "炉层温度越上限", "severity": "P1"}],
"todos": [{"title": "白班复测 3 层温度", "priority": "high"}],
"safety": "注意 3 层高温区巡检。",
})
# 2) 生成摘要(llm_generate=None 时离线降级;注入则用 LLM 输出)
summary = generate_handover_brief(record, llm_generate=my_llm_fn)
# 3) 编译移动端只读卡片 props
props = render_handover_brief(record, summary=summary)
```
## 测试与验收
```bash
# 单元测试(在 core/shift-handover 目录下)
python -m unittest discover -s tests -v # 32 用例全通过
# 验收脚本(4 能力点:配置合法 / LLM+降级 / 配置点驱动展示 / ≤2 分钟)
python scripts/verify_handover_brief.py
```
+75
View File
@@ -0,0 +1,75 @@
# -*- coding: utf-8 -*-
"""iAOP-Core · 交接班摘要模块(shift-handover)—— 对齐 PRD 场景C / 5.4「④ LLM 网关」。
对外暴露 issue #53「移动端交接班摘要生成(NL)」的能力:把本班关键事件/能耗/待办/
异常归一为 ``ShiftRecord``,按模板配置(``HandoverBriefConfig``)经 LLM(或离线
确定性降级)生成交接班摘要,并编译成移动端只读卡片可直接消费的 props。
LLM 以依赖注入形式传入(``llm_generate``),内核不绑定具体云端 SDK;未注入或调用
失败时自动降级为 ``render_deterministic_brief``(PRD「模型服务故障自动降级」、
「可用性 ≥ 99.8%」),保证离线/降级时交接班链路不中断。
测试:`python -m unittest discover -s tests -v`(在 core/shift-handover 目录下)。
验收:`python scripts/verify_handover_brief.py`。
"""
from __future__ import annotations
from .handover import ( # noqa: F401
DEFAULT_COLLAPSE_THRESHOLD,
DEFAULT_MAX_EVENTS,
DEFAULT_MAX_TODOS,
DEFAULT_PROMPT_TEMPLATE,
DEFAULT_PROMPT_VERSION,
DEFAULT_REQUIRE_CONFIRM,
HANDOVER_CONFIG_SCHEMA_ID,
HANDOVER_CONFIG_SCHEMA_VERSION,
HANDOVER_GENERATION_BUDGET_MS,
MAX_ALLOWED_EVENTS,
MAX_ALLOWED_TODOS,
VALID_FONT_SIZES,
VALID_SECTIONS,
HandoverBriefConfig,
HandoverConfigError,
HandoverConfigValidationResult,
ShiftEvent,
ShiftRecord,
ShiftTodo,
build_llm_input,
generate_handover_brief,
load_handover_config,
normalize_shift_record,
render_deterministic_brief,
render_handover_brief,
validate_handover_config,
within_generation_budget,
)
__all__ = [
"DEFAULT_COLLAPSE_THRESHOLD",
"DEFAULT_MAX_EVENTS",
"DEFAULT_MAX_TODOS",
"DEFAULT_PROMPT_TEMPLATE",
"DEFAULT_PROMPT_VERSION",
"DEFAULT_REQUIRE_CONFIRM",
"HANDOVER_CONFIG_SCHEMA_ID",
"HANDOVER_CONFIG_SCHEMA_VERSION",
"HANDOVER_GENERATION_BUDGET_MS",
"MAX_ALLOWED_EVENTS",
"MAX_ALLOWED_TODOS",
"VALID_FONT_SIZES",
"VALID_SECTIONS",
"HandoverBriefConfig",
"HandoverConfigError",
"HandoverConfigValidationResult",
"ShiftEvent",
"ShiftRecord",
"ShiftTodo",
"build_llm_input",
"generate_handover_brief",
"load_handover_config",
"normalize_shift_record",
"render_deterministic_brief",
"render_handover_brief",
"validate_handover_config",
"within_generation_budget",
]
+685
View File
@@ -0,0 +1,685 @@
# -*- coding: utf-8 -*-
"""移动端交接班摘要生成(issue #53 / PRD 5.4「④ LLM 网关」+ 场景C「⑤ 配置化驾驶舱」)。
对应 PRD **场景C(辅助)**:
交接班 → LLM 汇总本班关键事件/能耗/待办 → 生成交接班报告 → 推送下一班。
以及 PRD 验收口径:
- 「LLM 报警解释/交接班报告/ NL查询可用 …… **交接班报告生成 ≤ 2 分钟**」;
- 「驾驶舱按 Ti 布局渲染,**移动端交接班摘要可用**」(PRD line 284/351)。
本模块把"本班原始数据 → 移动端交接班摘要"这条链路**模板化、可配置、可测试**:
1. **班次数据归一化**(``ShiftRecord`` / ``normalize_shift_record``):把本班
关键事件 / 能耗 / 待办 / 异常告警等异构输入归一为一份内存模型;
2. **移动端摘要配置**(``HandoverBriefConfig``):摘要包含哪些章节、移动端
展示策略(折叠/字号/最多条数/是否高利害人工确认)、绑定的 LLM 提示词模板名
(``shift_handover`` v1.0.1,见 ``core/llm-gateway/config/prompts.template.yaml``);
3. **LLM 出参渲染**(``build_llm_input``):把班次数据填入 ``shift_handover``
提示词的占位符(``{query}``),产出给 LLM 网关的输入;
4. **摘要生成**(``generate_handover_brief``):调用传入的 LLM 生成函数;当
未配置 LLM(dry-run / 离线 / 单测)时回落到**确定性模板渲染**,保证:
- 离线/降级时仍有可用摘要(PRD「可用性 ≥ 99.8%」「模型服务故障自动降级」);
- 单测无需真实 LLM 即可断言结构与耗时(验收线 ≤ 2 分钟);
5. **移动端负载编译**(``render_handover_brief``):把摘要 + 班次要点编译成
移动端只读卡片可直接消费的 props dict(camelCase、扁平、可 ``json.dumps``)。
设计要点
--------
- **零运行时依赖**:与 cockpit / data-bus / llm-gateway 一致,只用标准库;
LLM 调用以**依赖注入**形式传入(``llm_generate: Callable[[str], str]``),
不在内核里硬编码云端 SDK,便于本地闭环与离线降级。
- **声明式 + 强校验**:``validate_handover_config`` 收集全部字段级错误
(沿用 #50 / #52 的多错误聚合风格),``load_handover_config`` 校验失败抛
``HandoverConfigError`` 并携带错误清单。
- **与未合并分支解耦**:不 import ``llm-gateway`` 的 ``PromptRegistry``(其
feature 分支尚未合入 main);仅按约定的 ``prompt_template`` 名字产出 LLM
输入文本,待 llm-gateway 合入后由调用方注入真正的 LLM 生成函数。
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional, Tuple
# ---------------------------------------------------------------------------
# 版本标识(被交接班摘要配置资产的 $schema 引用)
# ---------------------------------------------------------------------------
HANDOVER_CONFIG_SCHEMA_ID: str = "iAOP-cockpit-handover-brief-v1"
HANDOVER_CONFIG_SCHEMA_VERSION: int = 1
# ---------------------------------------------------------------------------
# 提示词模板绑定(与 core/llm-gateway/config/prompts.template.yaml 对齐)
# ---------------------------------------------------------------------------
# PRD 5.4「Prompt 版本管理」:交接班摘要绑定的提示词模板名 + 版本。
# prompts.template.yaml 中 shift_handover v1.0.1(current)正文含
# 「生产概况/异常事项/安全注意事项」三个章节(v1.0.1 相对 v1.0.0 补充安全章节)。
DEFAULT_PROMPT_TEMPLATE: str = "shift_handover"
DEFAULT_PROMPT_VERSION: str = "1.0.1"
# ---------------------------------------------------------------------------
# 移动端展示策略(默认值,可被模板配置资产覆盖)
# ---------------------------------------------------------------------------
# PRD NFR:移动端只读并发 ≥ 200;交接班摘要是只读卡片,默认折叠次要章节。
DEFAULT_MAX_EVENTS: int = 8
MAX_ALLOWED_EVENTS: int = 50
DEFAULT_MAX_TODOS: int = 5
MAX_ALLOWED_TODOS: int = 30
# 移动端章节折叠阈值:超过该条数则折叠为「展开更多」。
DEFAULT_COLLAPSE_THRESHOLD: int = 4
# 移动端字号档位(与驾驶舱主题档位对齐)。
VALID_FONT_SIZES: Tuple[str, ...] = ("sm", "md", "lg")
# 合法的章节顺序键(与 shift_handover 提示词三章节 + 摘要额外章节对齐)。
VALID_SECTIONS: Tuple[str, ...] = (
"overview", # 生产概况
"abnormal", # 异常事项
"safety", # 安全注意事项
"energy", # 能耗
"todos", # 待办(交下一班)
)
# 高利害标记:交接班摘要通常非高利害(scenarios.template.yaml: high_stakes=false),
# 但若本班出现 P0 告警 / 安全事件,可强制 require_confirm=true(PRD 高利害人工确认)。
DEFAULT_REQUIRE_CONFIRM: bool = False
# 验收线:交接班报告生成 ≤ 2 分钟(PRD line 350),单位毫秒。
HANDOVER_GENERATION_BUDGET_MS: int = 120_000
# ---------------------------------------------------------------------------
# 异常 / 结果
# ---------------------------------------------------------------------------
class HandoverConfigError(ValueError):
"""交接班摘要配置校验失败。``load_handover_config`` 在校验不通过时抛出。
``errors`` 收集全部字段级错误,沿用 #50 / #52 的多错误聚合风格。
"""
def __init__(self, errors: List[str]):
super().__init__(
"; ".join(errors) if errors else "handover brief config validation failed"
)
self.errors: List[str] = list(errors)
@dataclass
class HandoverConfigValidationResult:
"""``validate_handover_config`` 的返回值。"""
ok: bool
errors: List[str] = field(default_factory=list)
normalized: Optional[Dict[str, Any]] = None
# ---------------------------------------------------------------------------
# 班次数据内存模型
# ---------------------------------------------------------------------------
@dataclass
class ShiftEvent:
"""本班一条关键事件(告警 / 操作 / 质量波动等)。
Attributes:
time: 事件时间(ISO 字符串或可读文本,如 ``"03:12"``)。
title: 事件标题(移动端卡片主文案)。
severity: 严重度(``P0`` / ``P1`` / ``P2`` / ``info``;与告警面板对齐)。
detail: 可选详情(移动端「展开更多」)。
"""
time: str
title: str
severity: str = "info"
detail: Optional[str] = None
@dataclass
class ShiftTodo:
"""交接给下一班的待办。
Attributes:
title: 待办内容。
priority: 优先级(``high`` / ``medium`` / ``low``)。
due: 可选截止提示(如 ``"接班后 30min 内"``)。
"""
title: str
priority: str = "medium"
due: Optional[str] = None
@dataclass
class ShiftRecord:
"""一个班次的归一化数据(移动端交接班摘要的输入)。
所有字段均有合理默认值:哪怕只给 ``shift`` / ``operator`` 也能产出一份
可用摘要(降级友好)。
Attributes:
shift: 班次标识(如 ``"夜班 2026-08-05 00:00~08:00"``)。
operator: 交班人/值班长。
overview: 生产概况自由文本(产量/批次/质量达成等)。
events: 关键事件列表(按时间顺序)。
todos: 交接待办列表。
energy: 能耗小结(自由文本或结构化键值,如 ``"总电耗 12.4 万 kWh"``)。
safety: 安全注意事项自由文本。
next_shift: 下一班接班人(可选,用于推送)。
"""
shift: str
operator: str
overview: str = ""
events: List[ShiftEvent] = field(default_factory=list)
todos: List[ShiftTodo] = field(default_factory=list)
energy: str = ""
safety: str = ""
next_shift: Optional[str] = None
# ---------------------------------------------------------------------------
# 摘要配置内存模型
# ---------------------------------------------------------------------------
@dataclass
class HandoverBriefConfig:
"""一份移动端交接班摘要配置(对应一份模板级配置资产)。
切换行业模板 = 加载另一份 ``HandoverBriefConfig``,移动端卡片代码零改动
(与 #52 告警面板配置化同一验收口径)。
"""
prompt_template: str = DEFAULT_PROMPT_TEMPLATE
prompt_version: str = DEFAULT_PROMPT_VERSION
sections: List[str] = field(
default_factory=lambda: list(VALID_SECTIONS)
)
max_events: int = DEFAULT_MAX_EVENTS
max_todos: int = DEFAULT_MAX_TODOS
collapse_threshold: int = DEFAULT_COLLAPSE_THRESHOLD
font_size: str = "md"
require_confirm: bool = DEFAULT_REQUIRE_CONFIRM
schema: str = HANDOVER_CONFIG_SCHEMA_ID
# ---------------------------------------------------------------------------
# 校验器
# ---------------------------------------------------------------------------
def _is_str_nonempty(value: Any) -> bool:
return isinstance(value, str) and value.strip() != ""
def validate_handover_config(config: Any) -> HandoverConfigValidationResult:
"""对一份交接班摘要配置资产做结构 + 语义校验,返回校验结果。
非法资产不提前返回:尽量收集全部字段级错误,便于配置台一次性展示
「错误列表(字段 + 原因)」,与 #50 / #52 行为一致。
"""
errors: List[str] = []
if not isinstance(config, dict):
return HandoverConfigValidationResult(
ok=False,
errors=[f"配置必须是对象(dict),实际为 {type(config).__name__}"],
)
# $schema(选填,但若给出必须对齐版本标识)
schema = config.get("$schema")
if schema is not None and schema != HANDOVER_CONFIG_SCHEMA_ID:
errors.append(
f"$schema: 应为 {HANDOVER_CONFIG_SCHEMA_ID!r},实际为 {schema!r}"
)
# promptTemplate:非空字符串
prompt_template = config.get(
"promptTemplate", config.get("prompt_template", DEFAULT_PROMPT_TEMPLATE)
)
if not _is_str_nonempty(prompt_template):
errors.append("promptTemplate: 缺失或非非空字符串")
# promptVersion:非空字符串(semver 由 llm-gateway 校验,这里只要求非空)
prompt_version = config.get(
"promptVersion", config.get("prompt_version", DEFAULT_PROMPT_VERSION)
)
if not _is_str_nonempty(prompt_version):
errors.append("promptVersion: 缺失或非非空字符串")
# sections:列表、每项在 VALID_SECTIONS 内、至少 1 项、无重复
raw_sections = config.get("sections", list(VALID_SECTIONS))
norm_sections: List[str] = []
if not isinstance(raw_sections, list) or not raw_sections:
errors.append(
f"sections: 缺失或非列表(至少 1 项,合法值 {list(VALID_SECTIONS)})"
)
else:
seen: set = set()
for i, s in enumerate(raw_sections):
if not _is_str_nonempty(s):
errors.append(f"sections[{i}]: 必须是非空字符串")
continue
if s not in VALID_SECTIONS:
errors.append(
f"sections[{i}]: 非法 {s!r},合法值 {list(VALID_SECTIONS)}"
)
continue
if s in seen:
errors.append(f"sections[{i}]: {s!r} 重复声明")
continue
seen.add(s)
norm_sections.append(s)
# maxEvents:1..MAX_ALLOWED_EVENTS
max_events = config.get("maxEvents", config.get("max_events", DEFAULT_MAX_EVENTS))
if (
isinstance(max_events, bool)
or not isinstance(max_events, int)
or not (1 <= max_events <= MAX_ALLOWED_EVENTS)
):
errors.append(
f"maxEvents: 必须是 1..{MAX_ALLOWED_EVENTS} 的整数,实际为 {max_events!r}"
)
# maxTodos:1..MAX_ALLOWED_TODOS
max_todos = config.get("maxTodos", config.get("max_todos", DEFAULT_MAX_TODOS))
if (
isinstance(max_todos, bool)
or not isinstance(max_todos, int)
or not (1 <= max_todos <= MAX_ALLOWED_TODOS)
):
errors.append(
f"maxTodos: 必须是 1..{MAX_ALLOWED_TODOS} 的整数,实际为 {max_todos!r}"
)
# collapseThreshold:0..max(maxEvents, maxTodos)
collapse_threshold = config.get(
"collapseThreshold", config.get("collapse_threshold", DEFAULT_COLLAPSE_THRESHOLD)
)
if (
isinstance(collapse_threshold, bool)
or not isinstance(collapse_threshold, int)
or collapse_threshold < 0
):
errors.append(
f"collapseThreshold: 必须是非负整数,实际为 {collapse_threshold!r}"
)
# fontSize:枚举
font_size = config.get("fontSize", config.get("font_size", "md"))
if font_size not in VALID_FONT_SIZES:
errors.append(
f"fontSize: 非法 {font_size!r},合法值 {list(VALID_FONT_SIZES)}"
)
# requireConfirm:布尔
require_confirm = config.get(
"requireConfirm", config.get("require_confirm", DEFAULT_REQUIRE_CONFIRM)
)
if not isinstance(require_confirm, bool):
errors.append(
f"requireConfirm: 必须是布尔,实际为 {type(require_confirm).__name__}"
)
if errors:
return HandoverConfigValidationResult(ok=False, errors=errors)
normalized: Dict[str, Any] = {
"$schema": HANDOVER_CONFIG_SCHEMA_ID,
"promptTemplate": prompt_template,
"promptVersion": prompt_version,
"sections": norm_sections or list(VALID_SECTIONS),
"maxEvents": max_events,
"maxTodos": max_todos,
"collapseThreshold": collapse_threshold,
"fontSize": font_size,
"requireConfirm": require_confirm,
}
return HandoverConfigValidationResult(ok=True, errors=[], normalized=normalized)
def load_handover_config(config: Any) -> HandoverBriefConfig:
"""从已解析的 dict 构造 ``HandoverBriefConfig``(校验失败抛 ``HandoverConfigError``)。
与 #52 ``load_alarm_config`` 对称:先校验再按规范化结果构造内存模型。
"""
result = validate_handover_config(config)
if not result.ok:
raise HandoverConfigError(result.errors)
assert result.normalized is not None
n = result.normalized
return HandoverBriefConfig(
prompt_template=n["promptTemplate"],
prompt_version=n["promptVersion"],
sections=n["sections"],
max_events=n["maxEvents"],
max_todos=n["maxTodos"],
collapse_threshold=n["collapseThreshold"],
font_size=n["fontSize"],
require_confirm=n["requireConfirm"],
schema=HANDOVER_CONFIG_SCHEMA_ID,
)
# ---------------------------------------------------------------------------
# 班次数据归一化(从 dict → ShiftRecord)
# ---------------------------------------------------------------------------
def normalize_shift_record(raw: Any) -> ShiftRecord:
"""把一份班次数据 dict 归一为 ``ShiftRecord``。
用于从驾驶舱 / 总线 / 告警面板聚合出的原始数据(JSON)转入内存模型。
宽松解析:缺字段给默认值;非法字段抛 ``HandoverConfigError`` 携带错误清单。
"""
errors: List[str] = []
if not isinstance(raw, dict):
raise HandoverConfigError(
[f"班次数据必须是对象(dict),实际为 {type(raw).__name__}"]
)
shift = raw.get("shift", "")
operator = raw.get("operator", "")
if not _is_str_nonempty(shift):
errors.append("shift: 缺失或非非空字符串(班次标识必填)")
if not _is_str_nonempty(operator):
errors.append("operator: 缺失或非非空字符串(交班人必填)")
events: List[ShiftEvent] = []
raw_events = raw.get("events", [])
if raw_events is None:
raw_events = []
if not isinstance(raw_events, list):
errors.append("events: 必须是列表")
else:
for i, e in enumerate(raw_events):
if not isinstance(e, dict):
errors.append(f"events[{i}]: 必须是对象(dict)")
continue
t = e.get("time")
title = e.get("title")
if not _is_str_nonempty(t):
errors.append(f"events[{i}].time: 缺失或非非空字符串")
continue
if not _is_str_nonempty(title):
errors.append(f"events[{i}].title: 缺失或非非空字符串")
continue
events.append(
ShiftEvent(
time=t,
title=title,
severity=e.get("severity", "info"),
detail=e.get("detail"),
)
)
todos: List[ShiftTodo] = []
raw_todos = raw.get("todos", [])
if raw_todos is None:
raw_todos = []
if not isinstance(raw_todos, list):
errors.append("todos: 必须是列表")
else:
for i, td in enumerate(raw_todos):
if not isinstance(td, dict):
errors.append(f"todos[{i}]: 必须是对象(dict)")
continue
title = td.get("title")
if not _is_str_nonempty(title):
errors.append(f"todos[{i}].title: 缺失或非非空字符串")
continue
todos.append(
ShiftTodo(
title=title,
priority=td.get("priority", "medium"),
due=td.get("due"),
)
)
if errors:
raise HandoverConfigError(errors)
return ShiftRecord(
shift=str(shift),
operator=str(operator),
overview=str(raw.get("overview", "")) or "",
events=events,
todos=todos,
energy=str(raw.get("energy", "")) or "",
safety=str(raw.get("safety", "")) or "",
next_shift=raw.get("next_shift"),
)
# ---------------------------------------------------------------------------
# LLM 输入渲染(填 shift_handover 提示词占位符)
# ---------------------------------------------------------------------------
def build_llm_input(
record: ShiftRecord,
config: Optional[HandoverBriefConfig] = None,
) -> Tuple[str, str]:
"""把班次数据填入 ``shift_handover`` 提示词,产出给 LLM 网关的输入。
返回 ``(prompt_template_name, rendered_text)``:
- ``prompt_template_name`` 来自配置(默认 ``shift_handover``),调用方据此
绑定 llm-gateway 的 ``PromptRegistry`` 版本(PRD「运行时绑定模板版本」);
- ``rendered_text`` 是提示词正文(占位符 ``{query}`` 已替换为本班结构化摘要)。
与 prompts.template.yaml 中 ``shift_handover v1.0.1`` 正文
(「生成交接班摘要,包含:生产概况、异常事项、安全注意事项。班次:{query}」)
对齐:把结构化班次数据序列化为 ``{query}`` 占位的班次文本。
"""
if config is None:
config = HandoverBriefConfig()
# 班次结构化摘要文本(作为 {query} 的填充)
parts: List[str] = [f"班次:{record.shift}", f"交班人:{record.operator}"]
if record.overview:
parts.append(f"生产概况:{record.overview}")
if record.events:
lines = ["关键事件:"]
for e in record.events:
tag = f"[{e.severity}]" if e.severity and e.severity != "info" else ""
line = f"- {e.time} {tag} {e.title}".rstrip()
if e.detail:
line += f"({e.detail})"
lines.append(line)
parts.append("\n".join(lines))
if record.energy:
parts.append(f"能耗:{record.energy}")
if record.safety:
parts.append(f"安全注意事项:{record.safety}")
if record.todos:
lines = ["待办(交下一班):"]
for td in record.todos:
line = f"- [{td.priority}] {td.title}"
if td.due:
line += f"({td.due})"
lines.append(line)
parts.append("\n".join(lines))
query = "\n".join(parts)
# shift_handover 提示词正文:生成交接班摘要,包含:生产概况、异常事项、安全注意事项。班次:{query}
rendered = (
"生成交接班摘要,包含:生产概况、异常事项、安全注意事项。班次:\n" + query
)
return config.prompt_template, rendered
# ---------------------------------------------------------------------------
# 确定性降级渲染(无 LLM 时的离线摘要)
# ---------------------------------------------------------------------------
def render_deterministic_brief(
record: ShiftRecord,
config: Optional[HandoverBriefConfig] = None,
) -> str:
"""无 LLM(dry-run / 离线 / LLM 故障降级)时的确定性交接班摘要。
PRD NFR「可用性 ≥ 99.8%」「模型服务故障自动降级」:当 LLM 网关不可用时,
交接班摘要链路不能中断,回落到基于班次数据的模板拼接摘要。输出与
``shift_handover`` 提示词三章节(生产概况/异常事项/安全注意事项)结构对齐,
额外补充能耗与待办章节。
"""
if config is None:
config = HandoverBriefConfig()
sections = config.sections
blocks: List[str] = [f"# {record.shift} 交接班摘要", f"交班人:{record.operator}"]
def _add(title: str, body: str) -> None:
if body and body.strip():
blocks.append(f"\n## {title}\n{body.strip()}")
if "overview" in sections:
_add("生产概况", record.overview or "(本班生产平稳,无显著波动)")
if "abnormal" in sections:
if record.events:
lines = []
for e in record.events:
tag = f"[{e.severity}]" if e.severity and e.severity != "info" else ""
line = f"- {e.time} {tag} {e.title}".rstrip()
if e.detail:
line += f"({e.detail})"
lines.append(line)
_add("异常事项", "\n".join(lines))
else:
_add("异常事项", "本班无关键异常事件。")
if "safety" in sections:
_add("安全注意事项", record.safety or "无特殊安全注意事项。")
if "energy" in sections:
_add("能耗", record.energy or "(未填报能耗数据)")
if "todos" in sections:
if record.todos:
lines = [
f"- [{td.priority}] {td.title}" + (f"({td.due})" if td.due else "")
for td in record.todos
]
_add("待办(交下一班)", "\n".join(lines))
else:
_add("待办(交下一班)", "无待办事项。")
return "\n".join(blocks)
# ---------------------------------------------------------------------------
# 摘要生成主入口(注入 LLM,离线自动降级)
# ---------------------------------------------------------------------------
def generate_handover_brief(
record: ShiftRecord,
config: Optional[HandoverBriefConfig] = None,
llm_generate: Optional[Callable[[str], str]] = None,
) -> str:
"""生成交接班摘要文本(PRD 场景C,验收线 ≤ 2 分钟)。
Args:
record: 归一化班次数据。
config: 摘要配置(章节/移动端策略/提示词模板);None 用默认。
llm_generate: LLM 生成函数(输入提示词文本,返回摘要文本)。None 或
调用失败时**自动降级**为 ``render_deterministic_brief``(离线可用)。
Returns:
交接班摘要文本(含生产概况/异常事项/安全注意事项等章节)。
设计:LLM 以依赖注入传入,内核不绑定具体云端 SDK,保证本地闭环与离线降级。
"""
if config is None:
config = HandoverBriefConfig()
if llm_generate is None:
return render_deterministic_brief(record, config)
_, rendered = build_llm_input(record, config)
try:
text = llm_generate(rendered)
except Exception:
# LLM 故障 → 降级为确定性摘要(PRD「模型服务故障自动降级」)
return render_deterministic_brief(record, config)
if not _is_str_nonempty(text):
return render_deterministic_brief(record, config)
return text
# ---------------------------------------------------------------------------
# 移动端负载编译(摘要 + 班次要点 → 移动端只读卡片 props)
# ---------------------------------------------------------------------------
def render_handover_brief(
record: ShiftRecord,
config: Optional[HandoverBriefConfig] = None,
summary: Optional[str] = None,
llm_generate: Optional[Callable[[str], str]] = None,
generated_at: Optional[datetime] = None,
) -> Dict[str, Any]:
"""把班次数据 + 摘要编译成移动端只读卡片可直接消费的 props dict。
Args:
record: 班次数据。
config: 摘要配置;None 用默认。
summary: 已生成的摘要文本;None 则调用 ``generate_handover_brief`` 生成。
llm_generate: 透传给 ``generate_handover_brief``(None 时离线降级)。
generated_at: 摘要生成时间(用于卡片时间戳);None 取当前 UTC。
Returns:
扁平、camelCase、可 ``json.dumps`` 的 props dict,移动端 ``HandoverBrief``
组件据此渲染标题/章节/事件/待办/折叠策略/字号(**切换行业模板只换配置资产,
移动端代码零改动**,与 #52 同一验收口径)。
"""
if config is None:
config = HandoverBriefConfig()
if summary is None:
summary = generate_handover_brief(record, config, llm_generate)
if generated_at is None:
generated_at = datetime.now(timezone.utc)
# 按 maxEvents / maxTodos 截断,移动端展示策略由配置驱动
truncated_events = record.events[: config.max_events]
truncated_todos = record.todos[: config.max_todos]
# 是否含 P0 / safety 事件 → 强制 requireConfirm(PRD 高利害人工确认)
has_critical = any(e.severity == "P0" for e in record.events) or bool(
record.safety and record.safety.strip()
)
require_confirm = config.require_confirm or has_critical
props: Dict[str, Any] = {
"schema": config.schema,
"shift": record.shift,
"operator": record.operator,
"nextShift": record.next_shift,
"summary": summary,
"generatedAt": generated_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
"promptTemplate": config.prompt_template,
"promptVersion": config.prompt_version,
"sections": list(config.sections),
"events": [
{
"time": e.time,
"title": e.title,
"severity": e.severity,
"detail": e.detail,
}
for e in truncated_events
],
"todos": [
{"title": t.title, "priority": t.priority, "due": t.due}
for t in truncated_todos
],
"eventOverflow": max(0, len(record.events) - config.max_events),
"todoOverflow": max(0, len(record.todos) - config.max_todos),
"collapseThreshold": config.collapse_threshold,
"fontSize": config.font_size,
"requireConfirm": require_confirm,
# 演示「配置点真实驱动展示」:章节开关决定卡片渲染哪些区块
"showOverview": "overview" in config.sections,
"showAbnormal": "abnormal" in config.sections,
"showSafety": "safety" in config.sections,
"showEnergy": "energy" in config.sections,
"showTodos": "todos" in config.sections,
}
return props
def within_generation_budget(elapsed_ms: float) -> bool:
"""判断本次摘要生成是否满足 PRD 验收线「交接班报告生成 ≤ 2 分钟」。
供验收脚本断言耗时用。
"""
return elapsed_ms <= HANDOVER_GENERATION_BUDGET_MS
@@ -0,0 +1,208 @@
# -*- coding: utf-8 -*-
"""移动端交接班摘要验收脚本(issue #53,PRD 场景C 验收口径)。
验证四个能力点(覆盖 PRD「交接班报告生成 ≤ 2 分钟」「移动端交接班摘要可用」):
1. **Ti 行业模板配置资产合法**:交接班摘要配置通过 ``iAOP-cockpit-handover-brief-v1``
校验,并能构造 ``HandoverBriefConfig``;
2. **LLM 注入生成 + 离线降级**:注入 LLM 时用其输出;LLM 故障/未注入时自动降级为
确定性摘要(PRD「模型服务故障自动降级」「可用性 ≥ 99.8%」),且降级摘要含
「生产概况/异常事项/安全注意事项」三章节(对齐 shift_handover 提示词);
3. **配置点真实驱动展示**:``sections`` / ``maxEvents`` / ``requireConfirm`` 配置项
改变 → 移动端 props 随之变化(切换模板零改码,与 #52 同一验收口径);含 P0 事件
→ ``requireConfirm`` 被强制为 true(PRD 高利害人工确认);
4. **生成耗时达标**:确定性降级路径在 2 分钟验收线内(PRD line 350)。
用法(在 core/shift-handover 目录下):
python scripts/verify_handover_brief.py
退出码:0 = 全部通过;1 = 存在未达标项。
说明:本脚本不依赖 PyYAML(保持零运行时依赖),用与 Ti 模板资产等价的 dict 校验。
"""
from __future__ import annotations
import os
import sys
import time
# 本脚本位于 core/shift-handover/scripts/,需把 core/ 加入 sys.path
sys.path.insert(
0,
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
)
# 含连字符的目录无法直接以包名 import:挂载为 shift_handover 后导入子模块
import types # noqa: E402
_SHIFT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if "shift_handover" not in sys.modules:
_pkg = types.ModuleType("shift_handover")
_pkg.__path__ = [_SHIFT_DIR]
sys.modules["shift_handover"] = _pkg
from shift_handover.handover import ( # noqa: E402
HANDOVER_CONFIG_SCHEMA_ID,
HANDOVER_GENERATION_BUDGET_MS,
HandoverBriefConfig,
generate_handover_brief,
load_handover_config,
normalize_shift_record,
render_handover_brief,
within_generation_budget,
)
def _ti_config() -> dict:
"""与 templates/ti-cl4/dashboard/handover_brief.ti.yaml 等价的 dict。"""
return {
"$schema": HANDOVER_CONFIG_SCHEMA_ID,
"promptTemplate": "shift_handover",
"promptVersion": "1.0.1",
"sections": ["overview", "abnormal", "safety", "energy", "todos"],
"maxEvents": 8,
"maxTodos": 5,
"collapseThreshold": 4,
"fontSize": "md",
"requireConfirm": False,
}
def _ti_shift_record() -> dict:
"""Ti 氯化车间夜班典型班次数据。"""
return {
"shift": "夜班 2026-08-05 00:00~08:00",
"operator": "张工",
"overview": "TiCl4 产量 36.2t,质量达成率 98.6%。",
"events": [
{"time": "01:20", "title": "炉层温度越上限",
"severity": "P1", "detail": "CLF-01 第3层 942℃,已调风量回落"},
{"time": "03:05", "title": "夜巡正常", "severity": "info"},
],
"todos": [
{"title": "白班复测 3 层温度趋势", "priority": "high", "due": "接班后 1h"},
{"title": "补录 LIMS 03:00 批次", "priority": "medium"},
],
"energy": "总电耗 12.4 万 kWh",
"safety": "夜班无安全事件;注意 3 层高温区巡检。",
"next_shift": "李工",
}
def main() -> int:
all_ok = True
def _check(title: str, fn):
nonlocal all_ok
try:
ok = fn()
except Exception as exc: # noqa: BLE001
ok = False
print(f" ✗ {title}:异常 {exc!r}")
status = "✅" if ok else "❌"
print(f" {status} {title}")
all_ok = all_ok and ok
print("=" * 64)
print("移动端交接班摘要验收(issue #53 / PRD 场景C)")
print("=" * 64)
# 1) Ti 模板配置资产合法
def check_config_valid() -> bool:
cfg = load_handover_config(_ti_config())
return (
isinstance(cfg, HandoverBriefConfig)
and cfg.prompt_template == "shift_handover"
and cfg.prompt_version == "1.0.1"
and cfg.font_size == "md"
)
_check("Ti 模板配置资产合法(iAOP-cockpit-handover-brief-v1)", check_config_valid)
# 2) LLM 注入生成 + 离线降级
def check_llm_and_fallback() -> bool:
rec = normalize_shift_record(_ti_shift_record())
# 注入 LLM:用其输出
def llm(prompt: str) -> str:
return "【LLM 摘要】本班平稳,3 层温度曾越限已处置,交白班复测。"
with_llm = generate_handover_brief(rec, llm_generate=llm)
# 离线降级(不注入 LLM)
fallback = generate_handover_brief(rec)
# LLM 故障降级
def broken(prompt: str) -> str:
raise RuntimeError("LLM 网关不可达")
degraded = generate_handover_brief(rec, llm_generate=broken)
return (
with_llm.startswith("【LLM 摘要】")
and "交接班摘要" in fallback
and "生产概况" in fallback
and "异常事项" in fallback
and "安全注意事项" in fallback
and "交接班摘要" in degraded # 故障也降级为可用摘要
)
_check("LLM 注入生成 + 离线/故障降级为可用摘要", check_llm_and_fallback)
# 3) 配置点真实驱动移动端展示
def check_config_drives_props() -> bool:
rec = normalize_shift_record(_ti_shift_record())
props = render_handover_brief(rec)
# 默认配置下事件/待办齐全、章节开关正确
base_ok = (
len(props["events"]) == 2
and len(props["todos"]) == 2
and props["showOverview"] is True
and props["showAbnormal"] is True
and props["eventOverflow"] == 0
)
# sections 收窄 → 对应章节开关关闭
cfg2 = HandoverBriefConfig(sections=["overview"])
props2 = render_handover_brief(rec, cfg2)
sections_ok = (
props2["showOverview"] is True
and props2["showAbnormal"] is False
and props2["showTodos"] is False
)
# maxEvents 截断 + 溢出计数
many = {
"shift": "夜班", "operator": "张工",
"events": [{"time": f"0{i}:00", "title": f"事件{i}"} for i in range(12)],
}
rec_many = normalize_shift_record(many)
cfg3 = HandoverBriefConfig(max_events=5)
props3 = render_handover_brief(rec_many, cfg3)
trunc_ok = len(props3["events"]) == 5 and props3["eventOverflow"] == 7
# 含 P0 事件 → requireConfirm 被强制为 true
critical = {
"shift": "夜班", "operator": "张工",
"events": [{"time": "01:00", "title": "严重告警", "severity": "P0"}],
}
rec_crit = normalize_shift_record(critical)
props4 = render_handover_brief(rec_crit, HandoverBriefConfig(require_confirm=False))
crit_ok = props4["requireConfirm"] is True
return base_ok and sections_ok and trunc_ok and crit_ok
_check("配置点(sections/maxEvents/requireConfirm)真实驱动移动端展示", check_config_drives_props)
# 4) 生成耗时达标(≤ 2 分钟验收线)
def check_generation_budget() -> bool:
rec = normalize_shift_record(_ti_shift_record())
start = time.perf_counter()
for _ in range(50):
generate_handover_brief(rec) # 离线降级路径
elapsed_ms = (time.perf_counter() - start) * 1000.0
# 单次远小于预算(取平均更稳),且预算判定函数正确
per_call = elapsed_ms / 50.0
return (
per_call < HANDOVER_GENERATION_BUDGET_MS
and within_generation_budget(per_call)
and not within_generation_budget(HANDOVER_GENERATION_BUDGET_MS + 1)
)
_check("生成耗时达标(PRD:交接班报告生成 ≤ 2 分钟)", check_generation_budget)
print("=" * 64)
print("结果: " + ("全部通过 ✅" if all_ok else "存在未达标项 ❌"))
return 0 if all_ok else 1
if __name__ == "__main__":
raise SystemExit(main())
+17
View File
@@ -0,0 +1,17 @@
# -*- coding: utf-8 -*-
"""测试引导:把 `core/shift-handover` 以包名 `shift_handover` 挂载到 sys.modules。
目录名 `shift-handover` 含连字符,无法直接以包名 import;挂载后模块内相对导入
(`from .handover import ...`)在 unittest 发现机制下可正常解析。
(沿用 core/data-bus/tests/_bootstrap.py 的同一做法。)
"""
import os
import sys
import types
SHIFT_HANDOVER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, SHIFT_HANDOVER_DIR)
if "shift_handover" not in sys.modules:
pkg = types.ModuleType("shift_handover")
pkg.__path__ = [SHIFT_HANDOVER_DIR]
sys.modules["shift_handover"] = pkg
+400
View File
@@ -0,0 +1,400 @@
# -*- coding: utf-8 -*-
"""移动端交接班摘要生成单元测试(issue #53 / PRD 场景C / 5.4「④ LLM 网关」)。
覆盖:
1. 配置校验:合法配置通过、默认值与字段别名、sections 去重与枚举校验、
maxEvents/maxTodos/collapseThreshold/fontSize/requireConfirm 各类非法情况;
2. ``load_handover_config`` 校验失败抛 ``HandoverConfigError`` 并携带全部错误;
3. ``normalize_shift_record``:合法 dict → ShiftRecord;缺必填/非法结构被拒;
4. ``build_llm_input``:占位符填充正确、含班次/事件/能耗/待办;
5. ``render_deterministic_brief``:章节与配置 sections 联动;
6. ``generate_handover_brief``:注入 LLM 用其输出;LLM 抛异常/返回空 → 自动降级;
7. ``render_handover_brief``:props 结构正确、截断/溢出计数/高利害强制确认;
8. ``within_generation_budget``:2 分钟验收线判定。
"""
from __future__ import annotations
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _bootstrap # noqa: F401 把 core/shift-handover 挂载为 shift_handover 包
from shift_handover.handover import ( # type: ignore[import-not-found]
DEFAULT_PROMPT_TEMPLATE,
DEFAULT_PROMPT_VERSION,
HANDOVER_CONFIG_SCHEMA_ID,
HANDOVER_GENERATION_BUDGET_MS,
HandoverBriefConfig,
HandoverConfigError,
HandoverConfigValidationResult,
ShiftEvent,
ShiftRecord,
ShiftTodo,
build_llm_input,
generate_handover_brief,
load_handover_config,
normalize_shift_record,
render_deterministic_brief,
render_handover_brief,
validate_handover_config,
within_generation_budget,
)
def _valid_config() -> dict:
"""返回一份合法的交接班摘要配置(Ti 模板风格)。"""
return {
"$schema": HANDOVER_CONFIG_SCHEMA_ID,
"promptTemplate": "shift_handover",
"promptVersion": "1.0.1",
"sections": ["overview", "abnormal", "safety", "energy", "todos"],
"maxEvents": 8,
"maxTodos": 5,
"collapseThreshold": 4,
"fontSize": "md",
"requireConfirm": False,
}
def _full_record_dict() -> dict:
return {
"shift": "夜班 2026-08-05 00:00~08:00",
"operator": "张工",
"overview": "TiCl4 产量 36.2t,质量达成率 98.6%。",
"events": [
{"time": "01:20", "title": "炉层温度越上限", "severity": "P1",
"detail": "CLF-01 第3层 942℃ → 已调风量"},
{"time": "03:05", "title": "夜巡正常", "severity": "info"},
],
"todos": [
{"title": "白班复测 3 层温度趋势", "priority": "high", "due": "接班后 1h"},
{"title": "补录 LIMS 03:00 批次", "priority": "medium"},
],
"energy": "总电耗 12.4 万 kWh",
"safety": "夜班无安全事件;注意 3 层高温区巡检。",
"next_shift": "李工",
}
# ---------------------------------------------------------------------------
# 1) 配置校验
# ---------------------------------------------------------------------------
class TestValidateConfig(unittest.TestCase):
def test_valid_config_passes(self):
res = validate_handover_config(_valid_config())
self.assertTrue(res.ok, msg=res.errors)
self.assertEqual(res.errors, [])
self.assertEqual(res.normalized["$schema"], HANDOVER_CONFIG_SCHEMA_ID)
def test_defaults_and_aliases(self):
# snake_case 与 camelCase 都接受;缺省给默认值
cfg = {"prompt_template": "shift_handover", "prompt_version": "1.0.0"}
res = validate_handover_config(cfg)
self.assertTrue(res.ok, msg=res.errors)
n = res.normalized
self.assertEqual(n["maxEvents"], 8)
self.assertEqual(n["maxTodos"], 5)
self.assertEqual(n["fontSize"], "md")
self.assertEqual(n["collapseThreshold"], 4)
self.assertFalse(n["requireConfirm"])
# sections 默认为全部合法章节
self.assertEqual(
n["sections"],
["overview", "abnormal", "safety", "energy", "todos"],
)
def test_wrong_schema_rejected(self):
cfg = _valid_config()
cfg["$schema"] = "something-else"
res = validate_handover_config(cfg)
self.assertFalse(res.ok)
self.assertTrue(any("$schema" in e for e in res.errors))
def test_non_dict_rejected(self):
res = validate_handover_config(["not", "a", "dict"])
self.assertFalse(res.ok)
self.assertEqual(len(res.errors), 1)
def test_sections_validation(self):
# 非法章节、重复、空列表、非列表
cases = [
(["overview", "unknown"], "非法枚举"),
(["overview", "overview"], "重复"),
([], "空列表"),
("not-a-list", "非列表"),
]
for bad_sections, hint in cases:
cfg = _valid_config()
cfg["sections"] = bad_sections
res = validate_handover_config(cfg)
self.assertFalse(res.ok, msg=f"{hint} 应被拒绝: {res.errors}")
def test_numeric_bounds(self):
for field, bad in [
("maxEvents", 0), ("maxEvents", 51), ("maxEvents", True),
("maxTodos", 0), ("maxTodos", 31), ("maxTodos", "5"),
("collapseThreshold", -1), ("collapseThreshold", True),
]:
cfg = _valid_config()
cfg[field] = bad
res = validate_handover_config(cfg)
self.assertFalse(res.ok, msg=f"{field}={bad!r} 应被拒绝: {res.errors}")
def test_font_size_enum(self):
cfg = _valid_config()
cfg["fontSize"] = "xl"
res = validate_handover_config(cfg)
self.assertFalse(res.ok)
def test_require_confirm_type(self):
cfg = _valid_config()
cfg["requireConfirm"] = "yes"
res = validate_handover_config(cfg)
self.assertFalse(res.ok)
def test_aggregates_multiple_errors(self):
bad = {
"promptTemplate": "",
"sections": ["oops"],
"maxEvents": 0,
"fontSize": "xxl",
"requireConfirm": 1,
}
res = validate_handover_config(bad)
self.assertFalse(res.ok)
# 至少 5 条字段级错误被聚合
self.assertGreaterEqual(len(res.errors), 5)
# ---------------------------------------------------------------------------
# 2) load_handover_config
# ---------------------------------------------------------------------------
class TestLoadConfig(unittest.TestCase):
def test_load_valid(self):
cfg = load_handover_config(_valid_config())
self.assertIsInstance(cfg, HandoverBriefConfig)
self.assertEqual(cfg.prompt_template, "shift_handover")
self.assertEqual(cfg.prompt_version, "1.0.1")
self.assertEqual(cfg.font_size, "md")
def test_load_invalid_raises_with_errors(self):
with self.assertRaises(HandoverConfigError) as ctx:
load_handover_config({"promptTemplate": "", "maxEvents": 0})
# 异常携带错误清单
self.assertGreaterEqual(len(ctx.exception.errors), 2)
# ---------------------------------------------------------------------------
# 3) normalize_shift_record
# ---------------------------------------------------------------------------
class TestNormalizeRecord(unittest.TestCase):
def test_full_record(self):
rec = normalize_shift_record(_full_record_dict())
self.assertIsInstance(rec, ShiftRecord)
self.assertEqual(rec.shift, "夜班 2026-08-05 00:00~08:00")
self.assertEqual(rec.operator, "张工")
self.assertEqual(len(rec.events), 2)
self.assertEqual(rec.events[0].severity, "P1")
self.assertEqual(rec.todos[0].priority, "high")
self.assertEqual(rec.next_shift, "李工")
def test_minimal_record(self):
rec = normalize_shift_record({"shift": "白班", "operator": "王五"})
self.assertEqual(rec.events, [])
self.assertEqual(rec.todos, [])
self.assertEqual(rec.overview, "")
def test_missing_required_rejected(self):
with self.assertRaises(HandoverConfigError):
normalize_shift_record({"shift": "白班"}) # 缺 operator
def test_bad_event_rejected(self):
raw = _full_record_dict()
raw["events"][0] = {"time": "01:00"} # 缺 title
with self.assertRaises(HandoverConfigError):
normalize_shift_record(raw)
def test_bad_todo_rejected(self):
raw = _full_record_dict()
raw["todos"] = [{"priority": "high"}] # 缺 title
with self.assertRaises(HandoverConfigError):
normalize_shift_record(raw)
def test_non_dict_rejected(self):
with self.assertRaises(HandoverConfigError):
normalize_shift_record("not a dict")
# ---------------------------------------------------------------------------
# 4) build_llm_input
# ---------------------------------------------------------------------------
class TestBuildLLMInput(unittest.TestCase):
def test_renders_prompt_with_placeholders(self):
rec = normalize_shift_record(_full_record_dict())
name, text = build_llm_input(rec)
self.assertEqual(name, DEFAULT_PROMPT_TEMPLATE)
# 正文以 shift_handover 提示词开头
self.assertTrue(text.startswith("生成交接班摘要"))
# 班次与关键字段都被填入 {query}
self.assertIn("夜班 2026-08-05", text)
self.assertIn("张工", text)
self.assertIn("炉层温度越上限", text)
self.assertIn("总电耗", text)
self.assertIn("复测 3 层温度趋势", text)
def test_respects_config_template_name(self):
rec = normalize_shift_record({"shift": "夜班", "operator": "张工"})
cfg = HandoverBriefConfig(prompt_template="custom_tmpl")
name, _ = build_llm_input(rec, cfg)
self.assertEqual(name, "custom_tmpl")
# ---------------------------------------------------------------------------
# 5) render_deterministic_brief
# ---------------------------------------------------------------------------
class TestDeterministicBrief(unittest.TestCase):
def test_contains_all_sections(self):
rec = normalize_shift_record(_full_record_dict())
text = render_deterministic_brief(rec)
self.assertIn("交接班摘要", text)
self.assertIn("生产概况", text)
self.assertIn("异常事项", text)
self.assertIn("安全注意事项", text)
self.assertIn("能耗", text)
self.assertIn("待办", text)
def test_sections_filter(self):
rec = normalize_shift_record(_full_record_dict())
cfg = HandoverBriefConfig(sections=["overview"])
text = render_deterministic_brief(rec, cfg)
self.assertIn("生产概况", text)
# 只开 overview,异常/能耗等章节不应出现为标题
self.assertNotIn("## 异常事项", text)
self.assertNotIn("## 能耗", text)
# ---------------------------------------------------------------------------
# 6) generate_handover_brief(LLM 注入 + 降级)
# ---------------------------------------------------------------------------
class TestGenerateBrief(unittest.TestCase):
def test_uses_llm_output(self):
rec = normalize_shift_record(_full_record_dict())
def llm(prompt: str) -> str:
return "LLM 摘要:本班平稳,3 层温度曾越限已处置。"
out = generate_handover_brief(rec, llm_generate=llm)
self.assertEqual(out, "LLM 摘要:本班平稳,3 层温度曾越限已处置。")
def test_falls_back_when_no_llm(self):
rec = normalize_shift_record(_full_record_dict())
out = generate_handover_brief(rec) # llm_generate=None
self.assertIn("交接班摘要", out)
self.assertIn("炉层温度越上限", out)
def test_falls_back_on_llm_exception(self):
rec = normalize_shift_record(_full_record_dict())
def broken(prompt: str) -> str:
raise RuntimeError("LLM 网关不可达")
out = generate_handover_brief(rec, llm_generate=broken)
# LLM 故障 → 降级为确定性摘要
self.assertIn("交接班摘要", out)
def test_falls_back_on_empty_llm_output(self):
rec = normalize_shift_record(_full_record_dict())
def empty(prompt: str) -> str:
return " "
out = generate_handover_brief(rec, llm_generate=empty)
self.assertIn("交接班摘要", out)
# ---------------------------------------------------------------------------
# 7) render_handover_brief(移动端 props)
# ---------------------------------------------------------------------------
class TestRenderBriefProps(unittest.TestCase):
def test_props_structure(self):
rec = normalize_shift_record(_full_record_dict())
props = render_handover_brief(rec)
self.assertEqual(props["schema"], HANDOVER_CONFIG_SCHEMA_ID)
self.assertEqual(props["shift"], "夜班 2026-08-05 00:00~08:00")
self.assertEqual(props["operator"], "张工")
self.assertEqual(props["nextShift"], "李工")
self.assertEqual(props["promptTemplate"], DEFAULT_PROMPT_TEMPLATE)
self.assertEqual(props["promptVersion"], DEFAULT_PROMPT_VERSION)
self.assertEqual(len(props["events"]), 2)
self.assertEqual(len(props["todos"]), 2)
self.assertEqual(props["eventOverflow"], 0)
self.assertEqual(props["todoOverflow"], 0)
self.assertEqual(props["fontSize"], "md")
# generatedAt 是 ISO 时间戳
self.assertRegex(props["generatedAt"], r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
def test_truncation_and_overflow(self):
raw = {
"shift": "夜班", "operator": "张工",
"events": [{"time": f"0{i}:00", "title": f"事件{i}"} for i in range(12)],
"todos": [{"title": f"待办{i}"} for i in range(8)],
}
rec = normalize_shift_record(raw)
cfg = HandoverBriefConfig(max_events=5, max_todos=3)
props = render_handover_brief(rec, cfg)
self.assertEqual(len(props["events"]), 5)
self.assertEqual(len(props["todos"]), 3)
self.assertEqual(props["eventOverflow"], 7)
self.assertEqual(props["todoOverflow"], 5)
def test_critical_event_forces_confirm(self):
raw = {
"shift": "夜班", "operator": "张工",
"events": [{"time": "01:00", "title": "严重告警", "severity": "P0"}],
}
rec = normalize_shift_record(raw)
cfg = HandoverBriefConfig(require_confirm=False)
props = render_handover_brief(rec, cfg)
# 含 P0 事件 → requireConfirm 被强制为 true(PRD 高利害人工确认)
self.assertTrue(props["requireConfirm"])
def test_safety_text_forces_confirm(self):
raw = {
"shift": "夜班", "operator": "张工",
"safety": "注意 3 层高温区巡检。",
}
rec = normalize_shift_record(raw)
cfg = HandoverBriefConfig(require_confirm=False)
props = render_handover_brief(rec, cfg)
self.assertTrue(props["requireConfirm"])
def test_section_switches(self):
rec = normalize_shift_record(_full_record_dict())
cfg = HandoverBriefConfig(sections=["overview", "abnormal"])
props = render_handover_brief(rec, cfg)
self.assertTrue(props["showOverview"])
self.assertTrue(props["showAbnormal"])
self.assertFalse(props["showSafety"])
self.assertFalse(props["showEnergy"])
self.assertFalse(props["showTodos"])
def test_injected_summary_passes_through(self):
rec = normalize_shift_record({"shift": "夜班", "operator": "张工"})
props = render_handover_brief(rec, summary="自定义摘要")
self.assertEqual(props["summary"], "自定义摘要")
# ---------------------------------------------------------------------------
# 8) 验收线
# ---------------------------------------------------------------------------
class TestGenerationBudget(unittest.TestCase):
def test_within_budget(self):
self.assertTrue(within_generation_budget(60_000)) # 1 分钟
self.assertTrue(within_generation_budget(HANDOVER_GENERATION_BUDGET_MS))
self.assertFalse(within_generation_budget(HANDOVER_GENERATION_BUDGET_MS + 1))
if __name__ == "__main__":
unittest.main(verbosity=2)