Agent/examples/mode_workflow/_reeval_one.py

# -*- coding: utf-8 -*-
"""一次性:用当前 eval_prompt_template.md 对单条已存帖子重评(复用生产评估链路 evaluate_posts)。
支持 --escalate-model 演示 sonnet+flash-lite 组合(模糊带升级)。"""
import argparse, asyncio, json, sys
from datetime import datetime
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()

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

from examples.process_pipeline.script.search_eval.search_and_evaluate import evaluate_posts
from examples.process_pipeline.script.llm_evaluate_sources import (
    _EVAL_PRODUCT_FIELDS, build_eval_llm_call, DEFAULT_EVAL_MODEL,
)


def _load(query_id):
    return json.loads((MW / "runs" / "search_process" / f"{query_id}.json")
                      .read_text(encoding="utf-8"))


def _save(query_id, data):
    (MW / "runs" / "search_process" / f"{query_id}.json").write_text(
        json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--query-id", required=True)
    ap.add_argument("--case-id", required=True)
    ap.add_argument("--query", default="")
    ap.add_argument("--model", default=DEFAULT_EVAL_MODEL)
    ap.add_argument("--escalate-model", default="")
    ap.add_argument("--escalate-band", type=float, nargs=2, default=[4.0, 6.0])
    ap.add_argument("--max-images", type=int, default=4)
    ap.add_argument("--persist", action="store_true",
                    help="把新评估写回 DB(overall_score/knowledge_type/llm_evaluation),落库前先备份旧值")
    a = ap.parse_args()

    data = _load(a.query_id)
    query = a.query or data.get("query", "")
    src = next((s for s in data.get("results", []) if s.get("case_id") == a.case_id), None)
    if not src:
        raise SystemExit(f"未找到 case_id={a.case_id}")
    for k in _EVAL_PRODUCT_FIELDS:
        src.pop(k, None)

    llm_call, model_id = build_eval_llm_call(a.model)
    esc_llm = esc_model = None
    if a.escalate_model:
        esc_llm, esc_model = build_eval_llm_call(a.escalate_model)
    print(f"▶ 重评 {a.case_id}  初评={model_id}"
          + (f"  升级={esc_model} 带[{a.escalate_band[0]:g},{a.escalate_band[1]:g}]" if esc_model else "")
          + f"  query={query!r}\n")

    sources, cost = await evaluate_posts(
        [src], "", llm_call, model_id, max_concurrent=1,
        include_images=True, max_images=a.max_images, image_mode="url", query=query,
        escalate_llm=esc_llm, escalate_model=esc_model, escalate_band=tuple(a.escalate_band),
    )
    ev = sources[0]["llm_evaluation"]
    overall = db.overall_score(ev)
    pub = (src.get("post") or {}).get("publish_timestamp", "")
    adopted = db.is_adopted(overall, ev, pub)

    print("\n" + "=" * 60)
    print(f"最终评估模型 = {sources[0].get('_escalated') or model_id}")
    print(f"综合分(overall_score) = {overall}")
    print(f"  · 和内容制作知识相关 = {((ev.get('相关性') or {}).get('和内容制作知识相关') or {}).get('得分')}")
    print(f"  · 实现完整性/可复现门槛 = {db._repro_score(ev)}   (门槛 <4 → 不采纳)")
    print(f"  · 意图可控性        = {db._fixed_dim_score(ev, '意图可控性')}  (暂只采分)")
    print(f"采纳判定(is_adopted)  = {adopted}")
    print(f"总成本 ≈ ${cost:.4f}")

    if a.persist:
        if not isinstance(ev, dict) or ev.get("_error"):
            raise SystemExit("评估结果异常(_error),拒绝落库")
        # 1) 备份旧 DB 行(overall_score/knowledge_type/llm_evaluation/publish_time)
        old = next((p for p in db.fetch_posts(a.query_id, "process")
                    if p["case_id"] == a.case_id), None)
        if old is None:
            raise SystemExit(f"DB 无此行,无法落库: query={a.query_id} case={a.case_id}")
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        bpath = (MW / "runs" / "search_process"
                 / f"{a.query_id}.{a.case_id}.score_backup.{ts}.json")
        bpath.write_text(json.dumps({
            "query_id": a.query_id, "case_id": a.case_id,
            "old_overall_score": old.get("overall_score"),
            "old_knowledge_type": old.get("knowledge_type"),
            "old_llm_evaluation": old.get("llm_evaluation"),
            "old_adopted": old.get("adopted"),
        }, ensure_ascii=False, indent=2), encoding="utf-8")
        # 2) 写回 DB(派生列 overall_score/knowledge_type 由 update_post_eval 重算)
        n = db.update_post_eval(a.query_id, a.case_id, ev, table="search_process")
        # 3) 同步 runs json,保持后续重评输入一致
        src["llm_evaluation"] = ev
        _save(a.query_id, data)
        print(f"\n💾 旧值已备份 → {bpath.name}")
        print(f"✅ DB 已更新 {n} 行(overall={overall} 采纳={adopted})")


if __name__ == "__main__":
    asyncio.run(main())