From f09f893b50aaed15d39d6bbddb142853585276a9 Mon Sep 17 00:00:00 2001 From: fxbin Date: Mon, 12 Oct 2026 00:38:29 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(evaluation):=20SelectBench=20v0?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=8F=8C=E5=BC=95=E6=93=8E=E5=AF=B9=E7=85=A7?= =?UTF-8?q?=E7=9F=A9=E9=98=B5=E4=B8=8E=E9=87=91=E6=A0=87=E6=A0=A1=E5=87=86?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E9=93=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 四个手动离线脚本（校准工作流，人在环、按需跑，不接入调度器）： - selectbench_matrix：拉 AIHOT 精选 feed 与库内全池重放对齐成四象限 （首份结果：一致率 26.6%，TopicEye 对该源选出 77% vs 其自选 ~4%， 精选率偏高在单一信源上的集中实证）；AIHOT 信号仅内部对照、不导出 不分发不训练（用途边界与所有者决定记录于脚本 docstring） - selectbench_sample：分歧加权分层抽样导出标注候选 CSV - selectbench_import_labels：人工标注 CSV 批量导入 content_labels - selectbench_eval：金标集 1×4 percentile 扫描，P/R/F1 + FP/FN 错例清单 仅存于本分支（所有者决定），不合并 main。 --- backend/scripts/selectbench_eval.py | 151 +++++++++++++ backend/scripts/selectbench_import_labels.py | 75 +++++++ backend/scripts/selectbench_matrix.py | 217 +++++++++++++++++++ backend/scripts/selectbench_sample.py | 154 +++++++++++++ 4 files changed, 597 insertions(+) create mode 100644 backend/scripts/selectbench_eval.py create mode 100644 backend/scripts/selectbench_import_labels.py create mode 100644 backend/scripts/selectbench_matrix.py create mode 100644 backend/scripts/selectbench_sample.py diff --git a/backend/scripts/selectbench_eval.py b/backend/scripts/selectbench_eval.py new file mode 100644 index 00000000..545ad6b7 --- /dev/null +++ b/backend/scripts/selectbench_eval.py @@ -0,0 +1,151 @@ +"""SelectBench 门槛扫描：金标集上重放评分引擎，扫描 curation_percentile（v0）。 + +输入：content_labels（human / pick_mark 源）的标注条目 + 各自最新分析的六维 +分数（重放输入）。金标批作为一个固定批次过 score_items——考卷语义：同一批 +上比较参数。v0 扫 1×4 网格（curation_percentile ∈ {65,70,75,80}），每档输出 +precision / recall / F1 / 选中率 + FP/FN 错例清单。 + +注意（v0 已知局限，正式版需补）： +- 分数取「当前最新分析」而非冻结快照——requeue 重分析会使考卷漂移； + 结果 JSON 里记录了 analysis_id 清单供追溯，金标冻结表是后续切片。 +- 样本 <300 时指标噪声大于信号（F1 的 95% CI 约 ±8pp@200 条），读错例的 + 定性模式，不比小数点。 + +用法： + uv run python scripts/selectbench_eval.py [--percentiles 65,70,75,80] [--out DIR] +""" + +from __future__ import annotations # noqa: E402 + +import argparse # noqa: E402 +import asyncio # noqa: E402 +import json # noqa: E402 +import logging # noqa: E402 +import os # noqa: E402 +import sys # noqa: E402 +from datetime import UTC, datetime # noqa: E402 + +_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 select # noqa: E402 + +from app.core.database import async_session # noqa: E402 +from app.models.content import ContentItem # noqa: E402 +from app.models.content_label import ContentLabel # noqa: E402 +from app.services import scoring_engine # noqa: E402 +from app.services.scoring_inputs import build_scoring_input # noqa: E402 + +logger = logging.getLogger("selectbench_eval") +logging.basicConfig(level=logging.INFO, format="%(message)s") + + +async def load_gold_set() -> tuple[list, dict[int, str], dict[int, int]]: + """金标 = human/pick_mark 标注的条目；返回 (items, label_map, analysis_id_map)。""" + async with async_session() as db: + rows = ( + await db.execute( + select(ContentLabel, ContentItem) + .join(ContentItem, ContentItem.id == ContentLabel.content_id) + .where(ContentLabel.source.in_(("human", "pick_mark"))) + .where(ContentLabel.label != "uncertain") + ) + ).all() + items, label_map, analysis_map = [], {}, {} + for label_row, item in rows: + if not item.analyses: + continue + items.append(item) + label_map[item.id] = label_row.label + analysis_map[item.id] = item.analyses[-1].id + return items, label_map, analysis_map + + +def _metrics(scored, label_map) -> dict: + tp = fp = fn = 0 + fps, fns = [], [] + for bd, item in scored: + truth = label_map[item.content_id] + pred = bd.selected + if pred and truth == "worth_writing": + tp += 1 + elif pred and truth == "not_worth": + fp += 1 + fps.append({"content_id": item.content_id, "final_score": bd.final_score, "title": item.title[:60]}) + elif not pred and truth == "worth_writing": + fn += 1 + fns.append({"content_id": item.content_id, "final_score": bd.final_score, "title": item.title[:60]}) + precision = tp / (tp + fp) if tp + fp else None + recall = tp / (tp + fn) if tp + fn else None + f1 = 2 * precision * recall / (precision + recall) if precision and recall else None + total = len(scored) + return { + "n": total, + "selected": tp + fp, + "selection_rate": round((tp + fp) / total, 3) if total else None, + "precision": round(precision, 3) if precision is not None else None, + "recall": round(recall, 3) if recall is not None else None, + "f1": round(f1, 3) if f1 is not None else None, + "false_positives": fps, + "false_negatives": fns, + } + + +async def run(percentiles: list[int], out_dir: str) -> None: + items, label_map, analysis_map = await load_gold_set() + if not items: + raise SystemExit( + "content_labels 无 human/pick_mark 标注（uncertain 除外）——" + "先跑 selectbench_sample.py 导出 → 人工标注 → selectbench_import_labels.py 导入" + ) + inputs = [build_scoring_input(item) for item in items] + gold_pos = sum(1 for v in label_map.values() if v == "worth_writing") + logger.info("金标集 %d 条（正 %d / 负 %d）", len(items), gold_pos, len(items) - gold_pos) + + original = scoring_engine.CONFIG["curation_percentile"] + results: dict[str, dict] = {} + try: + for p in percentiles: + scoring_engine.CONFIG["curation_percentile"] = p + scored = scoring_engine.score_items(list(inputs)) + results[f"p{p}"] = _metrics(scored, label_map) + m = results[f"p{p}"] + logger.info( + "percentile=%-3d 选中率 %.1f%% P=%.3f R=%.3f F1=%.3f FP=%d FN=%d", + p, + (m["selection_rate"] or 0) * 100, + m["precision"] or 0, + m["recall"] or 0, + m["f1"] or 0, + len(m["false_positives"]), + len(m["false_negatives"]), + ) + finally: + scoring_engine.CONFIG["curation_percentile"] = original + + payload = { + "generated_at": datetime.now(UTC).isoformat(), + "gold_size": len(items), + "gold_positive": gold_pos, + "analysis_ids": analysis_map, # 溯源：重分析漂移时定位考卷变化 + "scoring_config_baseline": {"original_percentile": original}, + "results": results, + } + os.makedirs(out_dir, exist_ok=True) + out_path = os.path.join(out_dir, f"eval_{datetime.now(UTC):%Y%m%d_%H%M}.json") + with open(out_path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=2) + logger.info("结果已写入 %s（错例清单在其中）", out_path) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--percentiles", default="65,70,75,80") + parser.add_argument("--out", default="data/selectbench") + args = parser.parse_args() + asyncio.run(run([int(x) for x in args.percentiles.split(",")], args.out)) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/selectbench_import_labels.py b/backend/scripts/selectbench_import_labels.py new file mode 100644 index 00000000..46b9b047 --- /dev/null +++ b/backend/scripts/selectbench_import_labels.py @@ -0,0 +1,75 @@ +"""SelectBench 标注导入：把人工标注完的 CSV 批量写入 content_labels（human）。 + +输入：selectbench_sample.py 导出、label 列已填（worth_writing / not_worth / +uncertain）的 CSV。走 ContentLabelRepository.set_authoritative（与 admin +单条标注同一写路径，human 源覆盖任何已有来源），幂等可重复执行。 + +用法： + uv run python scripts/selectbench_import_labels.py --csv data/selectbench/annotation_batch_YYYYMMDD.csv [--dry-run] +""" + +from __future__ import annotations # noqa: E402 + +import argparse # noqa: E402 +import asyncio # noqa: E402 +import csv # noqa: E402 +import logging # noqa: E402 +import os # noqa: E402 +import sys # noqa: E402 + +_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 select # noqa: E402 + +from app.core.database import async_session # noqa: E402 +from app.models.user import User, UserRole # noqa: E402 +from app.repositories.content_label_repo import ContentLabelRepository # noqa: E402 + +logger = logging.getLogger("selectbench_import") +logging.basicConfig(level=logging.INFO, format="%(message)s") + +_VALID = {"worth_writing", "not_worth", "uncertain"} + + +async def import_labels(csv_path: str, dry_run: bool) -> None: + with open(csv_path, encoding="utf-8-sig") as f: + rows = list(csv.DictReader(f)) + labeled = [r for r in rows if (r.get("label") or "").strip()] + invalid = [r for r in labeled if r["label"].strip() not in _VALID] + if invalid: + raise SystemExit(f"非法 label 值：{[r['label'] for r in invalid[:5]]}，允许值 {sorted(_VALID)}") + skipped = len(rows) - len(labeled) + logger.info("CSV %d 行：待导入 %d，跳过未标注 %d", len(rows), len(labeled), skipped) + + async with async_session() as db: + admin = (await db.execute(select(User).where(User.role == UserRole.ADMIN.value).limit(1))).scalars().first() + labeled_by = admin.id if admin else None + if dry_run: + logger.info("[dry-run] 将以 labeled_by=%s 导入 %d 条", labeled_by, len(labeled)) + return + repo = ContentLabelRepository(db) + counts: dict[str, int] = {} + for row in labeled: + await repo.set_authoritative( + int(row["content_id"]), + row["label"].strip(), + source="human", + labeled_by=labeled_by, + ) + counts[row["label"].strip()] = counts.get(row["label"].strip(), 0) + 1 + await db.commit() + logger.info("导入完成：%s", counts) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--csv", required=True) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + asyncio.run(import_labels(args.csv, args.dry_run)) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/selectbench_matrix.py b/backend/scripts/selectbench_matrix.py new file mode 100644 index 00000000..e6d1f2ad --- /dev/null +++ b/backend/scripts/selectbench_matrix.py @@ -0,0 +1,217 @@ +"""SelectBench 对照矩阵：TopicEye 评分 × AIHOT 精选信号（v0 一次性对照实验）。 + +同一池内容（AIHOT all feed 已在 TopicEye 信源内），把「TopicEye 引擎是否精选」 +与「AIHOT 是否精选」（/feed.xml guid）对齐成四象限： + + 双高一致（都选） → 高置信正样本候选 + 分歧 1（AIHOT 选、TopicEye 未选）→ 人工重点标注（漏选归因） + 分歧 2（TopicEye 选、AIHOT 未选）→ 人工重点标注（口径差异归因） + 双低一致（都不选） → 高置信负样本候选 + +用途边界（所有者决定，2026-10-11）：AIHOT 精选信号仅用于内部一致性对照与 +标注样本排序——不作为金标标签、不导入为新数据源、不分发其内容、不用于 +模型训练。两项目均为开源（AIHOT MIT / TopicEye Apache-2.0），所有者评估 +后决定以此边界使用、不做额外的用途确认（评估记录见 roundtable Memory： +selectbench-gold-source）。 + +用法： + uv run python scripts/selectbench_matrix.py [--days 7] [--out DIR] + +输出：stdout 报告 + /matrix_.json（明细供 sample.py 分层抽样）。 +""" + +from __future__ import annotations # noqa: E402 + +import argparse # noqa: E402 +import asyncio # noqa: E402 +import json # noqa: E402 +import logging # noqa: E402 +import os # noqa: E402 +import re # noqa: E402 +import sys # noqa: E402 +import xml.etree.ElementTree as ET # noqa: E402 +from datetime import UTC, datetime, timedelta # noqa: E402 + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _ROOT not in sys.path: + sys.path.insert(0, _ROOT) + +import httpx # noqa: E402 +from sqlalchemy import select # noqa: E402 + +from app.core.database import async_session # noqa: E402 +from app.models.source import Source # noqa: E402 +from app.repositories.content_repo import ContentRepo # noqa: E402 +from app.services.scoring_engine import score_items # noqa: E402 +from app.services.scoring_inputs import build_scoring_inputs # noqa: E402 + +logger = logging.getLogger("selectbench_matrix") +logging.basicConfig(level=logging.INFO, format="%(message)s") + +FEED_URL = "https://aihot.news/feed.xml" +_ORIGINAL_URL_RE = re.compile(r'href="(https?://[^"]+)"[^>]*>阅读原文') + + +async def fetch_aihot_selected() -> list[dict]: + """拉精选 feed（最近 50 条），返回 guid/标题/发布时间/原文链接。""" + async with httpx.AsyncClient(timeout=20) as client: + resp = await client.get(FEED_URL, headers={"User-Agent": "TopicEye-SelectBench/0.1 (internal calibration)"}) + resp.raise_for_status() + root = ET.fromstring(resp.text) + items: list[dict] = [] + for node in root.findall("./channel/item"): + guid = (node.findtext("guid") or "").strip() + if not guid: + continue + original = "" + if (desc := node.findtext("description")) and (m := _ORIGINAL_URL_RE.search(desc)): + original = m.group(1) + items.append( + { + "guid": guid, + "title": (node.findtext("title") or "").strip(), + "published": node.findtext("pubDate") or "", + "original_url": original, + "category": node.findtext("category") or "", + } + ) + return items + + +def _parse_rss_date(value: str) -> datetime | None: + try: + from email.utils import parsedate_to_datetime + + dt = parsedate_to_datetime(value) + return dt.astimezone(UTC) if dt.tzinfo else dt.replace(tzinfo=UTC) + except Exception: + return None + + +async def build_matrix(days: int, out_dir: str) -> dict: + feed_items = await fetch_aihot_selected() + if not feed_items: + raise SystemExit("AIHOT 精选 feed 为空，中止") + pub_dates = [d for d in (_parse_rss_date(i["published"]) for i in feed_items) if d] + window_start = min(pub_dates) - timedelta(days=1) if pub_dates else datetime.now(UTC) - timedelta(days=days) + selected_guids = {i["guid"] for i in feed_items} + logger.info( + "AIHOT 精选 feed：%d 条，pubDate 跨度 %s ~ %s", + len(feed_items), + min(pub_dates).date() if pub_dates else "?", + max(pub_dates).date() if pub_dates else "?", + ) + + async with async_session() as db: + aihot_sources = (await db.execute(select(Source).where(Source.url.ilike("%aihot%")))).scalars().all() + if not aihot_sources: + raise SystemExit("信源表中未找到 AIHOT 信源（url 含 aihot）") + aihot_source_ids = {s.id for s in aihot_sources} + + # 全池重放：窗口内全部分析完成内容过一遍引擎，得到 selected 集合与名次 + repo = ContentRepo(db) + items, total = await repo.list_for_scoring( + time_cutoff=window_start, + limit=5000, + ) + scoring_inputs, item_map, _ = await build_scoring_inputs(db, items) + scored = score_items(scoring_inputs) + rank_by_id = {item.content_id: rank + 1 for rank, (_bd, item) in enumerate(scored)} + selected_ids = {item.content_id for bd, item in scored if bd.selected} + logger.info( + "TopicEye 全池重放：%d 条候选（窗口 %.1f 天），selected %d 条", + total, + (datetime.now(UTC) - window_start).total_seconds() / 86400, + len(selected_ids), + ) + + # AIHOT 源内容：guid 对齐（content url 尾段）+ 象限归类 + quadrants = {"both": [], "aihot_only": [], "te_only": [], "neither": [], "outside_feed_window": []} + aihot_items = [it for it in items if it.source_id in aihot_source_ids] + matched_guids: set[str] = set() + for it in aihot_items: + guid = (it.url or "").rsplit("/", 1)[-1] + if guid in selected_guids: + matched_guids.add(guid) + aihot_sel = guid in selected_guids + te_sel = it.id in selected_ids + # 「AIHOT 未选」只有在内容入库时间落在 feed 覆盖窗口内才可判定： + # 窗口外的内容它可能还没评到、或评过但已滚出最近 50 条——归为 + # 未知，不进四象限（单次 feed 快照的 v0 已知局限）。 + created = ( + it.created_at.replace(tzinfo=UTC) if it.created_at and it.created_at.tzinfo is None else it.created_at + ) + if not aihot_sel and created and created < window_start: + quadrant = "outside_feed_window" + elif aihot_sel and te_sel: + quadrant = "both" + elif aihot_sel: + quadrant = "aihot_only" + elif te_sel: + quadrant = "te_only" + else: + quadrant = "neither" + analysis = it.analyses[-1] if it.analyses else None + row = { + "content_id": it.id, + "guid": guid, + "title": (it.title or "")[:100], + "aihot_selected": aihot_sel, + "te_selected": te_sel, + "te_rank": rank_by_id.get(it.id), + "curation_score": float(analysis.curation_score) + if analysis and analysis.curation_score is not None + else None, + "summary_source": analysis.summary_source if analysis else None, + } + quadrants[quadrant].append(row) + + feed_only = selected_guids - matched_guids # AIHOT 精选但库内未对齐（空窗/回溯） + n_aihot = len(aihot_items) + n_scored = n_aihot - len(quadrants["outside_feed_window"]) + n_agree = len(quadrants["both"]) + len(quadrants["neither"]) + report = { + "generated_at": datetime.now(UTC).isoformat(), + "feed_items": len(feed_items), + "feed_window_start": window_start.isoformat(), + "aihot_pool_items": n_aihot, + "matched_selected": len(matched_guids), + "feed_only_unmatched": sorted(feed_only), + "quadrants": {k: len(v) for k, v in quadrants.items()}, + "agreement_rate": round(n_agree / n_scored, 3) if n_scored else None, + "n_in_window": n_scored, + "detail": quadrants, + } + + logger.info("=" * 56) + logger.info("四象限（AIHOT 源内容 %d 条）：", n_aihot) + for key, desc in ( + ("both", "双高一致（都选）→ 正样本候选"), + ("aihot_only", "分歧：AIHOT 选、TopicEye 未选 → 漏选归因重点"), + ("te_only", "分歧：TopicEye 选、AIHOT 未选 → 口径差异重点"), + ("neither", "双低一致（都不选）→ 负样本候选"), + ("outside_feed_window", "窗口外（AIHOT 判定未知，不计入一致率）"), + ): + logger.info(" %-11s %4d 条 %s", key, len(quadrants[key]), desc) + logger.info("一致率：%s", report["agreement_rate"]) + if feed_only: + logger.info("AIHOT 精选但库内未对齐 %d 条（抓取空窗/回溯补选）", len(feed_only)) + + os.makedirs(out_dir, exist_ok=True) + out_path = os.path.join(out_dir, f"matrix_{datetime.now(UTC):%Y%m%d}.json") + with open(out_path, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) + logger.info("明细已写入 %s", out_path) + return report + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--days", type=int, default=7, help="feed 窗口不足时的回退天数") + parser.add_argument("--out", default="data/selectbench", help="输出目录") + args = parser.parse_args() + asyncio.run(build_matrix(args.days, args.out)) + + +if __name__ == "__main__": + main() diff --git a/backend/scripts/selectbench_sample.py b/backend/scripts/selectbench_sample.py new file mode 100644 index 00000000..bea22f1e --- /dev/null +++ b/backend/scripts/selectbench_sample.py @@ -0,0 +1,154 @@ +"""SelectBench 标注样本导出：按对照矩阵分层 + 分歧加权（v0）。 + +从 matrix_.json 的四象限中按「分歧层超采样」配额抽标注候选，并补一层 +「非 AIHOT 源全池分层」（AIHOT 信源只覆盖部分分布，金标必须含其余信源内容； +负样本从全池分位抽样，禁按被扫描的分数变量分层——方法席复核结论）。 + +输出 CSV 供人工标注（label 列填 worth_writing / not_worth / uncertain）， +随后由 selectbench_import_labels.py 导入 content_labels（source=human）。 + +用法： + uv run python scripts/selectbench_sample.py [--matrix PATH] [--total 200] +""" + +from __future__ import annotations # noqa: E402 + +import argparse # noqa: E402 +import asyncio # noqa: E402 +import csv # noqa: E402 +import json # noqa: E402 +import logging # noqa: E402 +import os # noqa: E402 +import random # noqa: E402 +import sys # noqa: E402 +from datetime import UTC, datetime, timedelta # noqa: E402 + +_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 select # noqa: E402 + +from app.core.database import async_session # noqa: E402 +from app.models.source import Source # noqa: E402 +from app.repositories.content_repo import ContentRepo # noqa: E402 +from app.services.scoring_engine import score_items # noqa: E402 +from app.services.scoring_inputs import build_scoring_inputs # noqa: E402 + +logger = logging.getLogger("selectbench_sample") +logging.basicConfig(level=logging.INFO, format="%(message)s") + + +def _take(rows: list[dict], n: int, rng: random.Random) -> list[dict]: + return rng.sample(rows, min(n, len(rows))) + + +async def load_non_aihot_layer(quota: int, rng: random.Random) -> list[dict]: + """非 AIHOT 信源的全池分层样本：按最终名次分位抽（覆盖头/中/尾段）。 + + 用名次（而非 curation 分数）分层——curation 是扫描目标变量的一部分， + 直接按它分层会人为制造近可分金标。 + """ + async with async_session() as db: + aihot_ids = set((await db.execute(select(Source.id).where(Source.url.ilike("%aihot%")))).scalars().all()) + items, total = await ContentRepo(db).list_for_scoring( + time_cutoff=datetime.now(UTC) - timedelta(days=14), + limit=5000, + ) + scoring_inputs, item_map, _ = await build_scoring_inputs(db, items) + scored = score_items(scoring_inputs) + rows = [] + for rank, (bd, item) in enumerate(scored): + if item.source_id in aihot_ids: + continue + analysis = item_map[item.content_id].analyses[-1] if item_map.get(item.content_id) else None + rows.append( + { + "content_id": item.content_id, + "title": (item_map[item.content_id].title if item_map.get(item.content_id) else "")[:100], + "url": item_map[item.content_id].url if item_map.get(item.content_id) else "", + "te_selected": bd.selected, + "te_rank": rank + 1, + "curation_score": float(analysis.curation_score) + if analysis and analysis.curation_score is not None + else None, + "summary_source": analysis.summary_source if analysis else None, + } + ) + if not rows: + return [] + # 名次三分层（头 20% / 中 30-60% / 尾 80% 后），各抽 1/3 + n = len(rows) + bands = (rows[: max(1, n // 5)], rows[n // 3 : n * 3 // 5], rows[n * 4 // 5 :]) + picked: list[dict] = [] + for band in bands: + picked.extend(_take(band, quota // 3, rng)) + for row in picked: + row["quadrant"] = "non_aihot_pool" + logger.info("非 AIHOT 全池分层：候选 %d 条（总池 %d），抽 %d", len(rows), total, len(picked)) + return picked + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--matrix", default=None, help="matrix json 路径（默认 data/selectbench/ 最新一份）") + parser.add_argument("--total", type=int, default=200, help="标注候选总数") + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--out", default="data/selectbench", help="输出目录") + args = parser.parse_args() + + out_dir = os.path.join(_ROOT, args.out) + if args.matrix: + matrix_path = args.matrix + else: + candidates = sorted(f for f in os.listdir(out_dir) if f.startswith("matrix_") and f.endswith(".json")) + if not candidates: + raise SystemExit("data/selectbench/ 下没有 matrix_*.json，先跑 selectbench_matrix.py") + matrix_path = os.path.join(out_dir, candidates[-1]) + with open(matrix_path, encoding="utf-8") as f: + matrix = json.load(f) + + rng = random.Random(args.seed) + detail = matrix["detail"] + # 配额：分歧层 55%（重点）、双高 15%、双低 15%、非 AIHOT 全池 15% + quotas = { + "aihot_only": int(args.total * 0.30), + "te_only": int(args.total * 0.25), + "both": int(args.total * 0.15), + "neither": int(args.total * 0.15), + } + rows: list[dict] = [] + for quadrant, quota in quotas.items(): + picked = _take(detail[quadrant], quota, rng) + for row in picked: + row["quadrant"] = quadrant + rows.extend(picked) + logger.info("%-11s 配额 %3d → 实际 %3d", quadrant, quota, len(picked)) + + non_aihot_quota = args.total - len(rows) + if non_aihot_quota > 0: + rows.extend(asyncio.run(load_non_aihot_layer(non_aihot_quota, rng))) + + fieldnames = [ + "content_id", + "quadrant", + "title", + "url", + "aihot_selected", + "te_selected", + "te_rank", + "curation_score", + "summary_source", + "label", + "notes", + ] + out_path = os.path.join(out_dir, f"annotation_batch_{datetime.now(UTC):%Y%m%d}.csv") + with open(out_path, "w", newline="", encoding="utf-8-sig") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") + writer.writeheader() + writer.writerows(rows) + logger.info("标注候选 %d 条已导出 %s（label 列留空待人工填写）", len(rows), out_path) + + +if __name__ == "__main__": + main() From 3275592cca0ca60bafa5d5f5157f509c95034744 Mon Sep 17 00:00:00 2001 From: fxbin Date: Mon, 12 Oct 2026 02:05:22 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(scoring):=20=E8=AF=84=E5=88=86=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E6=94=B6=E6=8B=A2=E4=B8=BA=20scoring=5Fcaliber=20?= =?UTF-8?q?=E5=8D=95=E4=B8=80=E5=A3=B0=E6=98=8E=E5=B9=B6=E5=8A=A0=E6=BC=82?= =?UTF-8?q?=E7=A7=BB=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95=20(#110)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 同一套精选评分口径曾分散在三处各写一份且互相矛盾:prompt 教 LLM 五项 权重 30/25/20/15/10 加门槛 60,引擎 CONFIG 六维加门槛 55,picks SQL 硬编码 55/8 不读 CONFIG。CONFIG 全量迁入 scoring_caliber.py 作为唯一声明处, 三套 prompt 的口径文本从它渲染,picks 默认值改为运行时解析,today_picks fallback 常量同源;新增漂移回归测试锁死三处一致(先红后绿验证过)。 有意的行为变化仅一处:prompt 改教六维真实权重与门槛 55。 --- backend/app/services/_duckdb_picks_mixin.py | 22 ++- backend/app/services/llm/prompts/analysis.py | 51 +++++-- backend/app/services/scoring_caliber.py | 117 +++++++++++++++ backend/app/services/scoring_engine.py | 54 ++----- backend/app/services/today_picks.py | 6 +- backend/tests/test_scoring_caliber.py | 143 +++++++++++++++++++ 6 files changed, 334 insertions(+), 59 deletions(-) create mode 100644 backend/app/services/scoring_caliber.py create mode 100644 backend/tests/test_scoring_caliber.py diff --git a/backend/app/services/_duckdb_picks_mixin.py b/backend/app/services/_duckdb_picks_mixin.py index 4a5744f7..d6aec968 100644 --- a/backend/app/services/_duckdb_picks_mixin.py +++ b/backend/app/services/_duckdb_picks_mixin.py @@ -12,12 +12,26 @@ from app.services.scoring_engine import CONFIG as SCORING_CONFIG +def resolve_picks_params( + curation_threshold: float | None, + weight_bonus: float | None, +) -> tuple[float, float]: + """query_today_picks 的口径默认值统一取自 SCORING_CONFIG(#110)。 + + 此前签名里硬编码 55/8,CONFIG 调整时这里静默不跟随。 + """ + return ( + float(SCORING_CONFIG["curation_threshold"] if curation_threshold is None else curation_threshold), + float(SCORING_CONFIG["picks_weight_bonus"] if weight_bonus is None else weight_bonus), + ) + + class PicksMixin: def query_today_picks( self, hours: int = 48, - curation_threshold: float = 55, - weight_bonus: int = 8, + curation_threshold: float | None = None, + weight_bonus: float | None = None, risk_threshold: float | None = None, category: str | None = None, content_type: str | None = None, @@ -29,8 +43,10 @@ def query_today_picks( Top curated picks from the last N hours. Runs the curation scoring + source weight adjustment entirely in DuckDB. - Returns items with adjusted_curation_score >= threshold. + Returns items with adjusted_curation_score >= threshold. 门槛与加成 + 默认值运行时取自 SCORING_CONFIG(#110),签名不再硬编码数字。 """ + curation_threshold, weight_bonus = resolve_picks_params(curation_threshold, weight_bonus) conn = self._get_conn() cutoff = (datetime.now(UTC) - timedelta(hours=hours)).isoformat() # 忽略口径按用户隔离：全局屏蔽 + 当前用户个人忽略（匿名只见全局） diff --git a/backend/app/services/llm/prompts/analysis.py b/backend/app/services/llm/prompts/analysis.py index 8aad04bb..884e9664 100644 --- a/backend/app/services/llm/prompts/analysis.py +++ b/backend/app/services/llm/prompts/analysis.py @@ -3,8 +3,17 @@ Used by: - app.services.analysis.analyze_content + +评分口径(加权/门槛/风险扣分)一律从 app.services.scoring_caliber 渲染, +本文件不得手写数字——tests/test_scoring_caliber.py 锁死漂移(#110)。 """ +from app.services.scoring_caliber import ( + curation_threshold_text, + curation_weights_text, + risk_penalty_text, +) + SYSTEM_PROMPT = """你是一位资深内容策展分析师，负责评估内容的选题价值并决定是否入选精选。 你的评分标准参考了一线内容策展平台的精选规则： @@ -16,7 +25,8 @@ 所有评分范围 0-100。所有输出文本使用中文。语气直接、有态度、不说客套话。""" -ANALYSIS_PROMPT = """请对以下内容进行完整分析。 +ANALYSIS_PROMPT = ( + """请对以下内容进行完整分析。 标题：{title} 正文：{content} @@ -39,7 +49,11 @@ "info_density": <0-100, 信息密度：纯转发/空话=0-20, 有观点=40-60, 有数据/案例/方法=70-100>, "actionability": <0-100, 可操作性：纯资讯=10-30, 有参考价值=40-60, 能直接上手用=70-100>, "source_weight": <0-100, 来源权威度：匿名/营销号=10-30, 二手转载=40-60, 一手信源/官方/KOL=70-100>, - "curation_score": <0-100, 综合精选分（加权：信息密度30%+可操作性25%+创作者价值20%+爆文潜力15%+来源10%，风险分>70则扣20分）> + "curation_score": <0-100, 综合精选分（加权：""" + + curation_weights_text() + + """，""" + + risk_penalty_text() + + """）> }}, "recommendation": "中文摘要式推荐理由（50字以内）。先用一句话概括这条内容的核心信息，再点出为什么值得关注/适合写；不要翻译腔、不要英文、不要营销夸张词、不要只写行动建议。例：'OpenCode Zen 把优选编码模型做成统一网关，适合关注 AI 编程工具生态的人跟进'）", "creator_angles": [""], @@ -52,7 +66,10 @@ - 60-69：有参考价值但不够突出 - <60：信息量低/纯情绪/重复内容/过于个人化 -精选门槛为 60 分。""" +精选门槛为 """ + + curation_threshold_text() + + """。""" +) # ── English (HackerNews / international) prompts ──────────────────────────── @@ -75,7 +92,8 @@ All scores are 0-100. IMPORTANT: all output text must be Chinese, even if the source content is English. Be direct, opinionated, no platitudes.""" -ANALYSIS_PROMPT_EN = """Analyze the following content thoroughly. +ANALYSIS_PROMPT_EN = ( + """Analyze the following content thoroughly. Title: {title} Content: {content} @@ -99,7 +117,11 @@ "info_density": <0-100, pure share/empty talk=0-20, has opinions=40-60, has data/case/method=70-100>, "actionability": <0-100, pure news=10-30, reference value=40-60, directly actionable=70-100>, "source_weight": <0-100, anonymous/spam=10-30, second-hand=40-60, first-hand/official/KOL=70-100>, - "curation_score": <0-100, weighted: info density 30%+ actionability 25%+ creator value 20%+ viral potential 15%+ source 10%, risk>70 deducts 20 points> + "curation_score": <0-100, weighted: """ + + curation_weights_text("en") + + """, """ + + risk_penalty_text("en") + + """> }}, "recommendation": "中文摘要式推荐理由（≤50 chars）。Summarize core info in one sentence, then explain why it's worth writing about. 不要输出英文、不要营销夸张词、不要只写行动建议。Example: 'OpenCode Zen 把优选编码模型做成统一网关，适合关注 AI 编程工具生态的人跟进'", "creator_angles": [""], @@ -112,7 +134,10 @@ - 60-69: reference value but not outstanding - <60: low information / purely emotional / repetitive / overly personal -Curation threshold is 60 points.""" +Curation threshold is """ + + curation_threshold_text("en") + + """.""" +) # ── arXiv / academic paper prompts ────────────────────────────────────────── @@ -133,7 +158,8 @@ 所有评分范围 0-100。所有输出文本使用中文。语气直接、专业、不说客套话。""" -PAPER_ANALYSIS_PROMPT = """请对以下英文学术论文进行分析解读。 +PAPER_ANALYSIS_PROMPT = ( + """请对以下英文学术论文进行分析解读。 标题：{title} 正文（摘要/内容）：{content} @@ -156,7 +182,11 @@ "info_density": <0-100, 信息密度：纯概念/无实证=20-40, 有方法/实验=60-80, 有完整验证=90-100>, "actionability": <0-100, 可操作性：纯理论=10-30, 有开源代码/可复现=60-80, 能直接落地应用=90-100>, "source_weight": <0-100, 来源权威度：预印本未评审=40-60, 顶会/知名团队=70-90>, - "curation_score": <0-100, 综合精选分（加权：信息密度30%+可操作性25%+创作者价值20%+传播潜力15%+来源10%，风险分>70则扣20分）> + "curation_score": <0-100, 综合精选分（加权：""" + + curation_weights_text() + + """，""" + + risk_penalty_text() + + """）> }}, "recommendation": "中文推荐理由（50字以内）。先用一句话说清论文核心贡献，再点出为什么值得中文创作者关注。不要翻译腔、不要营销词。", "deep_read": {{ @@ -180,7 +210,10 @@ - 60-69：有参考价值但不够突出 - <60：增量工作/过于晦涩/受众极窄 -精选门槛为 60 分。""" +精选门槛为 """ + + curation_threshold_text() + + """。""" +) # ── Lite prescreen prompts ───────────────────────────────────────────────── diff --git a/backend/app/services/scoring_caliber.py b/backend/app/services/scoring_caliber.py new file mode 100644 index 00000000..c03bdc83 --- /dev/null +++ b/backend/app/services/scoring_caliber.py @@ -0,0 +1,117 @@ +""" +评分口径唯一真相源(single source of truth)。 + +历史上同一套精选评分口径分散在三个存储位置各写一份且互相矛盾(#110): + +- LLM prompt 教学(app/services/llm/prompts/analysis.py 的中/英/论文三套); +- 引擎 CONFIG(原 scoring_engine.py 顶部字典); +- picks SQL(app/services/_duckdb_picks_mixin.py 的签名默认值)。 + +本模块是这些数字的唯一声明处: + +- ``CONFIG``:引擎与 DuckDB/OLTP 查询的全部可调常数(自 scoring_engine + 迁入,scoring_engine 再导出以兼容既有 ``import CONFIG`` 的引用面); +- ``LLM_RISK_PENALTY_*``:仅存在于 prompt 教学的风险扣分规则,与引擎侧 + ``risk_soft_start``/``risk_threshold`` 软门是两个机制,不混用; +- ``curation_weights_text`` 等三个渲染函数:三套语言 prompt 的口径文本 + 从这里生成,prompt 文件不得再手写数字。 + +漂移防线:tests/test_scoring_caliber.py 用正则从 prompt 成品文本提取 +权重/门槛/扣分数字与本模块 CONFIG 断言一致——绕过本模块手改 prompt +数字,CI 即红。 +""" + +from __future__ import annotations + +CONFIG = { + # 6-dimension weights (must sum to 1.0) + "w_info_density": 0.25, # 信息密度 + "w_actionability": 0.20, # 可操作性 + "w_creator_value": 0.18, # 创作者价值 + "w_viral_potential": 0.15, # 爆文潜力 + "w_source_authority": 0.12, # 来源权威度 + "w_freshness": 0.10, # 时效新鲜度 + # Source weight mapping (source.weight 1-5 → bonus) + "source_weight_bonus_factor": 6.0, # per weight-unit above 3 + # Time decay + "time_decay_lambda": 0.02, # decay rate: exp(-lambda * hours) + "time_decay_floor": 0.3, # minimum time decay (never fully kill) + # Content quality gates + "quality_gate_min": 45, # below this, content is probably too thin for curation + "quality_gate_strong": 70, # full trust above this composite quality level + "quality_gate_floor": 0.55, # lowest multiplier for weak-but-not-risky content + "min_selected_base_score": 58, # do not select weak items just because a batch is weak + # Source diversity + "diversity_penalty_base": 0.85, # multiplier per same-source duplicate + "category_diversity_penalty_base": 0.92, + "same_source_grace": 1, # first item per source is free + "same_category_grace": 3, # first few items per category are free + # Risk + "risk_threshold": 82, # hard-exclude above this + "risk_soft_start": 45, # risk starts reducing rank after this + "risk_soft_floor": 0.55, # minimum multiplier before hard exclusion + # Curation threshold (minimum final score to be selected) + "curation_threshold": 55, # slightly lower than before, because scores are now more calibrated + # Fallback: when curation_score is 0 but other scores exist + "fallback_use_avg": True, + # ── Percentile-based curation mode ── + "curation_mode": "percentile", # "percentile" | "fixed" + "curation_percentile": 70, # top ~30% selected (P70 and above) → ~25-30% curation rate + # ── User feedback signal ── + "w_feedback": 0.15, # 15% weight for feedback_score + "feedback_score_min": -20.0, # one effective negative vote should not dominate ranking + "feedback_score_max": 20.0, # one effective positive vote should not dominate ranking + # ── picks 查询加成(#110 收编) ── + # query_today_picks 与 today_picks OLTP fallback 共用的信源权重加成 + # (每单位 source.weight(1-5)高于 3 的加分数),此前在两处各硬编码 8。 + "picks_weight_bonus": 8, + # ── curation_score 混合比例(#110 收编) ── + # base = curation_score * p + 六维加权和 * (1 - p),此前硬编码 0.6。 + "curation_blend_weight": 0.6, +} + +# ── LLM prompt 教学口径(引擎侧无对应机制,单独声明) ────────────────── +# prompt 教 LLM 组装 curation_score 时的风险扣分规则;引擎侧重用上面的 +# risk_soft_start/risk_threshold 软门,二者是不同机制,不互相推导。 +LLM_RISK_PENALTY_TRIGGER = 70 # 风险分超过此值 +LLM_RISK_PENALTY_POINTS = 20 # 扣除的分数 + +_WEIGHT_LABELS: dict[str, list[tuple[str, str]]] = { + "zh": [ + ("w_info_density", "信息密度"), + ("w_actionability", "可操作性"), + ("w_creator_value", "创作者价值"), + ("w_viral_potential", "爆文潜力"), + ("w_source_authority", "来源"), + ("w_freshness", "时效"), + ], + "en": [ + ("w_info_density", "info density"), + ("w_actionability", "actionability"), + ("w_creator_value", "creator value"), + ("w_viral_potential", "viral potential"), + ("w_source_authority", "source"), + ("w_freshness", "timeliness"), + ], +} + + +def curation_weights_text(lang: str = "zh") -> str: + """LLM curation_score 的加权口径文本,数值取自 CONFIG 六维权重。""" + labels = _WEIGHT_LABELS.get(lang, _WEIGHT_LABELS["zh"]) + sep = "+ " if lang == "en" else "+" + return sep.join(f"{label}{int(round(CONFIG[key] * 100))}%" for key, label in labels) + + +def curation_threshold_text(lang: str = "zh") -> str: + """LLM prompt 的精选门槛句,数值取自 CONFIG curation_threshold。""" + if lang == "en": + return f"Curation threshold is {CONFIG['curation_threshold']} points" + return f"精选门槛为 {CONFIG['curation_threshold']} 分" + + +def risk_penalty_text(lang: str = "zh") -> str: + """LLM curation_score 的风险扣分规则句。""" + 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}分" diff --git a/backend/app/services/scoring_engine.py b/backend/app/services/scoring_engine.py index 110faae3..3f810344 100644 --- a/backend/app/services/scoring_engine.py +++ b/backend/app/services/scoring_engine.py @@ -8,7 +8,9 @@ 4. Diversity: down-rank when same source/category dominates 5. Risk controls: hard-exclude high-risk items and softly penalize mid-risk items -All tuning constants live in the CONFIG dict at the top for easy adjustment. +All tuning constants live in app.services.scoring_caliber.CONFIG — the single +source of truth shared with the LLM prompts and the DuckDB picks query (#110). +Re-exported here as CONFIG for compatibility with existing importers. """ from __future__ import annotations @@ -23,48 +25,11 @@ ScoringRequestItem, ScoringResponse, ) +from app.services.scoring_caliber import CONFIG # ── Tunable configuration ──────────────────────────────────────────── - -CONFIG = { - # 6-dimension weights (must sum to 1.0) - "w_info_density": 0.25, # 信息密度 - "w_actionability": 0.20, # 可操作性 - "w_creator_value": 0.18, # 创作者价值 - "w_viral_potential": 0.15, # 爆文潜力 - "w_source_authority": 0.12, # 来源权威度 - "w_freshness": 0.10, # 时效新鲜度 - # Source weight mapping (source.weight 1-5 → bonus) - "source_weight_bonus_factor": 6.0, # per weight-unit above 3 - # Time decay - "time_decay_lambda": 0.02, # decay rate: exp(-lambda * hours) - "time_decay_floor": 0.3, # minimum time decay (never fully kill) - # Content quality gates - "quality_gate_min": 45, # below this, content is probably too thin for curation - "quality_gate_strong": 70, # full trust above this composite quality level - "quality_gate_floor": 0.55, # lowest multiplier for weak-but-not-risky content - "min_selected_base_score": 58, # do not select weak items just because a batch is weak - # Source diversity - "diversity_penalty_base": 0.85, # multiplier per same-source duplicate - "category_diversity_penalty_base": 0.92, - "same_source_grace": 1, # first item per source is free - "same_category_grace": 3, # first few items per category are free - # Risk - "risk_threshold": 82, # hard-exclude above this - "risk_soft_start": 45, # risk starts reducing rank after this - "risk_soft_floor": 0.55, # minimum multiplier before hard exclusion - # Curation threshold (minimum final score to be selected) - "curation_threshold": 55, # slightly lower than before, because scores are now more calibrated - # Fallback: when curation_score is 0 but other scores exist - "fallback_use_avg": True, - # ── Percentile-based curation mode ── - "curation_mode": "percentile", # "percentile" | "fixed" - "curation_percentile": 70, # top ~30% selected (P70 and above) → ~25-30% curation rate - # ── User feedback signal ── - "w_feedback": 0.15, # 15% weight for feedback_score - "feedback_score_min": -20.0, # one effective negative vote should not dominate ranking - "feedback_score_max": 20.0, # one effective positive vote should not dominate ranking -} +# 唯一声明处已迁至 app/services/scoring_caliber.py(#110):prompt 教学与 +# picks 查询共享同一份口径,tests/test_scoring_caliber.py 锁死漂移。 # ── Data containers ────────────────────────────────────────────────── @@ -188,10 +153,11 @@ def _compute_base_score(item: ScoringInput) -> tuple[float, dict]: dim_sum = sum(dimensions.values()) - # Blend: if LLM gave a curation_score, weight it 60% + 40% dimension sum - # Otherwise, use pure dimension sum + # Blend: if LLM gave a curation_score, weight it by curation_blend_weight + # (CONFIG, #110) and the rest on the dimension sum; otherwise pure sum if cs > 0: - base = cs * 0.6 + dim_sum * 0.4 + p = cfg["curation_blend_weight"] + base = cs * p + dim_sum * (1 - p) elif cfg["fallback_use_avg"]: # Fallback: average of available scores scores = [ diff --git a/backend/app/services/today_picks.py b/backend/app/services/today_picks.py index d58aa937..6904a617 100644 --- a/backend/app/services/today_picks.py +++ b/backend/app/services/today_picks.py @@ -24,9 +24,9 @@ logger = logging.getLogger(__name__) -# 与 DuckDB query_today_picks 的 weight_bonus 默认值（duckdb_service.py）一致， -# fallback 复刻 adjusted_curation_score 排序时使用。 -_OLTP_FALLBACK_WEIGHT_BONUS = 8 +# 与 DuckDB query_today_picks 共用同一加成口径（scoring_caliber.CONFIG 的 +# picks_weight_bonus，#110），fallback 复刻 adjusted_curation_score 排序时使用。 +_OLTP_FALLBACK_WEIGHT_BONUS = SCORING_CONFIG["picks_weight_bonus"] # 个性化重排的探索保留比例：最终页中预留给「基础分最强但未因个人 # 兴趣入选」条目的比例，防止兴趣匹配完全遮蔽高质量内容。 diff --git a/backend/tests/test_scoring_caliber.py b/backend/tests/test_scoring_caliber.py new file mode 100644 index 00000000..b69d9c4b --- /dev/null +++ b/backend/tests/test_scoring_caliber.py @@ -0,0 +1,143 @@ +"""评分口径单一真相源回归测试(#110)。 + +历史缺陷:同一套精选评分口径分散在 prompt 教学 / 引擎 CONFIG / picks SQL +三处各写一份且互相矛盾(prompt 教五项 30/25/20/15/10 + 门槛 60,引擎六维 +0.25/0.20/0.18/0.15/0.12/0.10 + 门槛 55,picks 硬编码 55/8 不读 CONFIG)。 + +本文件是「漂移即红」防线:从三套 prompt 成品文本里正则提取权重/门槛/扣分 +数字,与 scoring_caliber.CONFIG 断言一致——绕过 caliber 模块手改 prompt +数字、或让引擎/picks 另起一份口径,这里立刻变红。 +""" + +from __future__ import annotations + +import inspect +import re + +from app.services import scoring_engine +from app.services.llm.prompts import analysis as analysis_prompts +from app.services.scoring_caliber import ( + CONFIG, + LLM_RISK_PENALTY_POINTS, + LLM_RISK_PENALTY_TRIGGER, +) + +SIX_DIM_KEYS = ( + "w_info_density", + "w_actionability", + "w_creator_value", + "w_viral_potential", + "w_source_authority", + "w_freshness", +) + + +def _expected_weights() -> list[int]: + return [int(round(CONFIG[k] * 100)) for k in SIX_DIM_KEYS] + + +def _zh_weights(prompt: str) -> list[int]: + seg = re.search(r"加权：(.*?)）", prompt) + assert seg, "prompt 中找不到「加权：…」口径段" + return [int(x) for x in re.findall(r"(\d+)%", seg.group(1))] + + +def _en_weights(prompt: str) -> list[int]: + seg = re.search(r"weighted: (.*?), risk", prompt) + assert seg, "prompt 中找不到 'weighted: …' 口径段" + return [int(x) for x in re.findall(r"(\d+)%", seg.group(1))] + + +# ── CONFIG 自身完整性 ──────────────────────────────────────────────── + + +def test_six_dim_weights_sum_to_one(): + assert abs(sum(CONFIG[k] for k in SIX_DIM_KEYS) - 1.0) < 1e-9 + + +def test_engine_config_is_caliber_reexport(): + """引擎 CONFIG 必须就是 caliber 声明的同一对象,不允许复制成第二份。""" + assert scoring_engine.CONFIG is CONFIG + + +# ── 三套 prompt 教学 vs CONFIG ─────────────────────────────────────── + + +def test_zh_prompt_curation_weights_match_config(): + assert _zh_weights(analysis_prompts.ANALYSIS_PROMPT) == _expected_weights() + + +def test_en_prompt_curation_weights_match_config(): + assert _en_weights(analysis_prompts.ANALYSIS_PROMPT_EN) == _expected_weights() + + +def test_paper_prompt_curation_weights_match_config(): + assert _zh_weights(analysis_prompts.PAPER_ANALYSIS_PROMPT) == _expected_weights() + + +def test_prompts_curation_threshold_match_config(): + expected = CONFIG["curation_threshold"] + zh = re.search(r"精选门槛为 (\d+) 分", analysis_prompts.ANALYSIS_PROMPT) + paper = re.search(r"精选门槛为 (\d+) 分", analysis_prompts.PAPER_ANALYSIS_PROMPT) + en = re.search(r"Curation threshold is (\d+) points", analysis_prompts.ANALYSIS_PROMPT_EN) + assert zh and paper and en, "prompt 中找不到门槛句" + assert int(zh.group(1)) == expected + assert int(paper.group(1)) == expected + assert int(en.group(1)) == expected + + +def test_prompts_risk_penalty_match_caliber(): + for prompt in ( + analysis_prompts.ANALYSIS_PROMPT, + analysis_prompts.PAPER_ANALYSIS_PROMPT, + ): + m = re.search(r"风险分>(\d+)则扣(\d+)分", prompt) + assert m, "中文 prompt 中找不到风险扣分规则" + assert int(m.group(1)) == LLM_RISK_PENALTY_TRIGGER + assert int(m.group(2)) == LLM_RISK_PENALTY_POINTS + m_en = re.search(r"risk>(\d+) deducts (\d+) points", analysis_prompts.ANALYSIS_PROMPT_EN) + assert m_en, "英文 prompt 中找不到风险扣分规则" + assert int(m_en.group(1)) == LLM_RISK_PENALTY_TRIGGER + assert int(m_en.group(2)) == LLM_RISK_PENALTY_POINTS + + +def test_analysis_prompts_still_formatable(): + """prompt 拼接口径片段后仍必须能被 .format(title=, content=) 消费。 + + analysis.py:458 / :160 按此约定调用;拼接若引入未转义的单花括号, + 这里会在运行期之前先红。 + """ + for prompt in ( + analysis_prompts.ANALYSIS_PROMPT, + analysis_prompts.ANALYSIS_PROMPT_EN, + analysis_prompts.PAPER_ANALYSIS_PROMPT, + ): + rendered = prompt.format(title="标题X", content="正文Y") + assert "标题X" in rendered and "正文Y" in rendered + + +# ── picks / today_picks 消费方 ─────────────────────────────────────── + + +def test_picks_mixin_defaults_resolve_from_config(): + """query_today_picks 的门槛与加成默认值必须运行时读 CONFIG,不得硬编码。""" + from app.services._duckdb_picks_mixin import PicksMixin, resolve_picks_params + + sig = inspect.signature(PicksMixin.query_today_picks) + assert sig.parameters["curation_threshold"].default is None + assert sig.parameters["weight_bonus"].default is None + assert sig.parameters["risk_threshold"].default is None + + assert resolve_picks_params(None, None) == ( + float(CONFIG["curation_threshold"]), + float(CONFIG["picks_weight_bonus"]), + ) + # 显式传参按原样透传(today_picks 主链路传 curation_threshold=0 只取候选) + assert resolve_picks_params(0, 5) == (0.0, 5.0) + + +def test_today_picks_fallback_weight_bonus_matches_config(): + """OLTP fallback 的 weight_bonus 必须与 CONFIG 同源,不得再复制一份 8。""" + from app.services.today_picks import _OLTP_FALLBACK_WEIGHT_BONUS + + assert CONFIG["picks_weight_bonus"] == _OLTP_FALLBACK_WEIGHT_BONUS