lambdagent/tests/test_agent_eval.py

"""agent_eval 框架单测:用 mock 跑/收产物/judge,验证行为回归评估逻辑(可进 CI)。

真实 LLM 评估由 evals/run_evals.py 手动/nightly 触发(花钱、要 key),不在默认门禁。
"""
from __future__ import annotations

from agentpaas.engine.agent_eval import EvalTask, run_eval_task, summarize
from agentpaas.engine.pipeline import JudgeResult


def _judge(score):
    return lambda req: JudgeResult(score=score, passed=score >= req.threshold,
                                   reasons=["mock"], raw="", usage={})


def _run(output="done", steps=10, cost=0.1, ws="/ws"):
    return lambda task: {"output": output, "steps": steps, "cost_usd": cost, "workspace_path": ws}


def test_all_good_passes():
    task = EvalTask(id="t1", input="x", must_produce=["report.md"], rubric="好不好", threshold=0.6)
    r = run_eval_task(task, run_fn=_run(),
                      collect_artifacts_fn=lambda ws: {"report.md": "完整内容"},
                      judge_fn=_judge(0.8))
    assert r.files_ok and r.judge_passed and r.passed
    assert r.judge_score == 0.8 and r.steps == 10


def test_missing_file_fails():
    task = EvalTask(id="t2", must_produce=["paper.pdf", "report.md"], rubric="", threshold=0.6)
    r = run_eval_task(task, run_fn=_run(),
                      collect_artifacts_fn=lambda ws: {"report.md": "x"},
                      judge_fn=_judge(1.0))
    assert not r.files_ok
    assert r.missing_files == ["paper.pdf"]
    assert not r.passed


def test_judge_below_threshold_fails():
    task = EvalTask(id="t3", must_produce=[], rubric="质量", threshold=0.7)
    r = run_eval_task(task, run_fn=_run(),
                      collect_artifacts_fn=lambda ws: {"a.md": "敷衍"},
                      judge_fn=_judge(0.4))
    assert not r.judge_passed and not r.passed


def test_error_output_fails():
    task = EvalTask(id="t4", rubric="", must_produce=[])
    r = run_eval_task(task, run_fn=_run(output="[DASHSCOPE_ERROR] boom"),
                      collect_artifacts_fn=lambda ws: {},
                      judge_fn=_judge(1.0))
    assert r.error and not r.passed


def test_run_exception_captured():
    def boom(task):
        raise RuntimeError("run crashed")
    task = EvalTask(id="t5")
    r = run_eval_task(task, run_fn=boom,
                      collect_artifacts_fn=lambda ws: {}, judge_fn=_judge(1.0))
    assert "run crashed" in r.error and not r.passed


def test_no_rubric_skips_judge():
    task = EvalTask(id="t6", must_produce=["x.md"], rubric="")
    r = run_eval_task(task, run_fn=_run(),
                      collect_artifacts_fn=lambda ws: {"x.md": "ok"},
                      judge_fn=lambda req: (_ for _ in ()).throw(AssertionError("不该调 judge")))
    assert r.judge_passed and r.passed  # 无 rubric → 不调 judge,judge_passed 默认 True


def test_summarize():
    task = EvalTask(id="a", must_produce=["r.md"], rubric="", )
    good = run_eval_task(task, run_fn=_run(steps=5, cost=0.2),
                         collect_artifacts_fn=lambda ws: {"r.md": "x"}, judge_fn=_judge(1))
    bad = run_eval_task(EvalTask(id="b", must_produce=["miss.pdf"]),
                        run_fn=_run(steps=9, cost=0.3),
                        collect_artifacts_fn=lambda ws: {}, judge_fn=_judge(1))
    s = summarize([good, bad])
    assert s["total"] == 2 and s["passed"] == 1 and s["pass_rate"] == 0.5
    assert s["failures"] == ["b"]
    assert s["total_cost_usd"] == 0.5