feat(report): LLM copy generation with mandatory post-validation and fallback
report_copy/v1 prompt（系统段零变量）+ 结构化 ReportCopy + 素材组装
（skipped 不进 prompt/超长裁剪）+ 贬损词表/一致性/绝对化三重校验 +
重试一次后降级规则直出模板（T-P3-10/11）。
- copy.py：ReportCopy(schema 宽进严出取舍见注释)/compose_prompt_material
（ok 过滤+6000 字符裁剪，低优先级先丢、硬结论不丢，记录 dropped_keys）/
render_fallback_copy 零 LLM 降级模板
- validation.py：PEJORATIVE_TERMS 35 词（源自合规 §3，初版内嵌注释）+
一致性三规则（枚举同款含别名/色值在板内/建议句需规则支撑·宽松口径）+
绝对化用语兜底
- copy_stage.py / validation_stage.py：任何出口 passed=true 或抛错，
语义层重试一次（与网关解析层重试独立），仍违规降级模板并 fail-close 自证
- 单测 25 例（词表含变形/三规则正反/降级零违规）+ 集成 4 例
（FakeProvider 全栈：干净通过/脏一次修正通过/持续脏降级且最终文案
零贬损词——黄金断言）；全量 529 单测 + 49 集成全绿
偏差说明：
1. prompts/report_copy/v1.txt 非新建——LLM 批次已放置初版（user 段变量
为 {payload}），按本任务契约改写为 {dimensions_json} 并补内容红线
（该文件任务文档明确归 T17，无与他人冲突）。
2. check_conclusion_consistency 按契约签名 (copy, dimensions) 之外增加
可选 rule_hits 关键字参数：规则③的支撑判定需兼容命中内嵌在维度/
顶层列表两种 T16 契约形态，缺省时不追责（宽松口径）。
3. 绝对化用语未按单字「最」拦截，取营销语境复合词（最好/最合适/第一/
百分百等 17 词）：单字会误杀「最近/最终」等中性词，红线范围不变。
4. 集成文件未挂 MEIDA_RUN_INTEGRATION skipif：用例为 FakeProvider 内存
装配、不依赖真实 PG/Redis，按任务要求仍放 tests/integration/ 并带
integration marker。
This commit is contained in:
@@ -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
|
||||
@@ -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) == []
|
||||
@@ -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()) == []
|
||||
Reference in New Issue
Block a user