AgentEvalTool/backend/cli/run.py

"""CLI commands for evaluation execution."""

import asyncio
from typing import Any

import typer
from agenteval.evaluation.engine import EvalEngine
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="评测执行")
console = Console()


def _progress(event: str, data: dict[str, Any]) -> None:
    if event == "case_start":
        console.print(f"[用例 {data['index']}/{data['total']}] {data['case_id']}")
    elif event == "turn_end":
        console.print(f"  第 {data['round']} 轮回复，耗时 {data['latency_ms']}ms")
    elif event == "rule_result":
        style = "green" if data["passed"] else "red"
        console.print(
            f"    [{data['rule_type']}] {'通过' if data['passed'] else '失败'} - {data['reason']}", style=style
        )
    elif event == "error":
        console.print(f"执行异常: {data['error']}", style="red")


@app.command("start", help="启动一次评测")
def start_run(
    target_id: str = typer.Option(..., help="评测对象 ID"),
    scenario_id: str = typer.Option(..., help="评测场景 ID"),
    output_json: bool = typer.Option(False, "--output-json", help="运行结束后输出 JSON 报告"),
) -> None:
    target = TargetRepository().get(target_id)
    if not target:
        console.print(f"评测对象不存在: {target_id}", style="red")
        raise typer.Exit(1)

    scenario = ScenarioRepository().get(scenario_id)
    if not scenario:
        console.print(f"评测场景不存在: {scenario_id}", style="red")
        raise typer.Exit(1)

    console.print(f"开始评测: 对象={target.name}, 场景={scenario.name}")
    engine = EvalEngine(target=target, scenario=scenario)
    try:
        run = asyncio.run(engine.run(progress_callback=_progress))
    except Exception as exc:
        console.print(f"评测失败: {exc}", style="red")
        raise typer.Exit(1)

    console.print(f"评测完成: run_id={run.id}, status={run.status.value}")
    if run.summary:
        console.print_json(data=run.summary)

    if output_json:
        from agenteval.evaluation.report import render_json_report

        console.print(render_json_report(run.id))


@app.command("list", help="列出评测执行记录")
def list_runs() -> None:
    runs = RunRepository().list_all()
    if not runs:
        console.print("暂无评测记录")
        return

    table = Table(show_header=True, header_style="bold")
    table.add_column("Run ID")
    table.add_column("对象 ID")
    table.add_column("场景 ID")
    table.add_column("状态")
    table.add_column("开始时间")
    table.add_column("完成时间")

    for run in runs:
        table.add_row(
            run.id or "",
            run.target_id,
            run.scenario_id,
            run.status.value,
            str(run.started_at)[:19] if run.started_at else "",
            str(run.completed_at)[:19] if run.completed_at else "",
        )
    console.print(table)


@app.command("status", help="查看评测执行状态")
def run_status(run_id: str) -> None:
    run = RunRepository().get(run_id)
    if not run:
        console.print(f"评测记录不存在: {run_id}", style="red")
        raise typer.Exit(1)

    console.print_json(data=run.model_dump())


@app.command("logs", help="查看评测执行日志（对话轮次）")
def run_logs(run_id: str) -> None:
    repo = RunRepository()
    run = repo.get(run_id)
    if not run:
        console.print(f"评测记录不存在: {run_id}", style="red")
        raise typer.Exit(1)

    turns = repo.get_turns(run_id)
    if not turns:
        console.print("暂无对话记录")
        return

    table = Table(show_header=True, header_style="bold")
    table.add_column("Case ID")
    table.add_column("轮次")
    table.add_column("消息 ID")
    table.add_column("耗时(ms)")
    table.add_column("发送时间")
    table.add_column("回复时间")

    for turn in turns:
        table.add_row(
            turn.case_id,
            str(turn.round_index),
            turn.question_msg_id or "",
            str(turn.latency_ms) if turn.latency_ms is not None else "",
            str(turn.sent_at)[:19] if turn.sent_at else "",
            str(turn.received_at)[:19] if turn.received_at else "",
        )
    console.print(table)