"""Report generation for evaluation runs.""" import json from datetime import datetime from pathlib import Path from typing import Any, Optional from jinja2 import Template from agenteval.storage.db import DATA_DIR from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository HTML_TEMPLATE = """ 评测报告 - {{ report.run_id }}

评测报告

评测对象:{{ report.target_name }}({{ report.target_id }})

评测场景:{{ report.scenario_name }}({{ report.scenario_id }})

执行时间:{{ report.started_at }} 至 {{ report.completed_at or '进行中' }}

{{ report.summary.total_cases }}
用例总数
{{ report.summary.passed_cases }}
通过用例
{{ report.summary.total_rules }}
规则总数
{{ "%.2f"|format(report.summary.pass_rate * 100) }}%
规则通过率
{% for case in report.cases %}
用例 {{ case.case_id }}
{% for turn in case.turns %}
用户消息
{{ turn.sent_text }}
智能体回复({{ turn.latency_ms }}ms)
{{ turn.reply_text or '(无回复)' }}
{% endfor %}
{% for result in case.results %}
{{ '通过' if result.passed else '失败' }} {{ result.rule_type }}: {{ result.reason }}
{% endfor %}
{% endfor %}
""" def _extract_text(data: Any) -> str: if data is None: return "" if isinstance(data, str): return data if isinstance(data, dict): body = data.get("msgBody") or data.get("content", "") if isinstance(body, dict): return body.get("content", "") return str(body) return str(data) def generate_report(run_id: str, session=None) -> dict[str, Any]: """Build a structured report dict for a run.""" run_repo = RunRepository(session) target_repo = TargetRepository(session) scenario_repo = ScenarioRepository(session) run = run_repo.get(run_id) if not run: raise ValueError(f"run not found: {run_id}") target = target_repo.get(run.target_id) scenario = scenario_repo.get(run.scenario_id) turns = run_repo.get_turns(run_id) results = run_repo.get_results(run_id) # Group by case case_map: dict[str, dict[str, Any]] = {} for turn in turns: case_map.setdefault(turn.case_id, {"turns": [], "results": []}) sent = turn.get_sent_message() reply = turn.get_reply() case_map[turn.case_id]["turns"].append( { "round": turn.round_index, "sent_text": _extract_text(sent.get("msgBody")), "reply_text": _extract_text(reply.get("msgBody") if reply else None), "latency_ms": turn.latency_ms, "question_msg_id": turn.question_msg_id, } ) for result in results: case_map.setdefault(result.case_id, {"turns": [], "results": []}) case_map[result.case_id]["results"].append( { "rule_type": result.rule_type, "passed": result.passed, "score": result.score, "reason": result.reason, } ) cases = [] for case_id in sorted(case_map.keys()): item = case_map[case_id] cases.append( { "case_id": case_id, "turns": sorted(item["turns"], key=lambda x: x["round"]), "results": item["results"], } ) summary = run.summary or {} return { "run_id": run.id, "target_id": run.target_id, "target_name": target.name if target else "未知", "scenario_id": run.scenario_id, "scenario_name": scenario.name if scenario else "未知", "status": run.status.value, "started_at": run.started_at.isoformat() if run.started_at else None, "completed_at": run.completed_at.isoformat() if run.completed_at else None, "summary": { "total_cases": summary.get("total_cases", 0), "passed_cases": summary.get("passed_cases", 0), "failed_cases": summary.get("failed_cases", 0), "total_rules": summary.get("total_rules", 0), "passed_rules": summary.get("passed_rules", 0), "pass_rate": summary.get("pass_rate", 0.0), }, "cases": cases, } def render_json_report(run_id: str, session=None) -> str: """Render a report as JSON string.""" report = generate_report(run_id, session) return json.dumps(report, ensure_ascii=False, indent=2) def render_html_report(run_id: str, session=None) -> str: """Render a report as HTML string.""" report = generate_report(run_id, session) template = Template(HTML_TEMPLATE) return template.render(report=report) def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path: """Generate and save a report to disk.""" output_dir = output_dir or DATA_DIR / "reports" output_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if fmt == "html": content = render_html_report(run_id) path = output_dir / f"report_{run_id}_{timestamp}.html" elif fmt == "json": content = render_json_report(run_id) path = output_dir / f"report_{run_id}_{timestamp}.json" else: raise ValueError(f"unsupported report format: {fmt}") path.write_text(content, encoding="utf-8") return path