"""报告文案的纯逻辑：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)