lambdagent/evals/run_evals.py

#!/usr/bin/env python3
"""第一方 agent 行为回归 — live 评估 runner(手动/nightly,要真 LLM、花钱)。

用法:
  AGENTPAAS_API_KEY=<key> python3 evals/run_evals.py [golden/xxx.yaml ...]
不传文件则跑 evals/golden/ 下全部。结果打印汇总 + 落 evals/last_report.json。

框架(打分逻辑)在 agentpaas.engine.agent_eval(已单测,进 CI);本 runner 只负责
把真实"跑 agent / 收产物 / judge"接进去。judge 复用 pipeline.default_judge。
"""
from __future__ import annotations

import glob
import json
import os
import sys
import time
import urllib.request

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"))

from agentpaas.engine.agent_eval import EvalTask, run_eval_task, summarize  # noqa: E402

BASE = os.environ.get("AGENTPAAS_URL", "http://127.0.0.1:8000")
KEY = os.environ.get("AGENTPAAS_API_KEY", "")


def _api(method, path, body=None):
    url = f"{BASE}{path}"
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Content-Type": "application/json",
                                          "Authorization": f"Bearer {KEY}"})
    with urllib.request.urlopen(req, timeout=1800) as r:
        return json.loads(r.read())


def _resolve_agent_id(template: str) -> str:
    """按模板找一个已安装实例(取第一个 active)。"""
    agents = _api("GET", "/api/v1/agents").get("agents", [])
    for a in agents:
        if a.get("agent_template") == template and a.get("status", "active") == "active":
            return a["id"]
    raise RuntimeError(f"未找到模板 {template} 的已安装实例")


def _run_fn(task: EvalTask) -> dict:
    aid = task.agent_id or _resolve_agent_id(task.agent_template)
    # 同步 /run:返回时已完成
    res = _api("POST", f"/api/v1/agents/{aid}/run", {"input": task.input})
    return {
        "output": res.get("output", ""),
        "steps": (res.get("usage") or {}).get("steps", res.get("steps", 0)),
        "cost_usd": res.get("cost_usd", 0.0),
        "workspace_path": res.get("workspace_path", ""),
        "_run_id": res.get("id", ""), "_agent_id": aid,
    }


def _collect_fn(workspace_path: str) -> dict:
    # 直接读磁盘(workspace_path 是本机绝对路径)
    files = {}
    if not workspace_path or not os.path.isdir(workspace_path):
        return files
    skip = {"config.yml", "cost.json", "input.json", "manifest.json", "output.json", "trace.json"}
    for dp, _dn, fn in os.walk(workspace_path):
        for f in fn:
            if f in skip or f.startswith("."):
                continue
            full = os.path.join(dp, f)
            try:
                with open(full, encoding="utf-8", errors="ignore") as fh:
                    files[os.path.relpath(full, workspace_path)] = fh.read()
            except Exception:
                pass
    return files


def _judge_fn(req):
    from agentpaas.engine.pipeline import default_judge
    return default_judge(req)


def _load_tasks(paths):
    import yaml
    tasks = []
    for p in paths:
        spec = yaml.safe_load(open(p, encoding="utf-8"))
        tmpl = spec.get("agent_template", "")
        for t in spec.get("tasks", []):
            tasks.append(EvalTask(
                id=t["id"], agent_template=tmpl, agent_id=t.get("agent_id", ""),
                input=t.get("input", ""), must_produce=t.get("must_produce", []),
                rubric=t.get("rubric", ""), threshold=float(t.get("threshold", 0.6))))
    return tasks


def main():
    if not KEY:
        print("✗ 需要 AGENTPAAS_API_KEY 环境变量(同前端用的那把)。"); return 2
    paths = sys.argv[1:] or sorted(glob.glob(os.path.join(os.path.dirname(__file__), "golden", "*.yaml")))
    tasks = _load_tasks(paths)
    print(f"跑 {len(tasks)} 个 golden 任务 @ {BASE}\n")
    results = []
    for t in tasks:
        t0 = time.time()
        r = run_eval_task(t, run_fn=_run_fn, collect_artifacts_fn=_collect_fn, judge_fn=_judge_fn)
        results.append(r)
        flag = "✅" if r.passed else "❌"
        print(f"{flag} {t.id:24} files_ok={r.files_ok} judge={r.judge_score:.2f} "
              f"steps={r.steps} ${r.cost_usd:.3f} {int(time.time()-t0)}s"
              + (f"  缺:{r.missing_files}" if r.missing_files else "")
              + (f"  err:{r.error[:60]}" if r.error else ""))
    s = summarize(results)
    print("\n=== 汇总 ===")
    print(json.dumps(s, ensure_ascii=False, indent=2))
    out = os.path.join(os.path.dirname(__file__), "last_report.json")
    json.dump({"summary": s, "results": [r.__dict__ for r in results]},
              open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"\n报告已落:{out}")
    return 0 if s["passed"] == s["total"] else 1


if __name__ == "__main__":
    sys.exit(main())