lambdagent/evals/bench_fc.py

#!/usr/bin/env python3
"""FC vs 文本路径 benchmark — 可复现地对比"原生 function-calling"对真实 agent 质量的影响。

在内核里直接跑(绕过 HTTP 鉴权),对同一 golden 任务、同一 agent、同一模型(qwen-plus),
一次开 nativeToolCalls 一次关,收产物 → default_judge 打分 → 出对比表。改了 FC 一键重测。

用法:
  python3 evals/bench_fc.py [golden/literature-mapper.yaml ...]   # 默认跑 literature-mapper
  MODEL=qwen-plus MAXSTEPS=14 python3 evals/bench_fc.py
要 ~/.agentpaas/.env 里的 DASHSCOPE_API_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"))

# 加载 keys
_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

MODEL = os.environ.get("MODEL", "qwen-plus")
MAXSTEPS = int(os.environ.get("MAXSTEPS", "14"))
DB = os.path.expanduser("~/.agentpaas/data/agentpaas.db")


def _agent_cfg_for(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"])
    # 统一成 qwen-plus + 限步数,公平对比
    cfg["model"] = {"provider": "dashscope", "name": MODEL, "temperature": 0.0, "maxTokens": 4096}
    cfg.setdefault("react", {})
    cfg["react"]["maxSteps"] = MAXSTEPS
    cfg["react"]["toolTimeout"] = 60
    return cfg


def _run_path(cfg: dict, fc: bool, task: EvalTask) -> dict:
    d = tempfile.mkdtemp(prefix="bench_")
    _set_cwd(d); set_sandbox_root(d)
    try:
        c = json.loads(json.dumps(cfg))
        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,
                "_term": term._name}
    finally:
        set_sandbox_root(None); shutil.rmtree(d, ignore_errors=True)


def bench_task(cfg: dict, task: EvalTask) -> dict:
    row = {}
    for label, fc in [("text", False), ("fc", True)]:
        holder = {}

        def wrapped(t, _fc=fc, _h=holder):
            r = _run_path(cfg, _fc, t); _h.update(r); return r
        r = run_eval_task(task, run_fn=wrapped,
                          collect_artifacts_fn=lambda ws, _h=holder: _h.get("_files", {}),
                          judge_fn=default_judge)
        row[label] = {"files_ok": r.files_ok, "judge": round(r.judge_score, 2),
                      "passed": r.passed, "dt": round(holder.get("_dt", 0)), "term": holder.get("_term", "")}
    return row


def main():
    paths = sys.argv[1:] or [os.path.join(os.path.dirname(__file__), "golden", "literature-mapper.yaml")]
    print(f"FC vs 文本 benchmark | model={MODEL} maxSteps={MAXSTEPS}\n")
    allrows = {}
    for p in paths:
        spec = yaml.safe_load(open(p, encoding="utf-8"))
        cfg = _agent_cfg_for(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} ---")
            row = bench_task(cfg, task)
            allrows[task.id] = row
            for lbl in ("text", "fc"):
                r = row[lbl]
                print(f"  {lbl:5} files_ok={r['files_ok']} judge={r['judge']:.2f} "
                      f"passed={r['passed']} {r['dt']}s")
            print()
    print("=== 汇总(judge: 文本 → FC) ===")
    for tid, row in allrows.items():
        print(f"  {tid:20} {row['text']['judge']:.2f} → {row['fc']['judge']:.2f}  "
              f"{'FC↑' if row['fc']['judge'] > row['text']['judge'] else ''}")
    out = os.path.join(os.path.dirname(__file__), "bench_fc_report.json")
    json.dump(allrows, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"\n报告:{out}")


if __name__ == "__main__":
    main()