lambdagent/evals/bench_providers.py

#!/usr/bin/env python3
"""多 provider / 车道扫描 benchmark — 用数据决定每个 agent 该用哪条执行路径。

同一 golden 任务,自动跑一组 (provider, model, nativeToolCalls) 组合,收产物 →
default_judge 打分 → 出对比表。回答"这个 agent 用 claude-code-文本 / qwen-文本 /
qwen-FC 哪个质量高"——而非拍脑袋。

纠正一个常见误判:非 workspace 的 claude-code agent **不走 native 车道**(那只给
workspace.assistant),而是 react-over-text + claude-code 当纯文本 provider —— 最脆的路。
所以"换 qwen-FC 是降级"未必成立,要扫描才知道。

用法:
  python3 evals/bench_providers.py [golden/literature-mapper.yaml]
  COMBOS="claude-code:sonnet:0,dashscope:qwen-plus:0,dashscope:qwen-plus:1" python3 evals/bench_providers.py
要 ~/.agentpaas/.env(各 provider key)+ 本地 ollama(judge)。真调 LLM、花钱。
"""
from __future__ import annotations

import json
import os
import shutil
import sys
import tempfile
import time

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "agentpaas", "src"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "lambdagent", "src"))

_envf = os.path.expanduser("~/.agentpaas/.env")
if os.path.exists(_envf):
    for line in open(_envf):
        if "=" in line and not line.strip().startswith("#"):
            k, v = line.strip().split("=", 1)
            os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))

import sqlite3                                              # noqa: E402
import yaml                                                # noqa: E402
from lambdagent.fromconfig import from_config              # noqa: E402
from lambdagent.core import Context                        # noqa: E402
from lambdagent.builtin_tools.shell_tools import _set_cwd  # noqa: E402
from lambdagent.builtin_tools._sandbox import set_sandbox_root  # noqa: E402
from agentpaas.engine.agent_eval import EvalTask, run_eval_task  # noqa: E402
from agentpaas.engine.pipeline import default_judge        # noqa: E402

DB = os.path.expanduser("~/.agentpaas/data/agentpaas.db")
MAXSTEPS = int(os.environ.get("MAXSTEPS", "14"))
RUNS = int(os.environ.get("RUNS", "1"))   # 每个组合重复跑几次取均值(消 LLM 方差)


def _agg(results: list) -> dict:
    """对同一组合的 N 次 EvalResult 聚合:均值 + 极值 + 达标率。"""
    n = len(results) or 1
    js = [r.judge_score for r in results]
    return {
        "runs": len(results),
        "judge_avg": round(sum(js) / n, 2),
        "judge_min": round(min(js), 2) if js else 0.0,
        "judge_max": round(max(js), 2) if js else 0.0,
        "files_ok_rate": round(sum(1 for r in results if r.files_ok) / n, 2),
        "pass_rate": round(sum(1 for r in results if r.passed) / n, 2),
    }

# 组合:provider:model:fc(1/0)。默认 claude-code文本 vs qwen文本 vs qwen-FC。
_DEFAULT = "claude-code:sonnet:0,dashscope:qwen-plus:0,dashscope:qwen-plus:1"
COMBOS = [tuple(c.split(":")) for c in os.environ.get("COMBOS", _DEFAULT).split(",")]


def _base_cfg(template: str) -> dict:
    con = sqlite3.connect(DB); con.row_factory = sqlite3.Row
    a = con.execute("SELECT id,current_version FROM agents WHERE agent_template=? "
                    "AND status='active' LIMIT 1", (template,)).fetchone()
    if not a:
        raise RuntimeError(f"没有 {template} 的已安装实例")
    cfg = json.loads(con.execute("SELECT config FROM agent_versions WHERE agent_id=? AND version=?",
                                 (a["id"], a["current_version"])).fetchone()["config"])
    cfg.setdefault("react", {})["maxSteps"] = MAXSTEPS
    cfg["react"]["toolTimeout"] = 90
    return cfg


def _run_combo(base: dict, provider: str, model: str, fc: bool, task: EvalTask) -> dict:
    d = tempfile.mkdtemp(prefix="bench_"); _set_cwd(d); set_sandbox_root(d)
    try:
        c = json.loads(json.dumps(base))
        c["model"] = {"provider": provider, "name": model, "temperature": 0.0, "maxTokens": 4096}
        c["react"]["nativeToolCalls"] = fc
        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as f:
            yaml.dump(c, f, allow_unicode=True); p = f.name
        term = from_config(p); os.unlink(p)
        ctx = Context(workspace_path=d, run_id="bench")
        t0 = time.time(); out = term.apply(task.input, ctx); dt = time.time() - t0
        files = {fn: open(os.path.join(d, fn), encoding="utf-8", errors="ignore").read()
                 for fn in os.listdir(d) if os.path.isfile(os.path.join(d, fn))}
        return {"output": str(out), "steps": len(getattr(ctx, "trace", []) or []),
                "cost_usd": 0.0, "workspace_path": d, "_files": files, "_dt": dt}
    finally:
        set_sandbox_root(None); shutil.rmtree(d, ignore_errors=True)


def main():
    paths = sys.argv[1:] or [os.path.join(os.path.dirname(__file__), "golden", "literature-mapper.yaml")]
    print(f"provider/车道扫描 | maxSteps={MAXSTEPS}")
    print(f"组合: {['/'.join(c) for c in COMBOS]}\n")
    report = {}
    for p in paths:
        spec = yaml.safe_load(open(p, encoding="utf-8"))
        base = _base_cfg(spec["agent_template"])
        for t in spec.get("tasks", []):
            task = EvalTask(id=t["id"], input=t.get("input", ""),
                            must_produce=t.get("must_produce", []),
                            rubric=t.get("rubric", ""), threshold=float(t.get("threshold", 0.6)))
            print(f"=== {spec['agent_template']} / {task.id} ===")
            rows = {}
            for provider, model, fcs in COMBOS:
                fc = fcs in ("1", "true", "True")
                label = f"{provider}/{model}{'·FC' if fc else '·文本'}"
                run_results = []
                for run_i in range(RUNS):
                    holder = {}

                    def wrapped(tk, _p=provider, _m=model, _fc=fc, _h=holder):
                        r = _run_combo(base, _p, _m, _fc, tk); _h.update(r); return r
                    try:
                        rr = run_eval_task(task, run_fn=wrapped,
                                           collect_artifacts_fn=lambda ws, _h=holder: _h.get("_files", {}),
                                           judge_fn=default_judge)
                        run_results.append(rr)
                        print(f"  {label:26} run{run_i+1}/{RUNS} judge={rr.judge_score:.2f} "
                              f"files_ok={rr.files_ok} {round(holder.get('_dt',0))}s"
                              + (f" err:{rr.error[:30]}" if rr.error else ""))
                    except Exception as e:
                        print(f"  {label:26} run{run_i+1}/{RUNS} ✗ {str(e)[:50]}")
                if run_results:
                    rows[label] = _agg(run_results)
            report[task.id] = rows
            print(f"  --- {task.id} 聚合({RUNS} 轮均值)---")
            for lbl, a in sorted(rows.items(), key=lambda kv: -kv[1]["judge_avg"]):
                print(f"    {lbl:26} judge均={a['judge_avg']:.2f} "
                      f"[{a['judge_min']:.2f}~{a['judge_max']:.2f}] "
                      f"落盘率={a['files_ok_rate']:.0%} 通过率={a['pass_rate']:.0%}")
            if rows:
                best = max(rows.items(), key=lambda kv: kv[1]["judge_avg"])
                print(f"  → 最优: {best[0]}  (judge均 {best[1]['judge_avg']:.2f})\n")
    out = os.path.join(os.path.dirname(__file__), "bench_providers_report.json")
    json.dump(report, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"报告: {out}")


if __name__ == "__main__":
    main()