Blame for scripts/eval.py

"""评测：对 eval/cases.jsonl 逐条跑 Agent，按规则打分（关键词命中 / 工具调用正确 / JSON 字段正确）。
用法：uv run python scripts/eval.py [--no-tools] [--no-rag] [--limit N] [--cases eval/stress.jsonl] [--out result.jsonl]
"""
from __future__ import annotations

import argparse
import asyncio
import json
import re
import time
from pathlib import Path

from logicllm import config
from logicllm.agent.core import LogisticsAgent

CASES = config.ROOT / "eval" / "cases.jsonl"


def score(case: dict, answer: str, steps: list) -> tuple[float, str]:
    tools_called = [s.content["name"] for s in steps if s.kind == "tool_call"]
    notes = []
    ok = 1.0
    if "expect_tool" in case:
        if case["expect_tool"] not in tools_called:
            ok = 0.0; notes.append(f"未调用 {case['expect_tool']} (实际 {tools_called})")
    if "expect_tools" in case:  # 多个工具都必须被调用
        miss_t = [t for t in case["expect_tools"] if t not in tools_called]
        if miss_t:
            ok = 0.0; notes.append(f"未调用 {miss_t} (实际 {tools_called})")
    if "expect_no_tool" in case and tools_called:
        ok = 0.0; notes.append(f"不应调用工具，实际 {tools_called}")
    norm = answer.replace(" ", "")
    if "keywords" in case:
        miss = [k for k in case["keywords"] if k.replace(" ", "") not in norm]
        if miss:
            ok = min(ok, 1 - len(miss) / len(case["keywords"])); notes.append(f"缺关键词 {miss}")
    if "lead" in case:  # 结论必须出现在回答开头（前 20 字），防止"4 天……不是 3 天"这类对照句被 forbid 误判、也防止先错后对
        head = norm[:20]
        if not any(k.replace(" ", "") in head for k in case["lead"]):
            ok = 0.0; notes.append(f"开头未给出结论 {case['lead']}（实际开头「{head}」）")
    if "bold" in case:  # 结论必须出现在加粗段（**…**）里：数值套档题让模型先写比较过程、再加粗给结论，评分只看结论
        bolds = "".join(re.findall(r"\*\*(.+?)\*\*", norm))
        if not any(k.replace(" ", "") in bolds for k in case["bold"]):
            ok = 0.0; notes.append(f"加粗结论未命中 {case['bold']}（实际加粗「{bolds[:30]}」）")
    if "keywords_any" in case:  # 命中其一即可
        if not any(k.replace(" ", "") in norm for k in case["keywords_any"]):
            ok = 0.0; notes.append(f"任一关键词均未命中 {case['keywords_any']}")
    if "forbid" in case:  # 出现即零分（用于臆测/编造检查）
        hit = [k for k in case["forbid"] if k.replace(" ", "") in norm]
        if hit:
            ok = 0.0; notes.append(f"出现禁用词 {hit}")
    if "expect_json" in case:
        m = re.search(r"\{.*\}", answer, re.S)
        try:
            got = json.loads(m.group(0)) if m else {}
        except json.JSONDecodeError:
            got = {}
        exp = case["expect_json"]
        hit = sum(1 for k, v in exp.items() if str(got.get(k, "")).strip() == str(v))
        ok = min(ok, hit / len(exp)); notes.append(f"JSON 字段 {hit}/{len(exp)}")
    return ok, "; ".join(notes)


async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--no-tools", action="store_true")
    ap.add_argument("--no-rag", action="store_true")
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--cases", default=str(CASES), help="题库文件，默认 eval/cases.jsonl")
    ap.add_argument("--out", default="", help="把每题的回答/工具调用/得分写入 jsonl，便于分析失败题")
    a = ap.parse_args()
    cases = [json.loads(l) for l in Path(a.cases).read_text(encoding="utf-8").splitlines()
             if l.strip() and not l.strip().startswith("//")]
    if a.limit:
        cases = cases[:a.limit]
    agent = LogisticsAgent(use_tools=not a.no_tools, use_rag=not a.no_rag)
    print(f"后端 {config.LLM_BACKEND}:{config.LLM_MODEL}  工具={not a.no_tools} RAG={not a.no_rag}  用例 {len(cases)}\n")
    total, by_cat, records = 0.0, {}, []
    t0 = time.time()
    for i, c in enumerate(cases, 1):
        try:
            turns = c.get("turns") or [c["q"]]
            history, steps_all = [], []
            for t in turns:
                history.append({"role": "user", "content": t})
                res = await agent.run(history)
                history.append({"role": "assistant", "content": res.answer})
                steps_all.extend(res.steps)
            s, note = score(c, res.answer, steps_all)
        except Exception as e:  # noqa: BLE001
            s, note, res = 0.0, f"异常 {e}", None
        total += s
        by_cat.setdefault(c["cat"], []).append(s)
        flag = "✅" if s >= 0.99 else ("🟡" if s > 0 else "❌")
        qshow = c.get("q") or " ▸ ".join(c.get("turns", []))
        print(f"{flag} [{c['cat']}] {qshow[:44]}  {s:.2f}  {note}")
        records.append({"case": c, "score": s, "note": note, "answer": res.answer if res else "",
                        "tool_calls": [st.content for st in steps_all if st.kind == "tool_call"] if res else []})
    print(f"\n总分 {total / len(cases):.3f}  用时 {time.time() - t0:.0f}s")
    for k, v in by_cat.items():
        print(f"  {k}: {sum(v) / len(v):.3f} ({len(v)})")
    if a.out:
        Path(a.out).write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in records) + "\n", encoding="utf-8")
        print(f"明细已写入 {a.out}")
    await agent.client.aclose()


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