diff --git a/packages/report/copy.py b/packages/report/copy.py new file mode 100644 index 0000000..500e625 --- /dev/null +++ b/packages/report/copy.py @@ -0,0 +1,250 @@ +"""报告文案的纯逻辑：LLM 素材组装与降级模板渲染（无 IO、无 LLM 调用）。 + +对应架构 §7.1 步骤 17/18 中可独立单测的两块： + +- 素材组装：把 T16 的结构化诊断裁剪成 prompt 载荷。**skipped 维度不进 + prompt**——LLM 看不到缺失的维度就不会编造它，这是防幻觉的第一道闸 + （第二道闸在 validation.py 的生成后校验）。 +- 降级模板：持续违规时用规则直出拼一份零 LLM 的合规文案，是 T-P3-11 + 「绝不放行违规内容」的最终出口——有降级路径，fail-close 才不会变成 + 「永远失败」。 +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from typing import Any + +from pydantic import BaseModel + +from packages.shared.errors import AppError, ErrorCode + +#: 素材总长上限（字符）。REPORT_COPY 场景 max_tokens=3000，输入侧按架构 +#: §7.2 的裁剪思想控制：低重要度维度先丢，硬结论永不丢。 +MAX_MATERIAL_CHARS = 6000 + +#: 硬结论维度：体/脸/季三大枚举是下游（记忆、Agent、换装推荐）共同依赖的 +#: 画像锚点，任何预算压力下都不裁——宁可 prompt 变长，不可结论缺失。 +HARD_DIMENSION_KEYS: frozenset[str] = frozenset({"body_shape", "face_shape", "color_season"}) + +#: 维度裁剪优先级：数值越大越后裁。未登记的维度取默认优先级，新增维度 +#: 无需改这张表即可获得「先被裁」的保守待遇。 +_DIMENSION_PRIORITY: dict[str, int] = { + "body_shape": 100, + "face_shape": 100, + "color_season": 100, + "skin_tone": 80, + "color_palette": 80, + "style_impression": 60, + "figure_balance": 60, +} +_DEFAULT_PRIORITY = 50 + +#: 外层 JSON 键（{"dimensions":[],"rule_hits":[]}）的固定序列化开销估计。 +_SERIALIZATION_OVERHEAD = 64 + + +class CopySection(BaseModel): + """报告正文章节。title 短语即可，文案主体放 paragraphs。""" + + title: str + paragraphs: list[str] + + +class ReportCopy(BaseModel): + """REPORT_COPY 场景的结构化输出 schema（宽进严出的取舍）。 + + 字段越少，provider 的 JSON 约束成功率越高——解析失败会触发网关的 + 结构化重试，每次都是真金白银；文案的自由度全部放进 sections 的段落 + 文本，不靠 schema 枚举去硬卡。语义正确性不靠 schema：那属于生成后 + 校验（validation.py），两层各管各的。 + """ + + summary: str + sections: list[CopySection] + + def full_text(self) -> str: + """summary + 全部章节标题与段落，供校验器做全文扫描。""" + parts = [self.summary] + for section in self.sections: + parts.append(section.title) + parts.extend(section.paragraphs) + return "\n".join(parts) + + +def coerce_parsed_copy(parsed: Any) -> ReportCopy: + """gateway ``Completion.parsed`` → ReportCopy；类型不符即网关契约破裂。 + + 正常不会走到 raise（gateway 带 schema 调用成功即保证已过模型校验）， + 留这道闸是为了把意外的 TypeError 统一成 AppError，方便编排层按 + 错误码落任务失败与退款。 + """ + if not isinstance(parsed, ReportCopy): + msg = "REPORT_COPY 场景的 completion 未携带 ReportCopy 结构" + raise AppError(ErrorCode.LLM_STRUCTURED_OUTPUT_FAILED, message=msg) + return parsed + + +def _compact_dimension(dim: Mapping[str, Any]) -> dict[str, Any]: + """维度瘦身到 prompt 需要的字段：key/value/evidence + 两位小数置信度。 + + confidence 保留（低置信维度要用不确定表述，模型必须能看到它），只把 + 视觉层带来的浮点尾数舍掉，省字符且避免无意义的精度噪音。 + """ + confidence = dim.get("confidence") + compact: dict[str, Any] = { + "key": dim.get("key"), + "value": dim.get("value"), + "evidence": list(dim.get("evidence") or []), + } + if isinstance(confidence, (int, float)): + compact["confidence"] = round(float(confidence), 2) + return compact + + +def _compact_rule_hits(rule_hits: Sequence[Any]) -> list[dict[str, Any]]: + """规则命中瘦身：保留 key/维度/建议/解释/硬禁忌标记，其余字段不进 prompt。""" + compacted: list[dict[str, Any]] = [] + for hit in rule_hits: + if not isinstance(hit, Mapping): + continue + compact: dict[str, Any] = {"rule_key": hit.get("rule_key") or hit.get("key")} + for field in ("dimension", "suggestion", "explanation", "is_hard"): + if hit.get(field) is not None: + compact[field] = hit[field] + compacted.append(compact) + return compacted + + +def _compact_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +def select_prompt_dimensions( + result_diagnosis: Mapping[str, Any], + *, + max_chars: int = MAX_MATERIAL_CHARS, +) -> tuple[list[dict[str, Any]], list[str]]: + """取 status=ok 的维度并做超长裁剪；返回 (保留维度, 被裁掉的 key 列表)。 + + - skipped/error 维度直接不进素材（防幻觉第一道闸，见模块 docstring）； + - 超长时按优先级升序丢非硬结论维度（低重要度先丢，硬结论永不丢）； + - 若保留的全是硬结论仍超长，不再裁：硬结论缺位的代价远大于长 prompt， + 由 provider 侧 max_tokens 兜底。此时 dropped 可能为空但载荷超长。 + """ + raw_dimensions = [ + d for d in (result_diagnosis.get("dimensions") or []) if isinstance(d, Mapping) + ] + kept = [_compact_dimension(d) for d in raw_dimensions if d.get("status") == "ok"] + hits_len = len(_compact_json(_compact_rule_hits(result_diagnosis.get("rule_hits") or []))) + budget = max(max_chars - hits_len - _SERIALIZATION_OVERHEAD, 0) + + droppable = sorted( + (d for d in kept if str(d.get("key")) not in HARD_DIMENSION_KEYS), + key=lambda d: ( + _DIMENSION_PRIORITY.get(str(d.get("key")), _DEFAULT_PRIORITY), + str(d.get("key")), + ), + ) + dropped: list[str] = [] + while droppable and len(_compact_json(kept)) > budget: + victim = droppable.pop(0) + kept.remove(victim) + dropped.append(str(victim["key"])) + return kept, dropped + + +def compose_prompt_material( + result_diagnosis: Mapping[str, Any], + *, + max_chars: int = MAX_MATERIAL_CHARS, +) -> tuple[str, list[str]]: + """组装 {dimensions_json} 载荷，连同被裁掉的维度 key 一起返回。 + + CopyStage 一次调用拿全两者：载荷进 prompt，dropped_keys 落进 + result["copy"] 供查询侧与排查用（用户看到的报告少了哪些维度）。 + """ + kept, dropped = select_prompt_dimensions(result_diagnosis, max_chars=max_chars) + payload = { + "dimensions": kept, + "rule_hits": _compact_rule_hits(result_diagnosis.get("rule_hits") or []), + } + return _compact_json(payload), dropped + + +def build_prompt_material(result_diagnosis: Mapping[str, Any]) -> str: + """任务契约签名：只返回素材字符串（dropped_keys 用 compose_prompt_material 取）。""" + material, _dropped = compose_prompt_material(result_diagnosis) + return material + + +#: 维度 key → 中文标签。模板文案与校验器共用的最小字典；T16 若在维度里 +#: 自带 label 字段则优先用它的。 +_KEY_LABELS: dict[str, str] = { + "body_shape": "体型", + "face_shape": "脸型", + "color_season": "季型", + "skin_tone": "肤色", + "color_palette": "色板", +} + +_FALLBACK_SUMMARY = ( + "以下内容根据本次分析的结构化结论整理，每条结论都来自规则引擎对诊断值的直接匹配。" +) +_FALLBACK_EMPTY_NOTE = "本次分析暂无可用结论，请稍后重新生成报告。" + + +def dimension_label(dim: Mapping[str, Any]) -> str: + """维度的中文标签：优先维度自带 label，其次按 key 查表，兜底返回 key。""" + label = dim.get("label") + if isinstance(label, str) and label: + return label + key = str(dim.get("key") or "") + return _KEY_LABELS.get(key, key or "维度") + + +def _evidence_text(dim: Mapping[str, Any]) -> str: + evidence = dim.get("evidence") + if isinstance(evidence, Sequence) and not isinstance(evidence, str) and evidence: + return "判定依据：" + "；".join(str(item) for item in evidence) + if isinstance(evidence, str) and evidence: + return f"判定依据：{evidence}" + return "" + + +def render_fallback_copy( + dimensions: Sequence[Any], + rule_hits: Sequence[Any], +) -> ReportCopy: + """降级模板：按维度 value/evidence + 规则 suggestion/explanation 拼文案，零 LLM。 + + 持续违规（重试一次仍不过）时的最终出口。模板只做受控字符串拼接、 + 逐字复述诊断值与规则说明，不含任何自由生成文本，因此天然通过词表/ + 一致性/绝对化三重校验——ValidationStage 仍会对它跑一遍校验做 + fail-close 自证（见 validation_stage.py）。 + """ + ok_dims = [d for d in dimensions if isinstance(d, Mapping) and d.get("status") == "ok"] + + conclusion_paragraphs = [] + for dim in ok_dims: + value = dim.get("value") or "" + para = f"你的{dimension_label(dim)}结论为「{value}」。{_evidence_text(dim)}".rstrip() + conclusion_paragraphs.append(para) + + advice_paragraphs = [] + for hit in rule_hits: + if not isinstance(hit, Mapping): + continue + parts = [str(hit[field]) for field in ("suggestion", "explanation") if hit.get(field)] + if parts: + advice_paragraphs.append("· " + "——".join(parts)) + + sections: list[CopySection] = [] + if conclusion_paragraphs: + sections.append(CopySection(title="诊断结论", paragraphs=conclusion_paragraphs)) + if advice_paragraphs: + sections.append(CopySection(title="搭配建议", paragraphs=advice_paragraphs)) + if not sections: + sections.append(CopySection(title="说明", paragraphs=[_FALLBACK_EMPTY_NOTE])) + return ReportCopy(summary=_FALLBACK_SUMMARY, sections=sections) diff --git a/packages/report/copy_stage.py b/packages/report/copy_stage.py new file mode 100644 index 0000000..886c03a --- /dev/null +++ b/packages/report/copy_stage.py @@ -0,0 +1,63 @@ +"""流水线阶段：LLM 文案生成（架构 §7.1 步骤 17）。 + +只生成、不校验——生成后校验是下一阶段（步骤 18）的独立职责。把两者 +塞进同一个阶段会让「校验不可跳过」失去结构保证：跳过一个阶段是配置 +错误（会被装配层拦下），跳过半段代码不是。 +""" + +from __future__ import annotations + +from packages.llm.types import Scene +from packages.report.copy import ReportCopy, coerce_parsed_copy, compose_prompt_material +from packages.report.pipeline import ReportContext +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + + +class CopyStage: + """name="copy"：调 REPORT_COPY 场景生成结构化文案，产物落 result["copy"]。 + + 产物键：summary / sections / model / prompt_version / dropped_keys。 + dropped_keys 记录因超长被裁出 prompt 的维度——查询侧据此知道报告 + 少了哪些维度，避免「报告里没有」被当成「诊断说没有」。 + """ + + name = "copy" + + async def run(self, ctx: ReportContext) -> None: + diagnosis = ctx.report.result.get("diagnosis") + if not isinstance(diagnosis, dict): + # 没有 diagnosis 产物说明流水线装配顺序错了（copy 排在了 diagnosis 前面） + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message="copy 阶段缺少 diagnosis 产物（流水线阶段顺序错误）", + ) + if ctx.gateway is None: + raise AppError(ErrorCode.CONFIG_ERROR, message="报告流水线缺少 LLM 网关") + + material, dropped_keys = compose_prompt_material(diagnosis) + completion = await ctx.gateway.complete( + scene=Scene.REPORT_COPY, + prompt_vars={"dimensions_json": material}, + schema=ReportCopy, + user_id=ctx.report.user_id, + ref_type="report", + ref_id=str(ctx.report.report_id), + ) + copy = coerce_parsed_copy(completion.parsed) + + ctx.report.result["copy"] = { + "summary": copy.summary, + "sections": [section.model_dump() for section in copy.sections], + "model": completion.model, + "prompt_version": completion.prompt_version, + "dropped_keys": dropped_keys, + } + logger.info( + "report_copy_generated", + report_id=ctx.report.report_id, + model=completion.model, + dropped_dimensions=len(dropped_keys), + ) diff --git a/packages/report/validation.py b/packages/report/validation.py new file mode 100644 index 0000000..681648c --- /dev/null +++ b/packages/report/validation.py @@ -0,0 +1,352 @@ +"""生成后校验的纯逻辑（架构 §7.1 步骤 18，T-P3-10 / T-P3-11 🔴）。 + +prompt 红线管「生成前」，本模块管「生成后」：LLM 输出永远不可信任， +三重校验（贬损词表 / 文案-结论一致性 / 绝对化用语）全过才放行， +ValidationStage 据此决定重试与降级。与 gateway 的结构化重试是两层独立 +机制：那层管 JSON 解析合不合法，这层管内容合不合规，互不共享计数—— +解析成功但内容违规，网关不会替我们发现。 +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any + +from packages.report.copy import ReportCopy, dimension_label + +#: 贬损词表（初版内嵌版，正式版归合规词库任务 T-P0-07，届时改为读库+后台可维护）。 +#: 来源：《合规与安全方案》§3「内容伦理红线」——把「描述而非评判」「不使用 +#: 缺陷框架」翻译成具体词表。命中按子串匹配：「有点胖」「微胖」都命中「胖」， +#: 不做词干（中文无明显词干，初版宁多命中不放过）。 +PEJORATIVE_TERMS: tuple[str, ...] = ( + # 身材/外貌贬损（§3 反例：「你的脸型有点大，不好看」「腿短是你的缺点」） + "胖", + "肥", + "臃肿", + "丑", + "难看", + "土气", + "显老", + "老气", + "矮", + "腿短", + "短腿", + "水桶腰", + "大饼脸", + "虎背熊腰", + # 缺陷框架（§3：不使用缺陷框架——特征是比例事实，不是缺点） + "缺陷", + "缺点", + "瑕疵", + "短板", + "硬伤", + "失败", + "悲剧", + "灾难", + # 评判性观感 + "廉价", + "地摊货", + "邋遢", + "呆板", + "死板", + "笨拙", + # 医疗/身材干预越界（§3：不做医疗判断 → 一律转向穿搭与色彩建议） + "减肥", + "减重", + "瘦身", + "整形", + "整容", + "就医", + "皮肤病", +) + +#: 绝对化用语（广告法红线；system prompt 已禁，这里是兜底——prompt 不是 +#: 约束保证，只是愿望）。「最」不取单字：「最近/最终」等中性词误杀率高， +#: 取营销语境的高频复合词，误报代价见 check_unsupported_advice 的取舍注释。 +ABSOLUTE_CLAIM_TERMS: tuple[str, ...] = ( + "最好", + "最佳", + "最优", + "最合适", + "最适合", + "最完美", + "最显瘦", + "第一", + "唯一", + "彻底", + "百分百", + "百分之百", + "绝对", + "完美", + "永不过时", + "无可替代", + "独一无二", +) + +#: 校验类别 → 展平违规清单时的人读前缀。 +CHECK_PREFIXES: dict[str, str] = { + "pejorative": "贬损词", + "consistency": "结论不一致", + "absolute": "绝对化用语", +} + +_HEX_COLOR_RE = re.compile(r"#[0-9a-fA-F]{6}\b") +_ADVICE_MARKERS = ("建议", "推荐") + + +@dataclass(frozen=True, slots=True) +class EnumCategory: + """体型/脸型/季型三类枚举维度：key 识别串 + 全量枚举词 + 同义别名。 + + 一致性规则①的依据：这三类的结论词是全量枚举（需求文档 §4 维度字典）， + 文案里出现「同类的另一个枚举值」即可断定与诊断矛盾。 + """ + + label: str + key_tokens: tuple[str, ...] + vocabulary: tuple[str, ...] + #: 诊断值 → 同义说法（诊断说 X 型、文案写沙漏型不算矛盾）。 + aliases: Mapping[str, tuple[str, ...]] = field(default_factory=dict) + + +_ENUM_CATEGORIES: tuple[EnumCategory, ...] = ( + EnumCategory( + label="体型", + key_tokens=("body_shape", "体型"), + vocabulary=( + "A型", + "H型", + "O型", + "T型", + "X型", + "苹果型", + "梨型", + "沙漏型", + "倒三角型", + "矩形型", + ), + aliases={ + "O型": ("苹果型",), + "A型": ("梨型",), + "X型": ("沙漏型",), + "T型": ("倒三角型",), + "H型": ("矩形型", "直筒型"), + }, + ), + EnumCategory( + label="脸型", + key_tokens=("face_shape", "脸型"), + # 只收「X脸」复合词：单字（圆/方/长）会把「轮廓偏圆润」这类合规 + # 描述性用语（合规 §3 的正例）误杀。 + vocabulary=( + "鹅蛋脸", + "圆形脸", + "圆脸", + "方形脸", + "方脸", + "长形脸", + "长脸", + "心形脸", + "菱形脸", + "梨形脸", + ), + aliases={ + "鹅蛋": ("鹅蛋脸", "鹅蛋型"), + "圆": ("圆脸", "圆形脸"), + "方": ("方脸", "方形脸"), + "长": ("长脸", "长形脸"), + "心形": ("心形脸",), + "菱形": ("菱形脸",), + "梨形": ("梨形脸",), + }, + ), + EnumCategory( + label="季型", + key_tokens=("color_season", "season", "季型"), + # 十二季细分是自研描述性命名（TD-18），无法全量枚举，只收四季 + # 粗分词；细分红利是值本身进 allowed 集合，复述永远合法。 + vocabulary=("春季型", "夏季型", "秋季型", "冬季型", "春型", "夏型", "秋型", "冬型"), + aliases={ + "春": ("春型", "春季型"), + "夏": ("夏型", "夏季型"), + "秋": ("秋型", "秋季型"), + "冬": ("冬型", "冬季型"), + }, + ), +) + + +def _normalize_for_match(text: str) -> str: + """去空白 + 拉丁字母大写：「x 型」「X型」归一成同一写法再比较。""" + return re.sub(r"\s+", "", text).upper() + + +def check_pejorative(text: str) -> list[str]: + """全文扫描贬损词表，返回命中的词（按词表顺序）。""" + return [term for term in PEJORATIVE_TERMS if term in text] + + +def check_absolute_claims(text: str) -> list[str]: + """绝对化用语兜底检查，返回命中的词。""" + return [term for term in ABSOLUTE_CLAIM_TERMS if term in text] + + +def _find_dimension( + dimensions: Sequence[Mapping[str, Any]], key_tokens: Sequence[str] +) -> Mapping[str, Any] | None: + for dim in dimensions: + key = str(dim.get("key") or "") + if any(token in key for token in key_tokens): + return dim + return None + + +def _check_enum_consistency( + copy_text_normalized: str, dimensions: Sequence[Mapping[str, Any]] +) -> list[str]: + """规则①：体型/脸型/季型枚举词必须与诊断值同款（含同义别名）。 + + 维度 skipped/无值时，文案出现该类任何枚举词都算编造——LLM 没看到 + 这个维度却写出了结论，正是素材组装要防的幻觉从校验侧兜底。 + 每类只报第一条命中，避免同一问题刷屏冲淡修正指令。 + """ + issues: list[str] = [] + for category in _ENUM_CATEGORIES: + dim = _find_dimension(dimensions, category.key_tokens) + if dim is None: + continue + value = dim.get("value") + if dim.get("status") == "ok" and value: + canonical = str(value) + allowed = {_normalize_for_match(canonical)} + for alias in category.aliases.get(canonical, ()): # 值本身可能就是别名键 + allowed.add(_normalize_for_match(alias)) + # 诊断值也可能是别名列里的写法（如存了「沙漏型」），反查一次 + for canonical_key, alias_values in category.aliases.items(): + if _normalize_for_match(canonical) == _normalize_for_match(canonical_key) or any( + _normalize_for_match(canonical) == _normalize_for_match(a) for a in alias_values + ): + allowed.add(_normalize_for_match(canonical_key)) + allowed.update(_normalize_for_match(a) for a in alias_values) + else: + allowed = set() + for token in category.vocabulary: + normalized_token = _normalize_for_match(token) + if normalized_token in copy_text_normalized and normalized_token not in allowed: + issues.append( + f"{category.label}用词「{token}」与诊断值「{value or '未诊断'}」不一致" + ) + break + return issues + + +def _check_hex_colors(copy_text: str, dimensions: Sequence[Mapping[str, Any]]) -> list[str]: + """规则②：文案中的 #RRGGBB 必须来自诊断维度（色板）的取值集合。 + + 色值集合从全部维度的 value/evidence 里递归收集而非只认某个 key—— + T16 的色板维度 key 未定死，宽收集合只可能放宽校验，不会误杀合规色值。 + """ + used = {m.group(0).upper() for m in _HEX_COLOR_RE.finditer(copy_text)} + if not used: + return [] + dimension_blob = str([dict(d) for d in dimensions]) + allowed = {m.group(0).upper() for m in _HEX_COLOR_RE.finditer(dimension_blob)} + return [f"色值「{color}」不在诊断给出的色板取值范围内" for color in sorted(used - allowed)] + + +def _check_unsupported_advice( + copy_text: str, + dimensions: Sequence[Mapping[str, Any]], + rule_hits: Sequence[Mapping[str, Any]] | None, +) -> list[str]: + """规则③（宽松口径）：建议/推荐句式必须有规则命中支撑。 + + 取舍：误报的代价是一次无谓重试甚至降级到模板（文案质量明显变差、 + 多花一次 LLM 调用），漏报的底线由贬损词表与枚举一致性兜住，因此只 + 标记两种可确定的情形： + ① 文案给出建议但规则引擎零命中（建议完全无据）； + ② 建议句点名了某维度，而该维度既无内嵌 rule_hits、也无顶层命中指向它。 + 其余（句式识别不全、支撑关系模糊）一律放行——宁漏勿误杀。 + """ + sentences = re.split(r"[。！？!?;\n]", copy_text) + advice_sentences = [s for s in sentences if any(marker in s for marker in _ADVICE_MARKERS)] + if not advice_sentences: + return [] + + top_level_dims = { + str(hit.get("dimension")) for hit in (rule_hits or []) if hit.get("dimension") + } + embedded_exists = any(dim.get("rule_hits") for dim in dimensions) + + if not top_level_dims and not embedded_exists: + return ["文案包含建议/推荐表述，但规则引擎没有任何命中来支撑该建议"] + + issues: list[str] = [] + for sentence in advice_sentences: + snippet = sentence.strip()[:30] + for dim in dimensions: + if dim.get("status") != "ok": + continue + key = str(dim.get("key") or "") + label = dimension_label(dim) + if not label or label not in sentence: + continue + supported = bool(dim.get("rule_hits")) or key in top_level_dims + if not supported: + issues.append(f"建议句「{snippet}」提到{label}，但该维度没有规则命中支撑") + return issues + + +def check_conclusion_consistency( + copy: ReportCopy, + dimensions: list[dict[str, Any]], + *, + rule_hits: Sequence[Mapping[str, Any]] | None = None, +) -> list[str]: + """文案-结论一致性（T-P3-10，初版三条规则）。返回不一致清单。 + + ① 体型/脸型/季型枚举词必须与诊断值同款；维度未诊断时不得出现该类词； + ② 具体色值 #RRGGBB 必须在诊断（色板维度）取值集合内； + ③ 建议/推荐句式必须有规则命中支撑（宽松口径，见 _check_unsupported_advice）。 + + ``rule_hits`` 为可选参数：规则③的支撑判定兼容两种契约形态（命中内嵌 + 在维度里 / 顶层列表），两者都没有时按宽松口径不追责。 + """ + dict_dimensions = [d for d in dimensions if isinstance(d, Mapping)] + full_text = copy.full_text() + issues = _check_enum_consistency(_normalize_for_match(full_text), dict_dimensions) + issues.extend(_check_hex_colors(full_text, dict_dimensions)) + issues.extend(_check_unsupported_advice(full_text, dict_dimensions, rule_hits)) + return issues + + +def collect_violations( + copy: ReportCopy, + dimensions: list[dict[str, Any]], + *, + rule_hits: Sequence[Mapping[str, Any]] | None = None, +) -> dict[str, list[str]]: + """三重校验的汇总视图，键固定为 pejorative / consistency / absolute。""" + full_text = copy.full_text() + return { + "pejorative": check_pejorative(full_text), + "consistency": check_conclusion_consistency(copy, dimensions, rule_hits=rule_hits), + "absolute": check_absolute_claims(full_text), + } + + +def validate_copy( + copy: ReportCopy, + dimensions: list[dict[str, Any]], + *, + rule_hits: Sequence[Mapping[str, Any]] | None = None, +) -> list[str]: + """展平的违规清单，条目带类别前缀——可直接作为重试的修正指令。""" + breakdown = collect_violations(copy, dimensions, rule_hits=rule_hits) + return [ + f"{CHECK_PREFIXES[key]}：{item}" + for key in ("pejorative", "consistency", "absolute") + for item in breakdown[key] + ] diff --git a/packages/report/validation_stage.py b/packages/report/validation_stage.py new file mode 100644 index 0000000..831ab8f --- /dev/null +++ b/packages/report/validation_stage.py @@ -0,0 +1,173 @@ +"""流水线阶段：生成后校验（架构 §7.1 步骤 18，T-P3-10 / T-P3-11 🔴）。 + +出口只有两种：``passed=true``（干净通过 / 重试后通过 / 降级模板）， +或抛错让整条流水线失败。**绝不放行未过校验的文案**——宁可任务失败退款， +也不让一条贬损文案到达用户（T-P3-11 的硬性禁忌违反数必须为 0）。 + +重试语义：这里的「重试一次」是**语义层**修正（把违规清单作为修正指令 +追加进 messages 重新生成），与 gateway 的结构化重试（解析层，JSON 合不 +合法）互不共享计数——解析成功但内容违规，网关不会替我们发现。 +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import ValidationError + +from packages.llm.types import Completion, Message, Scene +from packages.report.copy import ( + ReportCopy, + coerce_parsed_copy, + compose_prompt_material, + render_fallback_copy, +) +from packages.report.pipeline import ReportContext +from packages.report.validation import validate_copy +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +#: 语义层修正的最大重试次数（架构 §7.1：校验失败最多重试后降级）。 +#: 再多就是花钱买降级——持续违规说明模型在该素材上不可救。 +_MAX_SEMANTIC_RETRIES = 1 + +#: 降级模板的 model 标记：查询侧/排查侧凭它知道这份文案没经过 LLM。 +_FALLBACK_MODEL = "fallback-template" + + +def _clean_checks() -> dict[str, int]: + """全过时的 checks 计数（最终放行文案的违规数必然全零）。""" + return {"pejorative": 0, "consistency": 0, "absolute": 0} + + +def _copy_payload( + copy: ReportCopy, base: dict[str, Any], model: str, prompt_version: str | None +) -> dict[str, Any]: + """组装 result["copy"]：文案主体换新，dropped_keys 等诊断类元数据沿用。""" + return { + **base, + "summary": copy.summary, + "sections": [section.model_dump() for section in copy.sections], + "model": model, + "prompt_version": prompt_version, + } + + +class ValidationStage: + """name="validation"：三重校验 → 违规则重试一次 → 仍违规则降级模板。 + + 产物落 result["validation"]：passed 恒为 true（或本阶段抛错）， + degraded=true 表示最终文案是规则直出模板；violations_first_pass + 留存第一轮违规清单，供报告查询侧与质量分析回看降级原因。 + """ + + name = "validation" + + async def run(self, ctx: ReportContext) -> None: + result = ctx.report.result + diagnosis = result.get("diagnosis") + copy_payload = result.get("copy") + if not isinstance(diagnosis, dict) or not isinstance(copy_payload, dict): + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message="validation 阶段缺少 copy/diagnosis 产物（流水线阶段顺序错误）", + ) + try: + copy = ReportCopy.model_validate(copy_payload) + except ValidationError as exc: + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message=f"copy 产物结构不合法：{exc.error_count()} 处校验错误", + ) from exc + + dimensions = [d for d in (diagnosis.get("dimensions") or []) if isinstance(d, dict)] + rule_hits = [h for h in (diagnosis.get("rule_hits") or []) if isinstance(h, dict)] + + violations = validate_copy(copy, dimensions, rule_hits=rule_hits) + if not violations: + result["validation"] = {"passed": True, "checks": _clean_checks()} + return + + logger.warning( + "report_copy_violations_found", + report_id=ctx.report.report_id, + violations=violations, + ) + retry = await self._retry_with_fixes(ctx, diagnosis, violations) + if retry is not None: + retry_copy, retry_completion = retry + retry_violations = validate_copy(retry_copy, dimensions, rule_hits=rule_hits) + if not retry_violations: + result["copy"] = _copy_payload( + retry_copy, + copy_payload, + model=retry_completion.model, + prompt_version=retry_completion.prompt_version, + ) + result["validation"] = { + "passed": True, + "retried": True, + "violations_first_pass": violations, + "checks": _clean_checks(), + } + return + logger.warning( + "report_copy_retry_still_violating", + report_id=ctx.report.report_id, + violations=retry_violations, + ) + + # 降级：规则直出模板，零 LLM。模板由受控拼接生成，理论必过校验； + # 仍不过（词表演进/诊断值自带违规词）则 fail-close：抛错让流水线 + # 失败并退款——这正是「任何出口要么 passed 要么抛错」的含义。 + fallback = render_fallback_copy(dimensions, rule_hits) + residual = validate_copy(fallback, dimensions, rule_hits=rule_hits) + if residual: + raise AppError( + ErrorCode.INTERNAL_ERROR, + message=f"降级模板文案未通过校验，拒绝放行：{residual}", + ) + result["copy"] = _copy_payload( + fallback, copy_payload, model=_FALLBACK_MODEL, prompt_version=None + ) + result["validation"] = { + "passed": True, + "degraded": True, + "violations_first_pass": violations, + "checks": _clean_checks(), + } + logger.warning( + "report_copy_degraded_to_template", + report_id=ctx.report.report_id, + first_pass_violations=len(violations), + ) + + async def _retry_with_fixes( + self, + ctx: ReportContext, + diagnosis: dict[str, Any], + violations: list[str], + ) -> tuple[ReportCopy, Completion] | None: + """带修正指令重新生成一次；网关层故障原样抛出（属基础设施错误， + 不该被降级模板掩盖——那是内容问题出口，不是可用性出口）。""" + if ctx.gateway is None: + raise AppError(ErrorCode.CONFIG_ERROR, message="报告流水线缺少 LLM 网关") + material, _dropped = compose_prompt_material(diagnosis) + correction = ( + "上一版报告文案未通过内容校验，违规清单如下：\n" + + "\n".join(f"- {item}" for item in violations) + + "\n请严格遵守系统提示词的内容红线，重新生成整份文案：只复述给定诊断值，" + "删除全部违规表述，不要解释修改过程。" + ) + completion = await ctx.gateway.complete( + scene=Scene.REPORT_COPY, + messages=[Message(role="user", content=correction)], + prompt_vars={"dimensions_json": material}, + schema=ReportCopy, + user_id=ctx.report.user_id, + ref_type="report", + ref_id=str(ctx.report.report_id), + ) + return coerce_parsed_copy(completion.parsed), completion diff --git a/prompts/report_copy/v1.txt b/prompts/report_copy/v1.txt index df5cacc..fc86b25 100755 --- a/prompts/report_copy/v1.txt +++ b/prompts/report_copy/v1.txt @@ -1,17 +1,20 @@ [system] -你为「美搭」的美学报告撰写文案。输入是结构化诊断结论与规则命中结果，输出是给用户阅读的报告正文。 +你是「美搭」的形象顾问，为用户的美学报告撰写文案。输入是结构化诊断结论与规则命中结果，输出是给用户阅读的报告正文。 -硬性约束： -- 只能使用输入中给出的诊断值与知识条目，不得引入任何未提供的事实。 -- 不得违反输入中标记为硬禁忌的规则，一条都不行。 +内容红线（违反任何一条，整篇文案即作废重写）： +- 结论只能复述输入中给出的诊断值，不得新增、改写或推断任何维度结论；输入未提供的维度一律不提。 +- 描述而非评判：客观陈述特征与搭配路径，不评判外貌。 +- 不出现任何贬损性词汇，例如胖、丑、矮、土气、显老、缺陷一类表述。 +- 不使用绝对化用语，例如最好、第一、彻底、百分百。 - 不做医疗、减重、整形、命理相关的结论或暗示。 -- 不评判外貌，不使用贬损性词汇；描述而非评价。 -- 置信度低的维度必须使用不确定表述，不得断言。 +- 置信度低的维度使用不确定表述，不得断言。 写作要求： -- 简体中文，平实具体，避免营销腔与绝对化用语。 +- 简体中文，平实具体，避免营销腔。 - 每个结论后面跟一句「为什么」，依据来自输入的规则命中说明。 - 专业术语首次出现时补白话解释。 [user] -{payload} +以下是结构化诊断素材（仅含已成功诊断的维度与规则命中结果），请据此生成报告文案： + +{dimensions_json} diff --git a/tests/integration/test_report_copy_stage.py b/tests/integration/test_report_copy_stage.py new file mode 100644 index 0000000..af964b5 --- /dev/null +++ b/tests/integration/test_report_copy_stage.py @@ -0,0 +1,268 @@ +"""报告 copy / validation 两阶段的集成验证（FakeProvider 替换真实 provider）。 + +不依赖真实 PG/Redis（本组用例只读写内存中的 RpReport.result），但按批次 +约定放 integration：这里组装的是真实网关栈（PromptRegistry 读仓库里的 +report_copy/v1.txt、路由、预算、结构化解析），验证「prompt 模板 → +gateway 结构化输出 → 阶段产物 → 语义重试/降级」的完整咬合，纯单测的桩 +盖不住这条链。不测整条流水线编排（T15 已覆盖），只直接构造 ctx 调阶段。 +""" + +from __future__ import annotations + +import itertools +import json +from collections.abc import Sequence +from decimal import Decimal +from pathlib import Path +from typing import Any, cast + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.llm.budget import DailyBudgetGuard, InMemoryCostCounter +from packages.llm.circuit import CircuitBreakerRegistry +from packages.llm.cost import InMemoryAICallRecorder, ModelRate, RateTable +from packages.llm.gateway import LLMGateway +from packages.llm.prompts import PromptRegistry +from packages.llm.routing import SceneRouter, build_scene_configs +from packages.llm.types import ProviderRequest, ProviderResponse, Usage +from packages.report.copy import ReportCopy +from packages.report.copy_stage import CopyStage +from packages.report.pipeline import ReportContext +from packages.report.validation import check_pejorative, validate_copy +from packages.report.validation_stage import ValidationStage +from packages.shared.config import get_settings +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import RpInput, RpReport + +pytestmark = [pytest.mark.integration] + +REPO_PROMPTS = Path(__file__).resolve().parents[2] / "prompts" + + +class QueuedProvider: + """按队列返回预置响应，耗尽后重复最后一个——最后一条即「持续脏」语义。""" + + def __init__(self, outcomes: Sequence[ProviderResponse]) -> None: + self._outcomes = list(outcomes) + self.requests: list[ProviderRequest] = [] + + @property + def name(self) -> str: + return "p1" + + @property + def default_model(self) -> str: + return "fake-model" + + async def complete(self, request: ProviderRequest) -> ProviderResponse: + self.requests.append(request) + return self._outcomes[min(len(self.requests) - 1, len(self._outcomes) - 1)] + + async def aclose(self) -> None: + return None + + +_RATE = ModelRate( + input_cents_per_mtok=Decimal("100"), + cached_input_cents_per_mtok=Decimal("10"), + output_cents_per_mtok=Decimal("200"), +) + + +def _build_gateway(provider: QueuedProvider) -> LLMGateway: + return LLMGateway( + providers={"p1": provider}, + router=SceneRouter(build_scene_configs(primary_provider="p1", backup_provider=None)), + prompts=PromptRegistry(REPO_PROMPTS), + rates=RateTable({"fake-model": _RATE}, default=_RATE), + recorder=InMemoryAICallRecorder(), + budget=DailyBudgetGuard( + InMemoryCostCounter(), daily_budget_cents=Decimal("100000"), enabled=True + ), + breakers=CircuitBreakerRegistry(), + id_factory=itertools.count(1).__next__, + ) + + +def _diagnosis() -> dict[str, Any]: + return { + "dimensions": [ + { + "key": "body_shape", + "status": "ok", + "value": "X型", + "confidence": 0.76, + "evidence": ["shoulder_hip_ratio:1.02", "waist_hip_ratio:0.71"], + "rule_hits": ["R-BODY-0007"], + }, + { + "key": "face_shape", + "status": "ok", + "value": "鹅蛋", + "confidence": 0.81, + "evidence": ["face_ratio:1.35"], + "rule_hits": [], + }, + { + "key": "color_season", + "status": "ok", + "value": "浅暖柔和型", + "confidence": 0.82, + "evidence": ["contrast:low"], + "rule_hits": ["R-COLOR-0021"], + }, + { + # skipped 维度：不应出现在任何一次请求的素材里（防幻觉得到结构保证） + "key": "posture_check", + "status": "skipped", + "value": "", + "confidence": 0.0, + "evidence": [], + "rule_hits": [], + }, + ], + "rule_hits": [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型突出腰线", + "explanation": "X 型体型肩腰臀比例均衡，收腰设计能强化这一优势", + }, + { + "rule_key": "R-COLOR-0021", + "dimension": "color_season", + "suggestion": "主色选择低饱和暖调", + "explanation": "浅暖柔和型对比弱，低饱和暖调更显匀净", + }, + ], + } + + +CLEAN_COPY: dict[str, Any] = { + "summary": "本次分析完成：体型、脸型与色彩结论均来自照片与问卷的结构化测算。", + "sections": [ + { + "title": "体型与比例", + "paragraphs": [ + "你的体型结论为 X 型，肩腰臀比例均衡。", + "建议优先选择收腰版型，突出腰部线条。", + ], + }, + {"title": "脸型与领线", "paragraphs": ["脸型轮廓流畅，接近鹅蛋脸，多数领型都能协调。"]}, + {"title": "色彩方向", "paragraphs": ["色彩结论为浅暖柔和型，建议主色选择低饱和暖调。"]}, + ], +} + +DIRTY_COPY: dict[str, Any] = { + "summary": "整体看身材偏微胖，需要靠版型修饰，收腰款最合适。", + "sections": [ + {"title": "体型小结", "paragraphs": ["你的体型属于 X 型，轮廓数据完整。"]}, + ], +} + + +def _response(payload: dict[str, Any]) -> ProviderResponse: + return ProviderResponse( + text=json.dumps(payload, ensure_ascii=False), + model="fake-model", + usage=Usage(tokens_in=1200, tokens_out=400, tokens_cached=900), + ) + + +def _make_ctx(gateway: LLMGateway, *, result: dict[str, Any] | None = None) -> ReportContext: + report = RpReport( + report_id=910001, + user_id=42, + result=result if result is not None else {"diagnosis": _diagnosis()}, + ) + report_input = RpInput(input_id=910002, report_id=910001) + return ReportContext( + report=report, + report_input=report_input, + # 本组用例的阶段只读写内存中的 result，不触库；session 仅为满足契约形状 + session=cast("AsyncSession", None), + settings=get_settings(), + gateway=gateway, + ) + + +async def test_clean_llm_copy_passes_both_stages() -> None: + provider = QueuedProvider([_response(CLEAN_COPY)]) + ctx = _make_ctx(_build_gateway(provider)) + + await CopyStage().run(ctx) + await ValidationStage().run(ctx) + + copy_payload = ctx.report.result["copy"] + assert copy_payload["model"] == "fake-model" + assert copy_payload["prompt_version"] == "v1" + assert copy_payload["dropped_keys"] == [] + assert ctx.report.result["validation"] == { + "passed": True, + "checks": {"pejorative": 0, "consistency": 0, "absolute": 0}, + } + assert len(provider.requests) == 1 # 干净文案不需要任何重试 + # skipped 维度不出现在发给 LLM 的任何消息里（防幻觉第一道闸） + all_text = "".join(message.content for message in provider.requests[0].messages) + assert "posture_check" not in all_text + assert "X型" in all_text + + +async def test_dirty_once_copy_passes_after_one_semantic_retry() -> None: + provider = QueuedProvider([_response(DIRTY_COPY), _response(CLEAN_COPY)]) + ctx = _make_ctx(_build_gateway(provider)) + + await CopyStage().run(ctx) + await ValidationStage().run(ctx) + + validation = ctx.report.result["validation"] + assert validation["passed"] is True + assert validation["retried"] is True + assert "degraded" not in validation + assert validation["violations_first_pass"] # 第一轮违规清单非空（微胖/最合适） + # 重试通过后放行的是修正版文案 + assert ctx.report.result["copy"]["summary"] == CLEAN_COPY["summary"] + # 恰好两次调用：初稿 + 一次语义修正（与网关解析层重试互不叠加） + assert len(provider.requests) == 2 + retry_text = "".join(message.content for message in provider.requests[1].messages) + assert "贬损词" in retry_text # 修正指令带上了违规清单 + + +async def test_persistently_dirty_llm_degrades_to_zero_pejorative_copy() -> None: + provider = QueuedProvider([_response(DIRTY_COPY)]) # 恒返脏文案 + ctx = _make_ctx(_build_gateway(provider)) + + await CopyStage().run(ctx) + await ValidationStage().run(ctx) + + validation = ctx.report.result["validation"] + assert validation["passed"] is True + assert validation["degraded"] is True + assert validation["violations_first_pass"] + copy_payload = ctx.report.result["copy"] + assert copy_payload["model"] == "fallback-template" + assert copy_payload["prompt_version"] is None + # 黄金断言（T-P3-11）：最终触达用户的文案零贬损词、零违规 + final_copy = ReportCopy.model_validate(copy_payload) + assert check_pejorative(final_copy.full_text()) == [] + diagnosis = ctx.report.result["diagnosis"] + assert ( + validate_copy(final_copy, diagnosis["dimensions"], rule_hits=diagnosis["rule_hits"]) == [] + ) + # 初稿 + 一次重试，之后停止烧钱、改走零 LLM 模板 + assert len(provider.requests) == 2 + + +async def test_stages_fail_fast_when_predecessor_output_is_missing() -> None: + gateway = _build_gateway(QueuedProvider([_response(CLEAN_COPY)])) + + ctx = _make_ctx(gateway, result={}) + with pytest.raises(AppError) as copy_exc: + await CopyStage().run(ctx) + assert copy_exc.value.code == ErrorCode.TASK_STATE_INVALID + + ctx_without_copy = _make_ctx(gateway, result={"diagnosis": _diagnosis()}) + with pytest.raises(AppError) as validation_exc: + await ValidationStage().run(ctx_without_copy) + assert validation_exc.value.code == ErrorCode.TASK_STATE_INVALID diff --git a/tests/unit/test_report_copy.py b/tests/unit/test_report_copy.py new file mode 100644 index 0000000..ca36755 --- /dev/null +++ b/tests/unit/test_report_copy.py @@ -0,0 +1,211 @@ +"""报告文案纯逻辑：素材组装（skipped 不进 / 超长裁剪 / 硬结论不裁）与降级模板。""" + +from __future__ import annotations + +import json +from typing import Any + +from packages.report.copy import ( + build_prompt_material, + compose_prompt_material, + render_fallback_copy, +) +from packages.report.validation import validate_copy + + +def _dim( + key: str, + value: str = "X型", + *, + status: str = "ok", + confidence: float = 0.82, + evidence: tuple[str, ...] = ("shoulder_hip_ratio:1.02",), + rule_hits: tuple[str, ...] = ("R-BODY-0007",), +) -> dict[str, Any]: + return { + "key": key, + "status": status, + "value": value, + "confidence": confidence, + "evidence": list(evidence), + "rule_hits": list(rule_hits), + } + + +def _diagnosis( + *dimensions: dict[str, Any], rule_hits: list[dict[str, Any]] | None = None +) -> dict[str, Any]: + return {"dimensions": list(dimensions), "rule_hits": rule_hits or []} + + +# --------------------------------------------------------------------------- +# 素材组装：skipped 不进 prompt（防幻觉第一道闸） +# --------------------------------------------------------------------------- + + +def test_build_prompt_material_excludes_skipped_dimensions() -> None: + diagnosis = _diagnosis( + _dim("body_shape", "X型"), + _dim("face_shape", "鹅蛋"), + _dim("occasion_preference", status="skipped", value=""), + ) + + material = build_prompt_material(diagnosis) + payload = json.loads(material) + + keys = [d["key"] for d in payload["dimensions"]] + assert keys == ["body_shape", "face_shape"] + # skipped 维度连同它的 key/value 都不能出现在载荷字符串里——LLM 看不到 + # 缺失维度就不会编造它 + assert "occasion_preference" not in material + + +def test_build_prompt_material_rounds_confidence() -> None: + diagnosis = _diagnosis(_dim("body_shape", "X型", confidence=0.876543)) + + payload = json.loads(build_prompt_material(diagnosis)) + + assert payload["dimensions"][0]["confidence"] == 0.88 + + +def test_build_prompt_material_keeps_key_value_evidence_and_rule_hits() -> None: + hits = [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型", + "explanation": "X 型体型肩腰臀比例均衡", + "is_hard": False, + "internal_field": "不该进 prompt 的字段", + } + ] + diagnosis = _diagnosis( + _dim("body_shape", "X型", evidence=("ratio:1.02", "ratio:0.71")), rule_hits=hits + ) + + material = build_prompt_material(diagnosis) + payload = json.loads(material) + + assert payload["dimensions"][0] == { + "key": "body_shape", + "value": "X型", + "evidence": ["ratio:1.02", "ratio:0.71"], + "confidence": 0.82, + } + assert payload["rule_hits"] == [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型", + "explanation": "X 型体型肩腰臀比例均衡", + "is_hard": False, + } + ] + assert "internal_field" not in material + + +# --------------------------------------------------------------------------- +# 素材组装：超长裁剪 +# --------------------------------------------------------------------------- + + +def test_overlong_material_drops_low_priority_dimensions_and_records_keys() -> None: + big_evidence = ("x" * 200,) * 3 # 每个维度约 600+ 字符 + diagnosis = _diagnosis( + _dim("body_shape", "X型", evidence=big_evidence), # 硬结论，优先级 100 + _dim("style_impression", evidence=big_evidence), # 优先级 60 + _dim("occasion_preference", evidence=big_evidence), # 未登记 → 默认优先级 50 + ) + + material, dropped = compose_prompt_material(diagnosis, max_chars=1000) + + # 默认优先级先丢，其次 style_impression；硬结论保留 + assert dropped == ["occasion_preference", "style_impression"] + payload = json.loads(material) + assert [d["key"] for d in payload["dimensions"]] == ["body_shape"] + assert len(material) <= 1000 + + +def test_hard_dimensions_are_never_dropped_even_when_over_budget() -> None: + huge_evidence = ("x" * 400,) * 5 + diagnosis = _diagnosis( + _dim("body_shape", "X型", evidence=huge_evidence), + _dim("face_shape", "鹅蛋", evidence=huge_evidence), + _dim("color_season", "浅暖柔和型", evidence=huge_evidence), + ) + + material, dropped = compose_prompt_material(diagnosis, max_chars=200) + + assert dropped == [] + payload = json.loads(material) + assert [d["key"] for d in payload["dimensions"]] == ["body_shape", "face_shape", "color_season"] + + +# --------------------------------------------------------------------------- +# 降级模板 +# --------------------------------------------------------------------------- + + +def test_render_fallback_copy_includes_values_evidence_and_rule_explanations() -> None: + dimensions = [ + _dim( + "body_shape", "X型", evidence=("shoulder_hip_ratio:1.02",), rule_hits=("R-BODY-0007",) + ), + _dim("color_season", "浅暖柔和型", evidence=("contrast:low",), rule_hits=("R-COLOR-0021",)), + ] + rule_hits = [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型突出腰线", + "explanation": "X 型体型肩腰臀比例均衡，收腰设计能强化这一优势", + } + ] + + fallback = render_fallback_copy(dimensions, rule_hits) + + titles = [s.title for s in fallback.sections] + assert titles == ["诊断结论", "搭配建议"] + conclusion = "\n".join(fallback.sections[0].paragraphs) + assert "「X型」" in conclusion and "shoulder_hip_ratio:1.02" in conclusion + assert "「浅暖柔和型」" in conclusion + advice = "\n".join(fallback.sections[1].paragraphs) + assert "优先选择收腰版型突出腰线" in advice + # 有 rule_hits 时必须含 explanation——解释链是模板文案的最低质量线 + assert "收腰设计能强化这一优势" in advice + + +def test_render_fallback_copy_without_hits_has_no_advice_section() -> None: + dimensions = [_dim("body_shape", "X型")] + fallback = render_fallback_copy(dimensions, []) + + assert [s.title for s in fallback.sections] == ["诊断结论"] + + +def test_render_fallback_copy_with_no_usable_input_explains_itself() -> None: + fallback = render_fallback_copy([_dim("occasion_preference", status="skipped", value="")], []) + + assert [s.title for s in fallback.sections] == ["说明"] + assert fallback.sections[0].paragraphs + + +def test_fallback_copy_passes_the_full_validation_suite() -> None: + """降级模板必须天然通过三重校验——降级路径的黄金性质在纯函数级先锁一遍。""" + dimensions = [ + _dim( + "body_shape", "X型", evidence=("shoulder_hip_ratio:1.02",), rule_hits=("R-BODY-0007",) + ), + _dim("face_shape", "鹅蛋", evidence=("face_ratio:1.35",)), + ] + rule_hits = [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型突出腰线", + "explanation": "X 型体型肩腰臀比例均衡", + } + ] + + fallback = render_fallback_copy(dimensions, rule_hits) + + assert validate_copy(fallback, dimensions, rule_hits=rule_hits) == [] diff --git a/tests/unit/test_report_validation.py b/tests/unit/test_report_validation.py new file mode 100644 index 0000000..ac14133 --- /dev/null +++ b/tests/unit/test_report_validation.py @@ -0,0 +1,211 @@ +"""生成后校验纯逻辑：贬损词表（含变形）、一致性三条规则、绝对化用语、降级路径。""" + +from __future__ import annotations + +from typing import Any + +from packages.report.copy import CopySection, ReportCopy, render_fallback_copy +from packages.report.validation import ( + check_absolute_claims, + check_conclusion_consistency, + check_pejorative, + validate_copy, +) + + +def _copy(summary: str = "", *paragraphs: str) -> ReportCopy: + return ReportCopy( + summary=summary, + sections=[CopySection(title="正文", paragraphs=list(paragraphs))], + ) + + +def _dim( + key: str, + value: str = "", + *, + status: str = "ok", + rule_hits: tuple[str, ...] = (), + evidence: tuple[str, ...] = (), +) -> dict[str, Any]: + return { + "key": key, + "status": status, + "value": value, + "confidence": 0.82, + "evidence": list(evidence), + "rule_hits": list(rule_hits), + } + + +# --------------------------------------------------------------------------- +# 贬损词表 +# --------------------------------------------------------------------------- + + +def test_pejorative_hits_variants_without_stemming() -> None: + # 「有点胖」「微胖」都命中「胖」——子串匹配即变形处理，不做词干 + assert check_pejorative("整体轮廓有点胖，建议宽松款") == ["胖"] + assert check_pejorative("微胖体型适合直筒剪裁") == ["胖"] + + +def test_pejorative_returns_every_matched_term_in_table_order() -> None: + assert check_pejorative("配色显老又土气，显得臃肿") == ["臃肿", "土气", "显老"] + + +def test_pejorative_clean_text_passes() -> None: + assert check_pejorative("肩腰臀比例均衡，收腰设计能强化这一优势。") == [] + + +def test_pejorative_terms_count_stays_within_initial_band() -> None: + """初版词表约定 20~40 词：太少护不住红线，太多误杀（正式版归 T-P0-07）。""" + from packages.report.validation import PEJORATIVE_TERMS + + assert 20 <= len(PEJORATIVE_TERMS) <= 40 + + +# --------------------------------------------------------------------------- +# 一致性规则①：体型/脸型/季型枚举词 +# --------------------------------------------------------------------------- + + +def test_enum_word_mismatching_diagnosis_is_flagged() -> None: + dimensions = [_dim("body_shape", "X型")] + copy = _copy("你的整体印象", "你的体型属于 A 型，肩线舒展。") + + issues = check_conclusion_consistency(copy, dimensions) + + assert any("体型" in issue and "A型" in issue for issue in issues) + + +def test_enum_word_matching_diagnosis_or_its_alias_passes() -> None: + dimensions = [_dim("body_shape", "X型")] + same_value = _copy("你的整体印象", "你的体型结论为 X 型，比例均衡。") + alias = _copy("你的整体印象", "接近沙漏型的轮廓，腰线明显。") + + assert check_conclusion_consistency(same_value, dimensions) == [] + assert check_conclusion_consistency(alias, dimensions) == [] + + +def test_enum_word_for_skipped_dimension_is_treated_as_fabrication() -> None: + dimensions = [_dim("face_shape", status="skipped")] + copy = _copy("你的整体印象", "你的圆脸轮廓柔和，适合纵向线条。") + + issues = check_conclusion_consistency(copy, dimensions) + + assert any("脸型" in issue for issue in issues) + + +def test_season_enum_mismatch_is_flagged() -> None: + dimensions = [_dim("color_season", "浅暖柔和型")] + copy = _copy("色彩结论", "你属于典型的夏季型，适合冷调色彩。") + + issues = check_conclusion_consistency(copy, dimensions) + + assert any("季型" in issue for issue in issues) + + +# --------------------------------------------------------------------------- +# 一致性规则②：色值必须在色板集合内 +# --------------------------------------------------------------------------- + + +def test_hex_color_outside_palette_is_flagged() -> None: + dimensions = [_dim("color_palette", "#A52A2A", evidence=("#F5F5DC",))] + inside = _copy("色彩方案", "主色可用 #A52A2A，辅色 #f5f5dc 提亮。") + outside = _copy("色彩方案", "主色建议 #AABBCC。") + + assert check_conclusion_consistency(inside, dimensions) == [] + issues = check_conclusion_consistency(outside, dimensions) + assert any("#AABBCC" in issue for issue in issues) + + +# --------------------------------------------------------------------------- +# 一致性规则③：建议句式必须有规则命中支撑 +# --------------------------------------------------------------------------- + + +def test_advice_with_no_rule_hits_at_all_is_flagged() -> None: + dimensions = [_dim("body_shape", "X型", rule_hits=())] + copy = _copy("总体", "建议优先选择收腰版型，突出腰部线条。") + + issues = check_conclusion_consistency(copy, dimensions) + + assert any("没有任何命中" in issue for issue in issues) + + +def test_advice_on_dimension_without_support_is_flagged() -> None: + dimensions = [ + _dim("body_shape", "X型", rule_hits=("R-BODY-0007",)), + _dim("color_season", "浅暖柔和型", rule_hits=()), + ] + copy = _copy("总体", "关于季型，建议主色选择高饱和冷调。") + + issues = check_conclusion_consistency(copy, dimensions) + + assert any("季型" in issue and "没有规则命中支撑" in issue for issue in issues) + + +def test_advice_supported_by_rule_hits_passes() -> None: + dimensions = [_dim("body_shape", "X型", rule_hits=("R-BODY-0007",))] + rule_hits = [{"rule_key": "R-BODY-0007", "dimension": "body_shape", "suggestion": "收腰"}] + copy = _copy("总体", "关于体型，建议优先选择收腰版型，突出腰部线条。") + + assert check_conclusion_consistency(copy, dimensions, rule_hits=rule_hits) == [] + + +# --------------------------------------------------------------------------- +# 绝对化用语 +# --------------------------------------------------------------------------- + + +def test_absolute_claims_are_flagged() -> None: + assert "最合适" in check_absolute_claims("这款收腰连衣裙最合适不过了") + assert "第一" in check_absolute_claims("你是本周第一位、也是第一优选用户") + assert "百分百" in check_absolute_claims("百分百提升气质") + + +def test_absolute_claims_tolerate_neutral_words_with_the_same_head() -> None: + # 「最」单字不拦：最近/最终是中性词，拦了会把正常文案打进重试 + assert check_absolute_claims("根据最近一次分析，你的体型结论保持稳定。") == [] + + +# --------------------------------------------------------------------------- +# 降级路径的纯函数级验证：脏文案违规 → 模板文案零违规 +# --------------------------------------------------------------------------- + + +def test_validate_copy_aggregates_all_three_check_categories() -> None: + dimensions = [_dim("body_shape", "X型")] + dirty = _copy("小结", "整体偏微胖，收腰款最合适。", "你的体型属于 A 型。") + + violations = validate_copy(dirty, dimensions) + + assert any(v.startswith("贬损词：") for v in violations) + assert any(v.startswith("结论不一致：") for v in violations) + assert any(v.startswith("绝对化用语：") for v in violations) + + +def test_degraded_fallback_is_clean_when_llm_copy_stays_dirty() -> None: + """重试后仍违规 → 降级模板：最终文案零违规（T-P3-11 黄金断言的纯函数版）。""" + dimensions = [ + _dim( + "body_shape", "X型", evidence=("shoulder_hip_ratio:1.02",), rule_hits=("R-BODY-0007",) + ), + _dim("face_shape", "鹅蛋", evidence=("face_ratio:1.35",)), + ] + rule_hits = [ + { + "rule_key": "R-BODY-0007", + "dimension": "body_shape", + "suggestion": "优先选择收腰版型突出腰线", + "explanation": "X 型体型肩腰臀比例均衡，收腰设计能强化这一优势", + } + ] + dirty = _copy("小结", "整体偏微胖，收腰款最合适。") + assert validate_copy(dirty, dimensions, rule_hits=rule_hits) != [] + + fallback = render_fallback_copy(dimensions, rule_hits) + + assert validate_copy(fallback, dimensions, rule_hits=rule_hits) == [] + assert check_pejorative(fallback.full_text()) == []