116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""确定性评分器。"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import re
|
|
import zipfile
|
|
|
|
from .models import AssertionResult, AssertionSpec, RunObservation
|
|
|
|
|
|
SUPPORTED_ASSERTIONS = {
|
|
"run_succeeded",
|
|
"response_nonempty",
|
|
"response_contains",
|
|
"response_not_contains",
|
|
"response_regex",
|
|
"artifact_exists",
|
|
"artifact_min_bytes",
|
|
"artifact_zip_valid",
|
|
"artifact_text_contains",
|
|
"max_cost_cny",
|
|
"max_duration_s",
|
|
}
|
|
|
|
|
|
def validate_assertion(spec: AssertionSpec) -> None:
|
|
if spec.type not in SUPPORTED_ASSERTIONS:
|
|
raise ValueError(f"不支持的 assertion.type: {spec.type!r}")
|
|
if spec.type.startswith("artifact_") and not spec.path:
|
|
raise ValueError(f"{spec.type} 需要 path")
|
|
if spec.type in {
|
|
"response_contains",
|
|
"response_not_contains",
|
|
"response_regex",
|
|
"artifact_text_contains",
|
|
} and not isinstance(spec.value, str):
|
|
raise ValueError(f"{spec.type} 需要字符串 value")
|
|
if spec.type in {"max_cost_cny", "max_duration_s"} and spec.max_value is None:
|
|
raise ValueError(f"{spec.type} 需要 max_value")
|
|
|
|
|
|
def _description(spec: AssertionSpec) -> str:
|
|
if spec.description:
|
|
return spec.description
|
|
if spec.path:
|
|
return f"{spec.type}: {spec.path}"
|
|
if spec.value is not None:
|
|
return f"{spec.type}: {spec.value}"
|
|
if spec.max_value is not None:
|
|
return f"{spec.type}: <= {spec.max_value}"
|
|
return spec.type
|
|
|
|
|
|
def evaluate_assertion(
|
|
spec: AssertionSpec, observation: RunObservation
|
|
) -> AssertionResult:
|
|
validate_assertion(spec)
|
|
passed = False
|
|
detail = ""
|
|
try:
|
|
if spec.type == "run_succeeded":
|
|
passed = observation.run_status == "idle" and not observation.run_error
|
|
detail = observation.run_error or observation.run_status
|
|
elif spec.type == "response_nonempty":
|
|
passed = bool(observation.response.strip())
|
|
detail = f"{len(observation.response)} chars"
|
|
elif spec.type == "response_contains":
|
|
passed = spec.value in observation.response
|
|
elif spec.type == "response_not_contains":
|
|
passed = spec.value not in observation.response
|
|
elif spec.type == "response_regex":
|
|
passed = re.search(spec.value, observation.response, re.MULTILINE) is not None
|
|
elif spec.type == "artifact_exists":
|
|
observation.artifact(spec.path)
|
|
passed = True
|
|
elif spec.type == "artifact_min_bytes":
|
|
data = observation.artifact(spec.path)
|
|
passed = len(data) >= spec.min_bytes
|
|
detail = f"{len(data)} bytes; required >= {spec.min_bytes}"
|
|
elif spec.type == "artifact_zip_valid":
|
|
data = observation.artifact(spec.path)
|
|
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
|
bad = archive.testzip()
|
|
passed = bad is None and bool(archive.namelist())
|
|
detail = f"first bad member: {bad}" if bad else f"{len(archive.namelist())} members"
|
|
elif spec.type == "artifact_text_contains":
|
|
text = observation.artifact(spec.path).decode("utf-8")
|
|
passed = spec.value in text
|
|
elif spec.type == "max_cost_cny":
|
|
assert spec.max_value is not None
|
|
passed = observation.cost_cny <= float(spec.max_value)
|
|
detail = f"{observation.cost_cny:.6f} CNY"
|
|
elif spec.type == "max_duration_s":
|
|
assert spec.max_value is not None
|
|
passed = observation.duration_s <= float(spec.max_value)
|
|
detail = f"{observation.duration_s:.3f}s"
|
|
except (FileNotFoundError, UnicodeDecodeError, zipfile.BadZipFile, OSError) as exc:
|
|
detail = f"{type(exc).__name__}: {exc}"
|
|
passed = False
|
|
return AssertionResult(
|
|
type=spec.type,
|
|
passed=passed,
|
|
weight=spec.weight,
|
|
description=_description(spec),
|
|
detail=detail,
|
|
)
|
|
|
|
|
|
def score_observation(
|
|
specs: tuple[AssertionSpec, ...], observation: RunObservation
|
|
) -> tuple[float, list[AssertionResult]]:
|
|
results = [evaluate_assertion(spec, observation) for spec in specs]
|
|
denominator = sum(result.weight for result in results)
|
|
numerator = sum(result.weight for result in results if result.passed)
|
|
return (numerator / denominator if denominator else 0.0), results
|