Agent/examples/mode_workflow/pipeline/search_eval.py

# -*- coding: utf-8 -*-
"""搜索 + 评估 · 任意 query → 多渠道搜索去重 → LLM 逐帖评估 → search_process/search_tools 表
(方向由 --mode-type 决定:工序 → search_process,工具 → search_tools)
================================================================================
引擎函数全部只读复用 search_and_evaluate.py(搜索/去重/转写/评估/英平台翻译)。

用法(一般由 server.py 起子进程调):
  python pipeline/search_eval.py --query-id q0004 --query "AI 人像 图片 生成 怎么做"
  python pipeline/search_eval.py --query-id q0005 --query "GPT image2 评测" \
      --synonyms "GPT image2 测评,GPT image2 实测" --platforms xhs,gzh --max-count 10
"""
import argparse
import asyncio
import copy
import json
import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[3]   # …/Agent
sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
load_dotenv()

from examples.process_pipeline.script.search_eval.search_and_evaluate import (
    search_all, evaluate_posts, transcribe_video_posts, build_query_overrides,
)
from examples.process_pipeline.script.llm_evaluate_sources import (
    build_eval_llm_call, EVAL_MODELS, DEFAULT_EVAL_MODEL, _format_post_for_eval,
)
from examples.process_pipeline.script.llm_helper import call_llm_with_retry

HERE = Path(__file__).resolve().parent
MW = HERE.parent
sys.path.insert(0, str(MW))
import db


async def _rescore_query_relevance(source, query, llm_call, model, sem):
    """评估去重的轻量重算:只判「和 query 相关」{得分,理由}。
    纯文本(不带图)、低 token —— 比整套多模态评估便宜得多。返回 (dict|None, cost)。"""
    post_block = _format_post_for_eval(source)
    system = ("你是内容评估器。只判断这篇帖子与给定【检索词】的相关程度——"
              "即「这帖是否回答/命中了这个检索词」。严格只输出一个 JSON 对象,"
              '形如 {"得分": <0到10的数字>, "理由": "<一句话>"},不要任何额外字段或解释。')
    user = f"【检索词】{query}\n\n【帖子】\n{post_block}"
    messages = [{"role": "system", "content": system},
                {"role": "user", "content": user}]

    def _v(d):
        if not isinstance(d, dict):
            return "需 JSON 对象"
        try:
            v = float(d.get("得分"))
        except (TypeError, ValueError):
            return "得分 缺失或非数字"
        return None if 0 <= v <= 10 else "得分需在 0-10"

    async with sem:
        data, cost = await call_llm_with_retry(
            llm_call=llm_call, messages=messages, model=model,
            temperature=0.1, max_tokens=300, validate_fn=_v,
            task_name=f"QueryRel[{source.get('case_id', '?')}]")
    return data, cost


async def run(args):
    phrasings = [args.query] + [s.strip() for s in (args.synonyms or "").split(",") if s.strip()]
    # 去重保序
    seen, uniq = set(), []
    for q in phrasings:
        if q not in seen:
            seen.add(q); uniq.append(q)
    phrasings = uniq
    platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]

    eval_llm, eval_model_id = build_eval_llm_call(args.eval_model)
    print(f"▶ {args.query_id}  query={args.query!r}  措辞={phrasings}  渠道={platforms}")

    overrides = await build_query_overrides(platforms, phrasings, eval_llm, eval_model_id)
    sources = await search_all(platforms, phrasings, args.max_count,
                               args.max_concurrent, query_overrides=overrides)
    print(f"🔎 去重后 {len(sources)} 帖")
    if not sources:
        print("❌ 搜索无结果"); return 1

    try:
        from examples.process_pipeline.script.extract_sources import _convert_timestamps
        _convert_timestamps(sources)
    except Exception:
        pass

    if not args.no_transcribe:
        n = await transcribe_video_posts(sources, concurrency=args.max_concurrent)
        if n:
            print(f"🎙️  视频转写 {n} 条")

    table = "search_tools" if args.mode_type == "工具" else "search_process"

    # ── 评估去重 ────────────────────────────────────────────────────────────────
    # 评估的相关性含两子项:「和内容制作知识相关」(与 query 无关)与「和 query 相关」
    # (query 专属)。同帖在别的相似 query 下评过时,质量/通用相关/时效等 query 无关分
    # 可直接复用,只需用一次轻量纯文本调用重算「和 query 相关」,免去整套多模态评估,省钱。
    # --force-eval 跳过去重,全部走完整评估。
    cost = 0.0
    if not args.no_eval:
        prior = {}
        if not args.force_eval:
            for s in sources:
                e = db.fetch_existing_eval(s["case_id"], table)
                if e:
                    prior[s["case_id"]] = e
        fresh = [s for s in sources if s["case_id"] not in prior]
        reused = [s for s in sources if s["case_id"] in prior]
        if reused:
            print(f"♻️ 评估去重:{len(reused)} 帖已评过 → 复用通用分+重算 query 相关分;"
                  f"{len(fresh)} 帖走完整评估")

        if fresh:
            esc_llm = esc_model = None
            if args.escalate_model:
                esc_llm, esc_model = build_eval_llm_call(args.escalate_model)
                print(f"⬆️  启用模糊带升级:{eval_model_id} 初评 → {esc_model} "
                      f"复核(带 [{args.escalate_band[0]:g},{args.escalate_band[1]:g}])")
            _, c = await evaluate_posts(
                fresh, "", eval_llm, eval_model_id, args.max_concurrent,
                include_images=not args.no_images, max_images=args.max_images,
                image_mode=args.image_mode, query=args.query,
                escalate_llm=esc_llm, escalate_model=esc_model,
                escalate_band=tuple(args.escalate_band),
            )  # evaluate_posts 就地把 llm_evaluation 挂到各 source 上
            cost += c

        if reused:
            sem = asyncio.Semaphore(args.max_concurrent)
            rr = await asyncio.gather(*[
                _rescore_query_relevance(s, args.query, eval_llm, eval_model_id, sem)
                for s in reused])
            for s, (qr, c) in zip(reused, rr):
                cost += c
                blob = copy.deepcopy(prior[s["case_id"]])
                if qr is not None:   # 重算成功才覆盖,失败则沿用旧 query 相关分
                    blob.setdefault("相关性", {})["和 query 相关"] = {
                        "得分": qr.get("得分"), "理由": qr.get("理由", "")}
                s["llm_evaluation"] = blob
                qr_s = (blob.get("相关性", {}).get("和 query 相关") or {}).get("得分", "?")
                print(f"   ♻️ [query={qr_s}] {s['case_id'][:24]}")
    for s in sources:
        s.pop("_image_data_urls", None)

    n = db.upsert_search_posts(args.query_id, args.query, sources, table=table)
    print(f"🗄️  {table} 入库 {n} 行 · 方向 {args.mode_type or '工序'} · 评估成本 ${cost:.4f}")

    out_dir = MW / "runs" / table
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / f"{args.query_id}.json").write_text(json.dumps({
        "query_id": args.query_id, "query": args.query, "phrasings": phrasings,
        "platforms": platforms, "total": len(sources), "results": sources,
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    return 0


def main():
    p = argparse.ArgumentParser(description="搜索+评估 → search_process/search_tools")
    p.add_argument("--query-id", required=True, help="如 q0004(server 自动分配)")
    p.add_argument("--query", required=True, help="基准 query(评估锚点)")
    p.add_argument("--synonyms", default="", help="逗号分隔的同义措辞(可选)")
    p.add_argument("--mode-type", default="", choices=["", "工序", "工具"],
                   help="解构方向,决定写哪张表(工具 → search_tools;其余 → search_process)")
    p.add_argument("--platforms", default="xhs,gzh")
    p.add_argument("--max-count", type=int, default=10)
    p.add_argument("--eval-model", default=DEFAULT_EVAL_MODEL, choices=list(EVAL_MODELS))
    p.add_argument("--escalate-model", default="", choices=[""] + list(EVAL_MODELS),
                   help="模糊带升级用的强模型(如 sonnet);留空=不升级。初评落在 --escalate-band "
                        "的可复现性/意图可控性帖子交此模型复核")
    p.add_argument("--escalate-band", type=float, nargs=2, default=[4.0, 6.0],
                   metavar=("LO", "HI"), help="升级触发的闭区间,默认 4 6")
    p.add_argument("--max-concurrent", type=int, default=3)
    p.add_argument("--max-images", type=int, default=4)
    p.add_argument("--image-mode", choices=["url", "base64"], default="url")
    p.add_argument("--no-transcribe", action="store_true")
    p.add_argument("--no-eval", action="store_true")
    p.add_argument("--no-images", action="store_true")
    p.add_argument("--force-eval", action="store_true",
                   help="跳过评估去重,所有帖都走完整评估(换 prompt/模型对比时用)")
    args = p.parse_args()
    raise SystemExit(asyncio.run(run(args)))


if __name__ == "__main__":
    main()