From bc3284856ede815fca2edc19419353c43e6fde54 Mon Sep 17 00:00:00 2001 From: fxbin Date: Mon, 12 Oct 2026 23:48:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(scoring):=20=E5=85=AD=E7=BB=B4=E8=AF=84?= =?UTF-8?q?=E5=88=86=E5=8A=A0=E5=9B=9B=E6=A1=A3=E8=A1=8C=E4=B8=BA=E9=94=9A?= =?UTF-8?q?=E7=82=B9,=E4=B8=AD=E8=8B=B1=20prompt=20=E5=90=8C=E6=A0=87?= =?UTF-8?q?=E5=B0=BA=E6=B8=B2=E6=9F=93=E5=B9=B6=E9=99=84=E5=AF=B9=E7=85=A7?= =?UTF-8?q?=E8=B7=91=E5=88=86=E5=B7=A5=E5=85=B7=20(#114)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scores 块六维此前只有一句名称提示,LLM 打分跨样本跨语种不可比——可解释 打分的前提是同一个 85 分在任何一天意味着同一件事。scoring_caliber 新增 ANCHOR_BANDS 与中英两份锚点表(每维四档可判定行为描述),中/英 prompt 从 它渲染锚点段落,论文版按裁决不并入首期;漂移测试扩展为锚点逐字一致、 档位逐维提取比对、论文版排除断言。附 scripts/scoring_anchor_ab.py 对照 跑分工具:剥离锚点段可逐字节还原 main 版 prompt(独立复核已验证),全量 启用前先跑对照确认分布迁移幅度。 --- backend/app/services/llm/prompts/analysis.py | 9 + backend/app/services/scoring_caliber.py | 107 +++++++ backend/scripts/scoring_anchor_ab.py | 278 +++++++++++++++++++ backend/tests/test_scoring_anchor_ab.py | 86 ++++++ backend/tests/test_scoring_caliber.py | 50 ++++ 5 files changed, 530 insertions(+) create mode 100644 backend/scripts/scoring_anchor_ab.py create mode 100644 backend/tests/test_scoring_anchor_ab.py diff --git a/backend/app/services/llm/prompts/analysis.py b/backend/app/services/llm/prompts/analysis.py index 884e9664..6997a101 100644 --- a/backend/app/services/llm/prompts/analysis.py +++ b/backend/app/services/llm/prompts/analysis.py @@ -11,6 +11,7 @@ from app.services.scoring_caliber import ( curation_threshold_text, curation_weights_text, + dimension_anchor_section, risk_penalty_text, ) @@ -60,6 +61,10 @@ "title_suggestions": [""] }} +""" + + dimension_anchor_section() + + """ + 精选分（curation_score）评判标准： - ≥80：重大发布/独家/强实用性工具/高传播力事件 - 70-79：扎实的产品更新/行业动态/有价值教程 @@ -128,6 +133,10 @@ "title_suggestions": [""] }} +""" + + dimension_anchor_section("en") + + """ + Curation score (curation_score) guidelines: - ≥80: major release / exclusive / highly actionable tool / high-spread event - 70-79: solid product update / industry development / valuable tutorial diff --git a/backend/app/services/scoring_caliber.py b/backend/app/services/scoring_caliber.py index c03bdc83..ae28fe1a 100644 --- a/backend/app/services/scoring_caliber.py +++ b/backend/app/services/scoring_caliber.py @@ -115,3 +115,110 @@ def risk_penalty_text(lang: str = "zh") -> str: if lang == "en": return f"risk>{LLM_RISK_PENALTY_TRIGGER} deducts {LLM_RISK_PENALTY_POINTS} points" return f"风险分>{LLM_RISK_PENALTY_TRIGGER}则扣{LLM_RISK_PENALTY_POINTS}分" + + +# ── 六维评分行为锚点(#114,C3 试点) ──────────────────────────────────── +# scores 块六维从「一句名称提示」升级为四档可判定行为描述,让打分跨样本、 +# 跨语种可比:同一个 85 分在任何一天意味着同一件事。每档写「可判定的行为 +# 特征」而非抽象程度词;中文为口径原文,英文为同一标尺译文。 +# 论文版 prompt 是否复用属独立决策(#114),首期只挂中/英两套。 +ANCHOR_BANDS: tuple[tuple[int, int], ...] = ((0, 30), (40, 60), (70, 85), (90, 100)) + +DIMENSION_ANCHORS_ZH: dict[str, list[str]] = { + "quality_score": [ + "纯转发、寒暄或情绪宣泄，信息量趋近于零，或逻辑断裂看不出主线", + "有明确观点且能自圆其说，但缺少数据、案例或可验证细节", + "论证完整，带数据/案例/步骤等可验证细节，主线清晰", + "一手深料：独家数据、完整方法论、系统实测或深度复盘", + ], + "hot_score": [ + "无讨论量，发布于冷门角落，看不到互动信号", + "小圈子在讨论（评论数十量级），尚未出圈", + "多个社区或账号在转述，评论区活跃，趋势榜可见", + "刷屏级：热搜头部、多平台同时爆发、大V密集跟进", + ], + "freshness_score": [ + "数月前的旧闻，或已被充分报道的旧信息", + "一周内的动态，或旧话题的新进展", + "24-72小时内的新发布或首发报道", + "独家首发、官宣/泄露后数小时、或当天突发", + ], + "creator_score": [ + "与目标领域无关，或没有可延展的选题空间", + "领域相关可作背景素材，但角度已被写烂", + "能直接展开成一篇内容：角度明确、读者清晰、有差异化空间", + "选题富矿：可系列化、多角度深挖，或正处话题空白期", + ], + "viral_score": [ + "话题封闭，只有当事人与极少数人关心", + "有一定共鸣点，但表达门槛高，难以通俗转述", + "有冲突、反常识或实用钩子，一句话能讲清，易被转发", + "踩中大众情绪点或强实用干货，天然适合标题与二创", + ], + "risk_score": [ + "常规技术/产品/资讯内容，无可预见争议", + "轻度敏感：涉争议人物、营销夸大或立场分歧，需注意措辞", + "高风险：未证实的重大指控、医疗/投资建议类、平台限流领域", + "红线：违法、色情、歧视、谣言等，不建议触碰", + ], +} + +DIMENSION_ANCHORS_EN: dict[str, list[str]] = { + "quality_score": [ + "pure repost, small talk, or venting with near-zero information, or incoherent logic", + "has a clear, self-consistent viewpoint but lacks data, cases, or verifiable detail", + "complete argument with data/cases/steps and verifiable detail, clear throughline", + "first-hand depth: exclusive data, full methodology, systematic hands-on testing, or deep retrospective", + ], + "hot_score": [ + "no discussion, posted in an obscure corner, no engagement signals", + "niche community discussing it (dozens of comments), not yet breaking out", + "multiple communities/accounts relaying it, active comments, visible on trend boards", + "explosive: top of trending, erupting across platforms, heavy influencer pickup", + ], + "freshness_score": [ + "months-old news, or information already widely covered", + "a development within the past week, or a new twist on an old topic", + "new release or first report within 24-72 hours", + "exclusive debut, hours after announcement/leak, or same-day breaking", + ], + "creator_score": [ + "irrelevant to the target domain, or no angle to build on", + "domain-relevant background material, but the angle is worn out", + "directly expandable into a piece: clear angle, clear audience, room to differentiate", + "topic goldmine: series-worthy, multi-angle, or filling a topical blank", + ], + "viral_score": [ + "closed topic, only stakeholders and a handful of people care", + "some resonance, but hard to convey in plain words", + "conflict/counter-intuitive/practical hook, explainable in one sentence, easy to share", + "mass emotional appeal or strong utility, naturally title- and remix-friendly", + ], + "risk_score": [ + "routine tech/product/news content, no foreseeable controversy", + "mildly sensitive: controversial figures, marketing hype, or divisive stances", + "high risk: unverified serious allegations, medical/investment advice, platform-throttled areas", + "red line: illegal, pornographic, discriminatory, or rumor content", + ], +} + + +def dimension_anchor_lines(lang: str = "zh") -> list[str]: + """六维行为锚点行(每维一行,四档以分号分隔),数值与文案同出本模块。""" + table = DIMENSION_ANCHORS_EN if lang == "en" else DIMENSION_ANCHORS_ZH + sep = "; " if lang == "en" else "；" + colon = ": " if lang == "en" else "：" + lines: list[str] = [] + for dim, descriptions in table.items(): + parts = [f"{lo}-{hi}={descriptions[i]}" for i, (lo, hi) in enumerate(ANCHOR_BANDS)] + lines.append(f"- {dim}{colon}" + sep.join(parts)) + return lines + + +def dimension_anchor_section(lang: str = "zh") -> str: + """prompt 的完整锚点段落(标题行 + 六行锚点),中/英共用同一标尺。""" + if lang == "en": + header = "Six-dimension scoring anchors (score by band; interpolate between bands):" + else: + header = "六维评分锚点（按档给分，档间取值按程度插值）：" + return header + "\n" + "\n".join(dimension_anchor_lines(lang)) diff --git a/backend/scripts/scoring_anchor_ab.py b/backend/scripts/scoring_anchor_ab.py new file mode 100644 index 00000000..5eb2c551 --- /dev/null +++ b/backend/scripts/scoring_anchor_ab.py @@ -0,0 +1,278 @@ +"""六维行为锚点对照跑分(#114,C3 试点)。 + +对同一批内容分别用「带锚点段」的当前 prompt 与「去掉锚点段」的旧 prompt +各跑一次 LLM 分析,对比六维分数与 curation_score 的分布迁移,回答三个问题: +锚点让分数整体变高/变低/收敛了吗?跨过精选门槛的条数变化多少?哪一维 +移动最大?全量启用锚点前跑一次,把输出贴回 #114;线上在 admin/ +scoring-dashboard 按 analyzed_at 分界观察分布迁移。 + +用法: + cd backend && uv run python scripts/scoring_anchor_ab.py \ + [--limit 30] [--hours 168] [--concurrency 4] [--out anchor_ab.json] + +需要 DATABASE_URL 与 LLM 配置(backend/.env)。论文(arXiv)内容跳过—— +论文版是否挂锚点是独立决策,不在本次对照范围。 +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +import sys +from datetime import UTC, datetime, timedelta +from statistics import median, quantiles +from typing import Any + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _ROOT not in sys.path: + sys.path.insert(0, _ROOT) + +from sqlalchemy import or_, select # noqa: E402 + +from app.core.database import async_session # noqa: E402 +from app.models.content import ContentItem # noqa: E402 +from app.services.analysis import _select_analysis_prompts # noqa: E402 +from app.services.llm import call_llm_json # noqa: E402 +from app.services.llm.prompts.analysis import ANALYSIS_PROMPT_EN # noqa: E402 +from app.services.scoring_caliber import CONFIG, dimension_anchor_section # noqa: E402 +from app.utils.prompt_safety import sanitize_prompt_input # noqa: E402 + +logger = logging.getLogger("scoring_anchor_ab") + +SIX_DIMS = ( + "quality_score", + "hot_score", + "freshness_score", + "creator_score", + "viral_score", + "risk_score", +) + + +# ── 纯函数(单测覆盖) ──────────────────────────────────────────────── + + +def strip_anchor_section(prompt: str, section: str) -> str: + """从已渲染 prompt 中移除锚点段落及其一组相邻空行,还原挂锚点前的版本。""" + wrapped = f"\n\n{section}\n\n" + if wrapped in prompt: + return prompt.replace(wrapped, "\n\n", 1) + if section in prompt: + return prompt.replace(section, "", 1) + raise ValueError("锚点段落未找到,无法构造对照(旧)prompt") + + +def _num(value: Any) -> float | None: + try: + parsed = float(value) + except (TypeError, ValueError): + return None + return parsed + + +def _stats(values: list[float]) -> dict[str, Any]: + if not values: + return {"n": 0} + if len(values) >= 2: + q = quantiles(values, n=4, method="inclusive") + p25, p75 = round(q[0], 1), round(q[2], 1) + else: + p25 = p75 = round(values[0], 1) + return { + "n": len(values), + "mean": round(sum(values) / len(values), 1), + "p25": p25, + "median": round(median(values), 1), + "p75": p75, + } + + +def summarize_dimensions( + old_rows: list[dict[str, float | None]], + new_rows: list[dict[str, float | None]], +) -> dict[str, Any]: + """按维对比新旧分布;curation_score 额外做配对的门槛跨越统计。""" + out: dict[str, Any] = {} + for key in (*SIX_DIMS, "curation_score"): + old_vals = [r[key] for r in old_rows if r.get(key) is not None] + new_vals = [r[key] for r in new_rows if r.get(key) is not None] + s_old, s_new = _stats(old_vals), _stats(new_vals) + entry: dict[str, Any] = {"old": s_old, "new": s_new} + if s_old.get("n") and s_new.get("n"): + entry["mean_shift"] = round(s_new["mean"] - s_old["mean"], 1) + out[key] = entry + + threshold = float(CONFIG["curation_threshold"]) + if old_rows and len(old_rows) == len(new_rows): + pairs = [(o.get("curation_score"), n.get("curation_score")) for o, n in zip(old_rows, new_rows, strict=True)] + out["_threshold"] = { + "threshold": threshold, + "old_selected": sum(1 for o, _ in pairs if (o or 0) >= threshold), + "new_selected": sum(1 for _, n in pairs if (n or 0) >= threshold), + "crossed_up": sum(1 for o, n in pairs if (o or 0) < threshold <= (n or 0)), + "crossed_down": sum(1 for o, n in pairs if (n or 0) < threshold <= (o or 0)), + } + return out + + +def render_markdown(summary: dict[str, Any]) -> str: + lines = [ + "# 六维行为锚点对照跑分(#114)", + "", + "| 维度 | n(旧/新) | 旧 mean | 新 mean | mean 迁移 | 旧 p25-p75 | 新 p25-p75 |", + "|---|---|---|---|---|---|---|", + ] + for key in (*SIX_DIMS, "curation_score"): + e = summary[key] + old, new = e["old"], e["new"] + shift = e.get("mean_shift") + lines.append( + f"| {key} | {old['n']}/{new['n']} " + f"| {old.get('mean', '-')} | {new.get('mean', '-')} " + f"| {shift if shift is not None else '-'} " + f"| {old.get('p25', '-')}-{old.get('p75', '-')} " + f"| {new.get('p25', '-')}-{new.get('p75', '-')} |" + ) + if "_threshold" in summary: + t = summary["_threshold"] + lines += [ + "", + f"精选门槛 {t['threshold']:g}:旧版入选 {t['old_selected']} 条 → 新版 {t['new_selected']} 条" + f"(升过门槛 {t['crossed_up']},跌出 {t['crossed_down']})", + ] + return "\n".join(lines) + + +# ── 采样与 LLM 对照 ───────────────────────────────────────────────── + + +async def fetch_samples(limit: int, hours: int) -> list[ContentItem]: + """近 N 小时有点击正文的内容,按抓取时间倒序;arXiv 与无正文样本在配对阶段剔除。""" + cutoff = datetime.now(UTC) - timedelta(hours=hours) + async with async_session() as session: + stmt = ( + select(ContentItem) + .where( + ContentItem.crawled_at >= cutoff, + or_( + ContentItem.raw_content.isnot(None), + ContentItem.summary.isnot(None), + ), + ) + .order_by(ContentItem.crawled_at.desc()) + .limit(limit * 3) + ) + rows = (await session.execute(stmt)).scalars().all() + + samples: list[ContentItem] = [] + for c in rows: + if len(samples) >= limit: + break + title = c.title or "" + text = c.raw_content or c.summary or "" + if not title or not text: + continue + _, _, _, is_arxiv = _select_analysis_prompts(c, title=title, content_text=text) + if is_arxiv: + continue + samples.append(c) + return samples + + +def build_prompt_pair(content: ContentItem) -> tuple[list[dict[str, str]], list[dict[str, str]]]: + """同一内容的 新(带锚点)/旧(去锚点) 两条 messages。""" + title = content.title or "" + text = content.raw_content or content.summary or "" + truncated = text[:3000] if text else "无正文内容" + system_prompt, tpl, _, _ = _select_analysis_prompts(content, title=title, content_text=text) + rendered = tpl.format( + title=sanitize_prompt_input(title, max_chars=500), + content=sanitize_prompt_input(truncated, max_chars=3000), + ) + section = dimension_anchor_section("en" if tpl is ANALYSIS_PROMPT_EN else "zh") + old_rendered = strip_anchor_section(rendered, section) + base = {"role": "system", "content": system_prompt} + return ( + [base, {"role": "user", "content": rendered}], + [base, {"role": "user", "content": old_rendered}], + ) + + +async def analyze_once(messages: list[dict[str, str]], sem: asyncio.Semaphore) -> dict[str, float | None] | None: + async with sem: + try: + result = await call_llm_json(messages) + except Exception as exc: # 单样本失败不拖垮整批 + logger.warning("LLM 调用失败,跳过该样本: %s", exc) + return None + scores = result.get("scores") or {} + curation = result.get("curation") or {} + row: dict[str, float | None] = {k: _num(scores.get(k)) for k in SIX_DIMS} + row["curation_score"] = _num(curation.get("curation_score")) + return row if any(v is not None for v in row.values()) else None + + +async def run(limit: int, hours: int, concurrency: int, out_path: str) -> None: + samples = await fetch_samples(limit, hours) + if not samples: + print("近窗口内没有可用样本,放宽 --hours 或先抓取内容。", file=sys.stderr) + sys.exit(1) + logger.info("采样 %d 条,开始对照跑分(并发 %d)…", len(samples), concurrency) + + sem = asyncio.Semaphore(concurrency) + old_rows: list[dict[str, float | None]] = [] + new_rows: list[dict[str, float | None]] = [] + + async def pair_one(content: ContentItem) -> None: + new_msgs, old_msgs = build_prompt_pair(content) + old_row, new_row = await asyncio.gather(analyze_once(old_msgs, sem), analyze_once(new_msgs, sem)) + if old_row is None or new_row is None: + logger.warning("样本「%s」有一侧失败,整对剔除", content.title[:40]) + return + old_rows.append(old_row) + new_rows.append(new_row) + + await asyncio.gather(*(pair_one(c) for c in samples)) + if not old_rows: + print("全部样本 LLM 调用失败,检查 LLM 配置。", file=sys.stderr) + sys.exit(1) + + summary = summarize_dimensions(old_rows, new_rows) + report = { + "generated_at": datetime.now(UTC).isoformat(), + "pairs": len(old_rows), + "summary": summary, + "old_rows": old_rows, + "new_rows": new_rows, + } + with open(out_path, "w", encoding="utf-8") as fh: + json.dump(report, fh, ensure_ascii=False, indent=2) + print(render_markdown(summary)) + print(f"\n明细已写入 {out_path}") + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--limit", type=int, default=30, help="配对样本数(默认 30)") + parser.add_argument("--hours", type=int, default=168, help="采样窗口小时数(默认 168)") + parser.add_argument("--concurrency", type=int, default=4, help="LLM 并发(默认 4)") + parser.add_argument("--out", default="anchor_ab.json", help="明细输出路径") + args = parser.parse_args(argv) + + logging.basicConfig(level=logging.WARNING) + logger.setLevel(logging.INFO) + asyncio.run( + run( + limit=args.limit, + hours=args.hours, + concurrency=args.concurrency, + out_path=args.out, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_scoring_anchor_ab.py b/backend/tests/test_scoring_anchor_ab.py new file mode 100644 index 00000000..fbdf884f --- /dev/null +++ b/backend/tests/test_scoring_anchor_ab.py @@ -0,0 +1,86 @@ +"""scoring_anchor_ab 对照跑分工具的纯函数单测(#114)。 + +覆盖:锚点段落剥离能还原旧 prompt(对照基线的正确性)、分布汇总与门槛 +跨越统计的数值正确性。LLM 与 DB 路径不在此测(运行期工具,由 #114 的 +对照跑分本身验证)。 +""" + +from __future__ import annotations + +import importlib.util +import os + +from app.services.llm.prompts.analysis import ANALYSIS_PROMPT, ANALYSIS_PROMPT_EN +from app.services.scoring_caliber import dimension_anchor_section + +_SCRIPT_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "scripts", + "scoring_anchor_ab.py", +) +_spec = importlib.util.spec_from_file_location("scoring_anchor_ab", _SCRIPT_PATH) +assert _spec and _spec.loader +ab = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(ab) + + +def test_strip_anchor_section_restores_pre_anchor_layout(): + rendered = ANALYSIS_PROMPT.format(title="T", content="C") + section = dimension_anchor_section() + old = ab.strip_anchor_section(rendered, section) + + assert "六维评分锚点" not in old + assert section not in old + assert "精选分（curation_score）评判标准：" in old + # 剥掉的恰是「段落 + 一组相邻空行」(2 个换行符),布局回到挂锚点前的形态 + assert len(rendered) - len(old) == len(section) + 2 + # 幂等:对已剥离文本再剥应报错(段落已不存在) + try: + ab.strip_anchor_section(old, section) + raise AssertionError("已剥离的 prompt 再剥离应报 ValueError") + except ValueError: + pass + + +def test_strip_anchor_section_en_prompt(): + rendered = ANALYSIS_PROMPT_EN.format(title="T", content="C") + old = ab.strip_anchor_section(rendered, dimension_anchor_section("en")) + assert "Six-dimension scoring anchors" not in old + assert "Curation score (curation_score) guidelines:" in old + + +def test_summarize_dimensions_and_threshold_crossing(): + # 2 组配对:均值迁移 + 门槛跨越(阈值取 CONFIG,默认 55) + threshold = ab.CONFIG["curation_threshold"] + old_rows = [ + {"curation_score": 50.0, "quality_score": 40.0}, + {"curation_score": 60.0, "quality_score": 60.0}, + ] + new_rows = [ + {"curation_score": 58.0, "quality_score": 55.0}, + {"curation_score": 52.0, "quality_score": 65.0}, + ] + summary = ab.summarize_dimensions(old_rows, new_rows) + + q = summary["quality_score"] + assert q["old"]["mean"] == 50.0 + assert q["new"]["mean"] == 60.0 + assert q["mean_shift"] == 10.0 + + t = summary["_threshold"] + assert t["threshold"] == float(threshold) + assert t["old_selected"] == 1 # 60 过线 + assert t["new_selected"] == 1 # 58 过线 + assert t["crossed_up"] == 1 # 50 → 58 + assert t["crossed_down"] == 1 # 60 → 52 + + +def test_summarize_dimensions_missing_dim_is_reported_not_raised(): + summary = ab.summarize_dimensions( + [{"curation_score": None, "risk_score": 10.0}], + [{"curation_score": 70.0, "risk_score": 30.0}], + ) + assert summary["curation_score"]["old"]["n"] == 0 + assert summary["curation_score"]["new"]["n"] == 1 + assert "mean_shift" not in summary["curation_score"] + assert summary["risk_score"]["mean_shift"] == 20.0 diff --git a/backend/tests/test_scoring_caliber.py b/backend/tests/test_scoring_caliber.py index b69d9c4b..507ce310 100644 --- a/backend/tests/test_scoring_caliber.py +++ b/backend/tests/test_scoring_caliber.py @@ -17,9 +17,14 @@ from app.services import scoring_engine from app.services.llm.prompts import analysis as analysis_prompts from app.services.scoring_caliber import ( + ANCHOR_BANDS, CONFIG, + DIMENSION_ANCHORS_EN, + DIMENSION_ANCHORS_ZH, LLM_RISK_PENALTY_POINTS, LLM_RISK_PENALTY_TRIGGER, + dimension_anchor_lines, + dimension_anchor_section, ) SIX_DIM_KEYS = ( @@ -141,3 +146,48 @@ def test_today_picks_fallback_weight_bonus_matches_config(): from app.services.today_picks import _OLTP_FALLBACK_WEIGHT_BONUS assert CONFIG["picks_weight_bonus"] == _OLTP_FALLBACK_WEIGHT_BONUS + + +# ── 六维行为锚点(#114) ─────────────────────────────────────────────── + +SIX_SCORE_FIELDS = { + "quality_score", + "hot_score", + "freshness_score", + "creator_score", + "viral_score", + "risk_score", +} + + +def test_anchor_tables_complete(): + """中英锚点表必须同维、每维恰四档、每档一句非空行为描述。""" + assert set(DIMENSION_ANCHORS_ZH) == SIX_SCORE_FIELDS + assert set(DIMENSION_ANCHORS_EN) == SIX_SCORE_FIELDS + for table in (DIMENSION_ANCHORS_ZH, DIMENSION_ANCHORS_EN): + for dim, descriptions in table.items(): + assert len(descriptions) == len(ANCHOR_BANDS), f"{dim} 档数不符" + assert all(isinstance(d, str) and d.strip() for d in descriptions), f"{dim} 有空档描述" + + +def test_prompts_contain_anchor_sections_verbatim(): + """中/英 prompt 的锚点段落必须与 caliber 渲染逐字一致;档位边界同源。""" + zh = analysis_prompts.ANALYSIS_PROMPT + en = analysis_prompts.ANALYSIS_PROMPT_EN + assert dimension_anchor_section() in zh, "中文 prompt 缺锚点段或被手改" + assert dimension_anchor_section("en") in en, "英文 prompt 缺锚点段或被手改" + + expected_bands = [f"{lo}-{hi}" for lo, hi in ANCHOR_BANDS] + for lang, prompt in (("zh", zh), ("en", en)): + for dim in sorted(SIX_SCORE_FIELDS): + m = re.search(rf"^- {dim}[：:].*$", prompt, re.M) + assert m, f"{lang} prompt 缺 {dim} 锚点行" + bands = re.findall(r"(\d+-\d+)=", m.group(0)) + assert bands == expected_bands, f"{lang} {dim} 档位与 ANCHOR_BANDS 不一致: {m.group(0)}" + + +def test_paper_prompt_excludes_anchor_section(): + """论文版是否挂锚点是独立决策(#114),首期明确不挂——锁住范围防误植。""" + assert "六维评分锚点" not in analysis_prompts.PAPER_ANALYSIS_PROMPT + for line in dimension_anchor_lines(): + assert line not in analysis_prompts.PAPER_ANALYSIS_PROMPT