#!/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()