zcbot/evaluation/models.py

"""评测配置与结果的数据模型。

只使用标准库，避免评测器反向污染 zcbot 的生产依赖。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Optional


DEFAULT_DIMENSION_WEIGHTS = {
    "task_quality": 40.0,
    "reliability": 20.0,
    "security": 15.0,
    "engineering": 15.0,
    "performance": 10.0,
}


class EvalConfigError(ValueError):
    """任务集或运行配置不合法。"""


@dataclass(frozen=True)
class AssertionSpec:
    type: str
    weight: float = 1.0
    value: Any = None
    path: str = ""
    min_bytes: int = 0
    max_value: Optional[float] = None
    description: str = ""

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> "AssertionSpec":
        if not isinstance(raw, dict):
            raise EvalConfigError("assertion 必须是 JSON object")
        kind = str(raw.get("type", "")).strip()
        if not kind:
            raise EvalConfigError("assertion.type 不能为空")
        weight = float(raw.get("weight", 1.0))
        if weight <= 0:
            raise EvalConfigError(f"assertion {kind!r} 的 weight 必须 > 0")
        return cls(
            type=kind,
            weight=weight,
            value=raw.get("value"),
            path=str(raw.get("path", "")).strip(),
            min_bytes=int(raw.get("min_bytes", 0)),
            max_value=(
                float(raw["max_value"])
                if raw.get("max_value") is not None
                else None
            ),
            description=str(raw.get("description", "")).strip(),
        )


@dataclass(frozen=True)
class EvalCase:
    id: str
    name: str
    dimension: str
    prompt: str
    assertions: tuple[AssertionSpec, ...]
    weight: float = 1.0
    repetitions: Optional[int] = None
    timeout_s: Optional[float] = None
    skill: str = ""
    model_profile: str = ""
    tags: tuple[str, ...] = ()

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> "EvalCase":
        if not isinstance(raw, dict):
            raise EvalConfigError("case 必须是 JSON object")
        case_id = str(raw.get("id", "")).strip()
        prompt = str(raw.get("prompt", "")).strip()
        dimension = str(raw.get("dimension", "")).strip()
        if not case_id or not prompt or not dimension:
            raise EvalConfigError("case.id、case.prompt、case.dimension 均不能为空")
        assertions = tuple(
            AssertionSpec.from_dict(item) for item in raw.get("assertions", [])
        )
        if not assertions:
            raise EvalConfigError(f"case {case_id!r} 至少需要一个 assertion")
        weight = float(raw.get("weight", 1.0))
        if weight <= 0:
            raise EvalConfigError(f"case {case_id!r} 的 weight 必须 > 0")
        repetitions = raw.get("repetitions")
        timeout_s = raw.get("timeout_s")
        return cls(
            id=case_id,
            name=str(raw.get("name", case_id)).strip() or case_id,
            dimension=dimension,
            prompt=prompt,
            assertions=assertions,
            weight=weight,
            repetitions=int(repetitions) if repetitions is not None else None,
            timeout_s=float(timeout_s) if timeout_s is not None else None,
            skill=str(raw.get("skill", "")).strip(),
            model_profile=str(raw.get("model_profile", "")).strip(),
            tags=tuple(str(v) for v in raw.get("tags", [])),
        )


@dataclass(frozen=True)
class EvalSuite:
    name: str
    version: str
    cases: tuple[EvalCase, ...]
    dimension_weights: dict[str, float] = field(
        default_factory=lambda: dict(DEFAULT_DIMENSION_WEIGHTS)
    )
    default_repetitions: int = 3
    pass_threshold: float = 0.8

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> "EvalSuite":
        if not isinstance(raw, dict):
            raise EvalConfigError("suite 文件顶层必须是 JSON object")
        cases = tuple(EvalCase.from_dict(item) for item in raw.get("cases", []))
        if not cases:
            raise EvalConfigError("suite 至少需要一个 case")
        ids = [case.id for case in cases]
        if len(ids) != len(set(ids)):
            raise EvalConfigError("case.id 不得重复")
        weights = {
            str(k): float(v)
            for k, v in (
                raw.get("dimension_weights") or DEFAULT_DIMENSION_WEIGHTS
            ).items()
        }
        if any(v <= 0 for v in weights.values()):
            raise EvalConfigError("dimension_weights 必须全部 > 0")
        unknown = sorted({case.dimension for case in cases} - set(weights))
        if unknown:
            raise EvalConfigError(f"case 使用了未声明维度: {', '.join(unknown)}")
        repetitions = int(raw.get("default_repetitions", 3))
        threshold = float(raw.get("pass_threshold", 0.8))
        if repetitions < 1:
            raise EvalConfigError("default_repetitions 必须 >= 1")
        if not 0 <= threshold <= 1:
            raise EvalConfigError("pass_threshold 必须在 0..1")
        return cls(
            name=str(raw.get("name", "zcbot-eval")).strip() or "zcbot-eval",
            version=str(raw.get("version", "1")).strip() or "1",
            cases=cases,
            dimension_weights=weights,
            default_repetitions=repetitions,
            pass_threshold=threshold,
        )


@dataclass
class RunObservation:
    response: str = ""
    duration_s: float = 0.0
    cost_cny: float = 0.0
    run_status: str = "idle"
    run_error: str = ""
    task_id: str = ""
    working_dir: str = ""
    model_profile: str = ""
    artifact_loader: Optional[Callable[[str], bytes]] = field(
        default=None, repr=False
    )

    def artifact(self, relative_path: str) -> bytes:
        if self.artifact_loader is None:
            raise FileNotFoundError(relative_path)
        return self.artifact_loader(relative_path)


@dataclass(frozen=True)
class AssertionResult:
    type: str
    passed: bool
    weight: float
    description: str
    detail: str = ""


@dataclass
class RepetitionResult:
    index: int
    observation: RunObservation
    assertions: list[AssertionResult]
    score: float


@dataclass
class CaseResult:
    case: EvalCase
    repetitions: list[RepetitionResult]
    score: float
    pass_at_1: bool
    pass_all: bool


def safe_case_slug(value: str) -> str:
    cleaned = "".join(c.lower() if c.isalnum() else "-" for c in value)
    cleaned = "-".join(part for part in cleaned.split("-") if part)
    return (cleaned or "case")[:48]


def resolve_under(base: Path, relative: str) -> Path:
    """将相对路径限制在 base 内，供本地 fixture/报告使用。"""
    target = (base / relative).resolve()
    root = base.resolve()
    try:
        target.relative_to(root)
    except ValueError as exc:
        raise EvalConfigError(f"路径越界: {relative!r}") from exc
    return target