Agent/examples/process_pipeline/script/llm_evaluate_sources.py

"""
基于 LLM 的 source 知识质量评估

对 source.json 中已匹配的帖子逐条做 rubric 评估，严格参照
test_script/evaluation/知识质量评估-rubric.{json,md} 的 **post 部分**：
先把帖子分到 procedure / step / tool（可多标签），再按"通用 + 命中类型"维度
各打 1-5，最后给出 report / discard 决策。

设计要点：
  - rubric 在运行时从文件加载并整段嵌入 prompt（rubric 改了评估自动跟随）。
  - prompt 各块从 eval_prompt_template.md 加载、按 str.format 填充（template 改了无需改代码）。
  - LLM 输出严格匹配 rubric 的 `post.output` 结构，写回到 source 的 `llm_evaluation` 字段。
  - decision=discard 的帖子从 source.json 移入 filtered_cases.json（filter_reason=llm_discard:...）。
  - 增量：已带 llm_evaluation 的 report 帖跳过；之前已 LLM 拒过的帖（被 extract 重新匹配回来）
    短路直接再剔除，不重复花钱。
  - fail-open：单帖 LLM 调用失败 → 保留该帖并标记，不因瞬时错误丢内容。

入口：evaluate_sources_with_llm(source_file, llm_call, model, requirement, ...) -> stats dict
由 run_pipeline.py 的 source 阶段在规则预筛（extract_sources）之后调用。
"""

import asyncio
import json
import logging
import re
import sys
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

# 直接当脚本跑时（python .../llm_evaluate_sources.py）需要项目根在 sys.path，
# 才能 import examples.* ；被 run_pipeline 作为模块导入时路径已就绪，此处无副作用。
_PROJECT_ROOT = Path(__file__).resolve().parents[3]
if str(_PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(_PROJECT_ROOT))

from examples.process_pipeline.script.llm_helper import call_llm_with_retry

logger = logging.getLogger(__name__)

# prompt 模板 (mod.md 风格的 SYSTEM + USER 两块)：主源放在 script/search_eval/ 下,
# 跟 batch_3forms / server / eval_one_sample 同目录（IDE 编辑那份就是 runtime 实际读的版本）。
# 若主源缺失则退到 test_script/search_eval/ 历史副本。首次加载后缓存。
# 注：原 rubric.json / rubric.md 已固化进本 template，不再外部 load（_RUBRIC_* / load_post_rubric
# / load_rubric_md 一并清掉）。
_PROMPT_TEMPLATE_PATH = (
    Path(__file__).resolve().parent / "search_eval" / "eval_prompt_template.md"
)
if not _PROMPT_TEMPLATE_PATH.exists():
    _PROMPT_TEMPLATE_PATH = (
        Path(__file__).resolve().parent.parent / "test_script" / "search_eval" / "eval_prompt_template.md"
    )
_PROMPT_TEMPLATE_CACHE: Optional[Dict[str, str]] = None

# mod.md 风格的中文 schema 知识类型枚举值(取代了旧 英文 procedure/step/tool)
_VALID_KNOWLEDGE_TYPES = {"工序", "能力", "工具"}
_MAX_BODY_CHARS = 8000      # 控制单帖 prompt token：正文/字幕截断上限
_MAX_COMMENTS = 20          # 评论最多带多少条（喂"评论反馈"维度）
_MAX_COMMENT_CHARS = 200    # 单条评论截断上限

# 上次评估的产物字段——dump source JSON 给 LLM 时必须剥掉,
# 否则 LLM 会"先验 anchoring"到旧分数,新评估失真。
_EVAL_PRODUCT_FIELDS = frozenset({
    "llm_evaluation",   # 之前 LLM 给的整段评估结果（scores/decision/reason/...）
    "images_sent",      # 评估时记录"实际发了几张图"——管线元数据,LLM 评内容无关
})


# ── 评估模型选择（可切换）────────────────────────────────────────────────────────
# key -> (backend, model_id)。backend="qwen" 走 create_qwen_llm_call（阿里云原生），
# 其余走 create_openrouter_llm_call。run_pipeline.py / search_and_evaluate.py 共用此工厂。
EVAL_MODELS: Dict[str, Tuple[str, str]] = {
    "qwen":              ("qwen",       "qwen3.5-plus"),
    "sonnet":            ("openrouter", "claude-sonnet-4-6"),
    "gemini-flash":      ("openrouter", "google/gemini-3-flash-preview"),
    "gemini-flash-lite": ("openrouter", "google/gemini-3.1-flash-lite"),   # 最快，适合大批量评估
    "gemini":            ("openrouter", "google/gemini-3-flash-preview"),  # 别名 → gemini-flash
    "gpt":               ("openrouter", "gpt-5.4"),
}
DEFAULT_EVAL_MODEL = "gemini-flash-lite"


def build_eval_llm_call(choice: str) -> Tuple[Callable, str]:
    """根据选择返回 (llm_call, model_id)。

    choice 可以是 EVAL_MODELS 的 key（qwen/sonnet/gemini/gpt），
    也可以直接传模型 id（含 'qwen' 走 qwen 后端，否则走 OpenRouter）。
    """
    if choice in EVAL_MODELS:
        backend, model_id = EVAL_MODELS[choice]
    else:
        model_id = choice
        backend = "qwen" if "qwen" in choice.lower() else "openrouter"

    if backend == "qwen":
        from agent.llm import create_qwen_llm_call
        return create_qwen_llm_call(model=model_id), model_id
    from agent.llm.openrouter import create_openrouter_llm_call
    return create_openrouter_llm_call(model=model_id), model_id


# ── prompt 模板加载 ─────────────────────────────────────────────────────────────

def load_prompt_template() -> Dict[str, str]:
    """解析 eval_prompt_template.md 成 {BLOCK_NAME: template_str}。

    分隔符 `=== BLOCK_NAME ===` 单独成行 → 下一块开始；第一个分隔符之前的文件头当注释跳过；
    **块内所有行字面保留**（包括 `#` 开头的 markdown 标题）——因为 template 内嵌了完整的
    rubric markdown 和 JSON schema，不能再剥 `#` 行（否则 H1/H2 标题被吃）。
    模板正文里如有 `{var}` 占位符，由 _build_eval_messages 在该块上调 str.format 填充。
    含字面 `{` / `}` 的块（如 USER_RUBRIC_JSON）不能走 .format（会爆 KeyError），代码里
    对它们直接取字面。

    首次加载后缓存在 _PROMPT_TEMPLATE_CACHE；改模板需重启 server。
    解析失败直接抛——prompt 模板是评估链路核心，没它评不了，不做 fail-open。
    """
    global _PROMPT_TEMPLATE_CACHE
    if _PROMPT_TEMPLATE_CACHE is not None:
        return _PROMPT_TEMPLATE_CACHE
    text = _PROMPT_TEMPLATE_PATH.read_text(encoding="utf-8")
    blocks: Dict[str, List[str]] = {}
    current: Optional[str] = None
    for line in text.splitlines():
        m = re.match(r"^===\s+([A-Z_]+)\s+===\s*$", line)
        if m:
            current = m.group(1)
            blocks[current] = []
            continue
        if current is None:
            continue  # 第一个 === 之前是文件头说明，跳过
        blocks[current].append(line)
    # 各块 join 回字符串，strip 掉首尾空行（块之间衔接由调用方加 \n\n）
    _PROMPT_TEMPLATE_CACHE = {k: "\n".join(v).strip("\n") for k, v in blocks.items()}
    return _PROMPT_TEMPLATE_CACHE


# ── 帖子内容格式化（喂给 LLM）──────────────────────────────────────────────────

def _extract_author(post: Dict[str, Any]) -> str:
    return (
        post.get("author")
        or post.get("channel_account_name")
        or post.get("channel")
        or ""
    )


def _extract_comments(source: Dict[str, Any]) -> List[str]:
    """从 source.comments 抽出评论文本，截断条数与长度。"""
    raw = source.get("comments") or []
    out: List[str] = []
    for c in raw[:_MAX_COMMENTS]:
        if isinstance(c, dict):
            text = c.get("content") or c.get("text") or c.get("comment") or ""
        else:
            text = str(c)
        text = (text or "").strip()
        if text:
            out.append(text[:_MAX_COMMENT_CHARS])
    return out


def _format_post_for_eval(source: Dict[str, Any]) -> str:
    """把一条 source 序列化为 JSON 字符串供 LLM 评估。

    现代 LLM 读结构化 JSON 比读"标签:值"自然语言更准——字段名直接告诉它语义,
    不需要靠 prompt 工程把字段标签写漂亮。所以直接 dump source 整段。

    保留两处截断防 token 爆:
      - post.body_text 截到 _MAX_BODY_CHARS（个别帖子正文几万字）
      - comments 数量截到 _MAX_COMMENTS
    其余字段全量给 LLM,由它自行判断哪些有用（如 images URL / channel_account_id 等）。
    """
    # 浅拷贝避免修改调用方的 source
    s = dict(source)
    post = dict(s.get("post") or {})

    body = post.get("body_text") or post.get("desc") or ""
    if isinstance(body, str) and len(body) > _MAX_BODY_CHARS:
        post["body_text"] = body[:_MAX_BODY_CHARS] + f"\n…(正文已截断,原长 {len(body)} 字)"
    s["post"] = post

    comments = s.get("comments") or []
    if isinstance(comments, list) and len(comments) > _MAX_COMMENTS:
        s["comments"] = comments[:_MAX_COMMENTS] + [
            {"_note": f"(评论已截断,共 {len(comments)} 条,只发前 {_MAX_COMMENTS})"}
        ]

    # 剥掉两类不该进 prompt 的字段:
    #   ① `_` 前缀内部字段（如 _quality_grade / _image_data_urls）——管线元数据,无信息量
    #   ② _EVAL_PRODUCT_FIELDS（如 llm_evaluation / images_sent）——上次评估的产物,
    #      喂回新评估会让 LLM"先验 anchoring"到旧分数,新评估失真
    s = {k: v for k, v in s.items()
         if not str(k).startswith("_") and k not in _EVAL_PRODUCT_FIELDS}
    return json.dumps(s, ensure_ascii=False, indent=2)


# ── prompt 构建（从模板各块组装） ─────────────────────────────────────────────

def _build_eval_messages(
    requirement: str,
    post_block: str,
    image_urls: Optional[List[str]] = None,
    query: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """从 eval_prompt_template.md (mod.md 风格) 拼出 system + user message。

    template 只有 SYSTEM + USER 两块——USER 块内含 检索词 / 知识类型分类 / 输出 schema /
    待评估帖子 / 注意事项 的完整 markdown。两个占位符 {query} 和 {post_block} 用 .replace
    替换 (不走 .format,因 schema JSON 含字面 `{`/`}` 会触发 .format KeyError)。

    requirement 仅作为 query 缺失时的降级 fallback。
    """
    tpl = load_prompt_template()
    system = tpl["SYSTEM"]

    # query 优先,requirement 降级,都没就给占位描述
    effective_query = query or requirement or "(未指定检索词)"
    user_text = tpl["USER"].replace("{query}", effective_query).replace("{post_block}", post_block)

    if image_urls:
        # 多模态：把帖子图片随文本一起发给模型(模板末尾已有『请结合配图判断』提示)
        user_content: List[Dict[str, Any]] = [{"type": "text", "text": user_text}]
        for u in image_urls:
            user_content.append({"type": "image_url", "image_url": {"url": u}})
        return [
            {"role": "system", "content": system},
            {"role": "user", "content": user_content},
        ]
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": user_text},
    ]


def _validate_eval(data: Dict[str, Any]) -> Optional[str]:
    """对 LLM 评估输出做最小结构校验(新版 mod.md schema)。返回错误描述或 None。

    校验字段（宽松，避免重试爆炸）:
      - 知识类型:  list, 值 ∈ {工序, 能力, 工具}（步骤→能力 软修复）
      - 相关性:    {和内容制作知识相关, 和 query 相关} 各自 {得分, 理由}
      - 质量:      {固定维度, 动态维度} 至少一个；具体子字段不强校验
    时效性不校验 —— 由 _calc_recency_score 注入, LLM 不该输出。

    脱壳:有些模型会把结果多套一层 {"output": {...}}, 这里就地解包。
    """
    if not isinstance(data, dict):
        return "输出不是 JSON 对象"
    # 脱壳: 有些模型(gemini-flash-lite 等)会照着 schema 多套一层 {"output": {...}}
    if "知识类型" not in data and isinstance(data.get("output"), dict) \
            and "知识类型" in data["output"]:
        inner = data["output"]
        data.clear()
        data.update(inner)

    # 英文 key fallback: LLM 偶发回归英文 schema, 自动 rename 成中文
    _EN_TO_CN = {
        "knowledge_type": "知识类型",
        "relevance":      "相关性",
        "quality":        "质量",
    }
    for en, cn in _EN_TO_CN.items():
        if cn not in data and en in data:
            data[cn] = data.pop(en)

    # 知识类型: 允许为空/缺失/非法; 软修复:
    #   ① 字符串 → list  ② 英文/旧 步骤 → 中文新枚举 (procedure→工序 / step→能力 / 步骤→能力)
    #   ③ 空 list 时反推: 看 质量.动态维度 里 LLM 填了哪些类型子块
    #   ④ 非法值过滤掉
    _KT_REMAP = {
        "procedure": "工序", "step": "能力", "tool": "工具",
        "步骤": "能力",  # 旧 schema 兼容
    }
    kt = data.get("知识类型")
    if isinstance(kt, str):
        kt = [kt]
    if isinstance(kt, list):
        kt = [_KT_REMAP.get(k, k) for k in kt]
        kt = [k for k in kt if k in _VALID_KNOWLEDGE_TYPES]
    else:
        kt = []
    if not kt:
        dyn = ((data.get("质量") or {}).get("动态维度") or {})
        inferred = [t for t in ("工序", "能力", "工具")
                    if isinstance(dyn.get(t), dict) and dyn[t]]
        if inferred:
            kt = inferred
    data["知识类型"] = kt

    # 相关性: 两子键必须各为 {得分, 理由} 对象, 得分 0-10
    rel = data.get("相关性")
    if not isinstance(rel, dict):
        return "相关性 必须是对象"
    for sub_key in ("和内容制作知识相关", "和 query 相关"):
        sub = rel.get(sub_key)
        if not isinstance(sub, dict):
            return f"相关性.{sub_key!r} 必须是 {{得分, 理由}} 对象"
        try:
            v = float(sub.get("得分"))
        except (TypeError, ValueError):
            return f"相关性.{sub_key}.得分 缺失或不是数字 (需 0-10)"
        if not (0 <= v <= 10):
            return f"相关性.{sub_key}.得分 必须在 0-10, 得到 {sub.get('得分')!r}"

    # 质量: 必须是 dict, 至少包含 固定维度 或 动态维度
    q = data.get("质量")
    if not isinstance(q, dict):
        return "质量 必须是对象"
    if not isinstance(q.get("固定维度"), dict) and not isinstance(q.get("动态维度"), dict):
        return "质量 必须至少包含 固定维度 或 动态维度"
    # 子字段不强校验, 由 prompt 引导 + 前端兜底; 这层保 batch 不被重试拖死
    return None


# ── 时效性: 代码计算, 不让 LLM 评 ────────────────────────────────────────────────

def _calc_recency_score(publish_ts: str) -> int:
    """按发布日期 vs 今日的天数差打 0-10 阶梯分:
    ≤30d=10 / ≤90d=8 / ≤180d=6 / ≤270d=3 / ≤365d=1 / >365d=0
    无法解析时间戳 → 0 (保守判过期)。"""
    from datetime import datetime
    try:
        d = datetime.strptime((publish_ts or "")[:10], "%Y-%m-%d")
    except (ValueError, TypeError):
        return 0
    days = (datetime.now() - d).days
    if days <= 30:  return 10
    if days <= 90:  return 8
    if days <= 180: return 6
    if days <= 270: return 3
    if days <= 365: return 1
    return 0


def _inject_recency(data: Dict[str, Any], source: Dict[str, Any]) -> None:
    """LLM 输出后注入时效性到 质量.固定维度.时效性.{得分}, 无 理由 (代码算的不解释)。"""
    if not isinstance(data, dict):
        return
    score = _calc_recency_score((source.get("post") or {}).get("publish_timestamp", ""))
    q = data.setdefault("质量", {})
    if not isinstance(q, dict):
        return
    fixed = q.setdefault("固定维度", {})
    if not isinstance(fixed, dict):
        return
    fixed["时效性"] = {"得分": score}


# ── 单帖评估 ────────────────────────────────────────────────────────────────────

def _source_key(source: Dict[str, Any]) -> Tuple[Any, Any]:
    return (source.get("platform"), source.get("channel_content_id"))


def _move_to_discard(
    source: Dict[str, Any],
    discarded: List[Dict[str, Any]],
    reason: Optional[str],
) -> None:
    """把一条 source 标上 llm_discard 原因并加入 discarded 列表（仅淘汰模式调用）。"""
    s_copy = dict(source)
    r = (reason or "").replace("\n", " ")[:120]
    s_copy["filter_reason"] = f"llm_discard:{r}" if r else "llm_discard"
    discarded.append(s_copy)


async def _evaluate_one(
    source: Dict[str, Any],
    requirement: str,
    llm_call: Callable,
    model: str,
    sem: asyncio.Semaphore,
    image_urls: Optional[List[str]] = None,
    query: Optional[str] = None,
) -> Tuple[Optional[Dict[str, Any]], float]:
    """评估单条 source，返回 (llm_evaluation, cost)。失败返回 (None, cost)。

    image_urls 非空时走多模态评估（把帖子配图一并发给模型，需模型支持图片，如 gemini）。
    query 非空时把它作为检索锚点喂给模型（判相关性看『这帖是否回答了这个检索词』）。
    rubric 详解 / 输出 schema 已固化进 eval_prompt_template.md,本函数不再传 rubric 参数。
    """
    post_block = _format_post_for_eval(source)
    messages = _build_eval_messages(requirement, post_block, image_urls, query)
    async with sem:
        data, cost = await call_llm_with_retry(
            llm_call=llm_call,
            messages=messages,
            model=model,
            temperature=0.1,
            max_tokens=2000,
            validate_fn=_validate_eval,
            task_name=f"LLM-Eval[{source.get('case_id', '?')}]",
        )
    # 时效性走代码注入 (LLM 不评), 写到 质量.固定维度.时效性.{得分}
    _inject_recency(data, source)
    return data, cost


# ── filtered_cases.json 追加 ────────────────────────────────────────────────────

def _append_to_filtered(raw_cases_dir: Path, discarded: List[Dict[str, Any]]) -> None:
    """把 LLM 判为 discard 的帖子并入 filtered_cases.json（与 extract_sources 同结构，按原因分组）。"""
    if not discarded:
        return
    filtered_file = raw_cases_dir / "filtered_cases.json"

    existing: List[Dict[str, Any]] = []
    existing_ids: set = set()
    if filtered_file.exists():
        try:
            with open(filtered_file, "r", encoding="utf-8") as f:
                data = json.load(f)
            for group in data.get("by_reason", {}).values():
                for s in group.get("sources", []):
                    existing.append(s)
                    existing_ids.add((s.get("platform"), s.get("channel_content_id")))
        except Exception as e:
            logger.warning("读取已有 filtered_cases.json 失败: %s", e)

    for s in discarded:
        key = (s.get("platform"), s.get("channel_content_id"))
        if key not in existing_ids:
            existing.append(s)
            existing_ids.add(key)

    by_reason: Dict[str, List[Dict[str, Any]]] = {}
    for s in existing:
        category = (s.get("filter_reason") or "unknown").split(":", 1)[0]
        by_reason.setdefault(category, []).append(s)

    output = {
        "total": len(existing),
        "by_reason": {
            cat: {"count": len(items), "sources": items}
            for cat, items in by_reason.items()
        },
    }
    with open(filtered_file, "w", encoding="utf-8") as f:
        json.dump(output, f, ensure_ascii=False, indent=2)


def _load_prior_llm_discards(raw_cases_dir: Path) -> set:
    """从 filtered_cases.json 收集之前已被 LLM 拒过的帖子 key，避免重复评估花钱。"""
    filtered_file = raw_cases_dir / "filtered_cases.json"
    keys: set = set()
    if not filtered_file.exists():
        return keys
    try:
        with open(filtered_file, "r", encoding="utf-8") as f:
            data = json.load(f)
        for category, group in data.get("by_reason", {}).items():
            if category != "llm_discard":
                continue
            for s in group.get("sources", []):
                keys.add((s.get("platform"), s.get("channel_content_id")))
    except Exception as e:
        logger.warning("读取 filtered_cases.json 失败: %s", e)
    return keys


# ── 主入口 ────────────────────────────────────────────────────────────────────

async def evaluate_sources_with_llm(
    source_file: Path,
    llm_call: Callable,
    model: str,
    requirement: str,
    max_concurrent: int = 3,
    apply_decision: bool = False,
) -> Dict[str, Any]:
    """
    对 source.json 中所有 source 逐条做 LLM rubric 评估，把结果写进每条的 `llm_evaluation` 字段。

    apply_decision（淘汰开关，默认 False —— 当前只标注不淘汰）：
      - False（标注模式）：评估并标注所有帖子，**全部保留在 source.json**，
        不据 decision 剔除、不动 filtered_cases.json。
        （rubric 阈值/权重尚待标定，先采集评分数据，淘汰留待标定后再开。）
      - True（淘汰模式）：decision=discard 的帖子移入 filtered_cases.json 并从 source.json 删除，
        历史已拒帖短路再剔除。将来阈值标定好后由调用方打开此开关即可启用门槛。

    Returns 统计 dict:
      evaluated     —— 本次真正调用 LLM 的条数
      reported      —— decision=report 的总数
      would_discard —— decision=discard 的总数（标注模式下仅统计、不实际剔除）
      discarded     —— 实际从 source.json 移除的条数（标注模式恒为 0）
      skipped       —— 跳过的条数（已评过复用 + 淘汰模式下历史拒帖短路）
      total_cost    —— LLM 调用累计成本
      llm_discard_details —— [{case_id, platform, title, filter_reason}]，淘汰模式下供研究反馈引用
    """
    source_file = Path(source_file)
    raw_cases_dir = source_file.parent

    with open(source_file, "r", encoding="utf-8") as f:
        data = json.load(f)
    sources: List[Dict[str, Any]] = data.get("sources", [])
    if not sources:
        return {
            "evaluated": 0, "reported": 0, "would_discard": 0, "discarded": 0,
            "skipped": 0, "total_cost": 0.0, "llm_discard_details": [],
        }

    # rubric 详解 / 输出 schema 已固化在 eval_prompt_template.md, 不再 load 外部 rubric 文件
    prior_discards = _load_prior_llm_discards(raw_cases_dir) if apply_decision else set()

    kept: List[Dict[str, Any]] = []          # 留在 source.json 的
    discarded: List[Dict[str, Any]] = []     # 移入 filtered 的（仅淘汰模式）
    to_eval: List[Dict[str, Any]] = []       # 需要真正跑 LLM 的
    skipped = 0

    for s in sources:
        existing_eval = s.get("llm_evaluation")
        # 1. 已评过（且非失败标记）→ 直接复用，不重复花钱
        if isinstance(existing_eval, dict) and not existing_eval.get("error"):
            if apply_decision and existing_eval.get("decision") == "discard":
                _move_to_discard(s, discarded, existing_eval.get("reason"))
            else:
                kept.append(s)
            skipped += 1
            continue
        # 2. 淘汰模式下：之前已被 LLM 拒过（被重新匹配回来）→ 短路再剔除
        if apply_decision and _source_key(s) in prior_discards:
            s_copy = dict(s)
            s_copy["filter_reason"] = "llm_discard:previously_rejected"
            discarded.append(s_copy)
            skipped += 1
            continue
        # 3. 需要评估
        to_eval.append(s)

    total_cost = 0.0
    evaluated = 0
    if to_eval:
        sem = asyncio.Semaphore(max_concurrent)
        results = await asyncio.gather(*[
            _evaluate_one(s, requirement, llm_call, model, sem)
            for s in to_eval
        ])
        for s, (llm_eval, cost) in zip(to_eval, results):
            total_cost += cost
            evaluated += 1
            if llm_eval is None:
                # fail-open：评估失败保留该帖，标记便于排查（error=True，下次会重评）
                s["llm_evaluation"] = {"decision": "report", "reason": "llm_eval_failed_kept", "error": True}
                kept.append(s)
                continue
            s["llm_evaluation"] = llm_eval
            is_discard = llm_eval.get("decision") == "discard"
            # 仅淘汰模式才真正移除；标注模式下 discard 帖也留在 source.json
            if apply_decision and is_discard:
                _move_to_discard(s, discarded, llm_eval.get("reason"))
            else:
                kept.append(s)

    # 统计基于最终 kept/discarded 的 llm_evaluation 决策（含本轮新评 + 历史 skip）
    reported = sum(
        1 for s in kept
        if isinstance(s.get("llm_evaluation"), dict)
        and s["llm_evaluation"].get("decision") == "report"
    )
    would_discard = sum(
        1 for s in (kept + discarded)
        if isinstance(s.get("llm_evaluation"), dict)
        and s["llm_evaluation"].get("decision") == "discard"
    )

    # 写回 source.json
    data["sources"] = kept
    data["total"] = len(kept)
    with open(source_file, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

    # 仅淘汰模式：并入 filtered_cases.json
    if apply_decision and discarded:
        _append_to_filtered(raw_cases_dir, discarded)

    # 给研究反馈用的摘要（仅淘汰模式有实际剔除）
    llm_discard_details: List[Dict[str, Any]] = []
    for s in discarded:
        post = s.get("post", {}) or {}
        title = post.get("title") or s.get("source_url", "")
        llm_discard_details.append({
            "case_id": s.get("case_id", ""),
            "platform": s.get("platform", ""),
            "title": title[:60] if title else "",
            "filter_reason": s.get("filter_reason", ""),
        })

    return {
        "evaluated": evaluated,
        "reported": reported,
        "would_discard": would_discard,
        "discarded": len(discarded),
        "skipped": skipped,
        "total_cost": round(total_cost, 4),
        "llm_discard_details": llm_discard_details,
    }


# ── CLI ────────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse

    from dotenv import load_dotenv
    load_dotenv()

    parser = argparse.ArgumentParser(description="对 source.json 做 LLM rubric 评估")
    parser.add_argument("source_file", type=Path, help="source.json 路径")
    parser.add_argument("--requirement", type=str, default="", help="采集需求 / 目标格子描述")
    parser.add_argument("--model", type=str, default=DEFAULT_EVAL_MODEL,
                        help=f"评估模型，可选 {list(EVAL_MODELS)} 或直接传模型 id（默认 {DEFAULT_EVAL_MODEL}）")
    parser.add_argument("--max-concurrent", type=int, default=3)
    parser.add_argument("--apply-decision", action="store_true",
                        help="按 LLM decision 实际淘汰 discard 帖（默认只标注不淘汰，阈值标定后再开）")
    args = parser.parse_args()

    _llm_call, _model_id = build_eval_llm_call(args.model)
    print(f"[eval-model] {args.model} -> {_model_id}")

    stats = asyncio.run(evaluate_sources_with_llm(
        source_file=args.source_file,
        llm_call=_llm_call,
        model=_model_id,
        requirement=args.requirement,
        max_concurrent=args.max_concurrent,
        apply_decision=args.apply_decision,
    ))
    print(json.dumps(stats, ensure_ascii=False, indent=2))