AgentEvalTool/backend/agenteval/intelligent_eval/report.py

"""Structured report for intelligent evaluation (票据 04).

Two concerns live here so they stay co-located:

* **Validation** — pydantic models describing the report contract OpenClaw
  submits. A malformed body yields a 422 before any state change happens.
* **Rendering** — ``render_report_markdown`` is a pure function (dict in,
  string out) with no DB or I/O, so it unit-tests trivially.
"""

import math
from typing import Any, Optional

from pydantic import BaseModel, Field, model_validator


class ReportEvidence(BaseModel):
    session_id: str = ""
    turn_index: Optional[int] = None
    user_said: str = ""
    assistant_replied: str = ""


class ReportFinding(BaseModel):
    issue: str = Field(min_length=1)
    severity: str = Field(min_length=1)
    dimension: str = Field(min_length=1)
    evidence: list[ReportEvidence] = Field(default_factory=list)
    suggestion: Optional[str] = None
    related_sop: Optional[str] = None


class ReportHighlight(BaseModel):
    description: str = ""
    dimension: Optional[str] = None


def normalize_scores(scores: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]:
    """归一 scores 到单一规范结构 ``{"overall": float|None, "dimensions": {维度: 分}}``。

    Analyst LLM 输出不可控，可能是扁平 ``{维度: 分}`` 或嵌套
    ``{overall, dimensions}``；统一为嵌套结构，缺失 overall 时取维度平均，
    非数值/非有限维度直接丢弃。ADR-0011：下游（前端/markdown）不再做双格式兼容。
    """
    if not scores:
        return scores
    nested = scores.get("dimensions")
    source = nested if isinstance(nested, dict) else scores
    dims = {
        str(k): float(v)
        for k, v in source.items()
        if k != "overall" and isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)
    }
    overall = scores.get("overall")
    if not isinstance(overall, (int, float)) or isinstance(overall, bool) or not math.isfinite(overall):
        overall = sum(dims.values()) / len(dims) if dims else None
    return {"overall": float(overall) if overall is not None else None, "dimensions": dims}


class ReportModel(BaseModel):
    summary: str = Field(min_length=1)
    scores: Optional[dict[str, Any]] = None
    findings: list[ReportFinding] = Field(default_factory=list)
    highlights: list[ReportHighlight] = Field(default_factory=list)
    priority_recommendations: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def _require_findings(self) -> "ReportModel":
        if not self.findings:
            raise ValueError("findings 不能为空")
        return self

    @model_validator(mode="after")
    def _normalize_scores(self) -> "ReportModel":
        self.scores = normalize_scores(self.scores)
        return self


_SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2}


def _severity_rank(severity: str) -> int:
    return _SEVERITY_ORDER.get((severity or "").lower(), 99)


def render_report_markdown(report: dict[str, Any], *, name: str = "", eval_id: str = "") -> str:
    """Render a validated report dict to Markdown. Pure function, no I/O."""
    validated = ReportModel.model_validate(report)

    lines: list[str] = []
    title = name or "智能评估报告"
    lines.append(f"# {title}")
    lines.append("")
    if eval_id:
        lines.append(f"> 评估 ID：`{eval_id}`")
        lines.append("")

    lines.append("## 总体概述")
    lines.append("")
    lines.append(validated.summary.strip())
    lines.append("")

    if validated.scores:
        lines.append("## 维度评分")
        lines.append("")
        lines.append("| 维度 | 分数 |")
        lines.append("| --- | --- |")
        overall = validated.scores.get("overall")
        if overall is not None:
            lines.append(f"| 综合 | {overall} |")
        for dimension, score in validated.scores.get("dimensions", {}).items():
            lines.append(f"| {dimension} | {score} |")
        lines.append("")

    lines.append("## 问题发现")
    lines.append("")
    findings = sorted(validated.findings, key=lambda f: _severity_rank(f.severity))
    for index, finding in enumerate(findings, start=1):
        lines.append(f"### {index}. {finding.issue}")
        lines.append("")
        lines.append(f"- **严重程度**：{finding.severity}")
        lines.append(f"- **维度**：{finding.dimension}")
        if finding.suggestion:
            lines.append(f"- **建议**：{finding.suggestion}")
        if finding.related_sop:
            lines.append(f"- **关联 SOP**：{finding.related_sop}")
        if finding.evidence:
            lines.append("")
            lines.append("**证据**：")
            lines.append("")
            for ev in finding.evidence:
                header_bits = []
                if ev.session_id:
                    header_bits.append(f"会话 `{ev.session_id}`")
                if ev.turn_index is not None:
                    header_bits.append(f"第 {ev.turn_index} 轮")
                header = "（" + "，".join(header_bits) + "）" if header_bits else ""
                lines.append(f"> {header}")
                if ev.user_said:
                    lines.append(f"> **用户**：{ev.user_said}")
                if ev.assistant_replied:
                    lines.append(f"> **对象**：{ev.assistant_replied}")
                lines.append("")
        lines.append("")

    if validated.highlights:
        lines.append("## 亮点")
        lines.append("")
        for highlight in validated.highlights:
            suffix = f"（{highlight.dimension}）" if highlight.dimension else ""
            lines.append(f"- {highlight.description}{suffix}")
        lines.append("")

    if validated.priority_recommendations:
        lines.append("## 优先改进建议")
        lines.append("")
        for recommendation in validated.priority_recommendations:
            lines.append(f"- {recommendation}")
        lines.append("")

    return "\n".join(lines).rstrip() + "\n"