1317552701
Introduce 智能评估 as an evaluation paradigm parallel to static evaluation, driven by OpenClaw. The platform supplies storage, lifecycle, and reporting; OpenClaw plans and executes. - Data model: IntelligentEval + Session + Message tables (new, not reusing exploration) - Lifecycle state machine: draft → planning → pending_approval → executing → completed/cancelled/failed - Session API: create/message (channel-forwarded)/close with turn accounting - Report API: pydantic-validated structured report, executing → completed, Markdown export (pure renderer) - Alembic migration for the three tables; domain glossary added to CONTEXT.md
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
"""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.
|
||
"""
|
||
|
||
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
|
||
|
||
|
||
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
|
||
|
||
|
||
_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("| --- | --- |")
|
||
for dimension, score in validated.scores.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"
|