Files
mini-openclaw-v2/backend/contracts/judgment.py
T
qiaoqichaoandClaude Code 501a4d9c9b feat(v2): T08 专项离线质检——统一适配 fixture、逐任务准入与预算守卫
覆盖 AC-16/17/18/19/20/38/39/40/41(BR-17/18/19/20/37/38/39、E-10/13/14/15/25)。

新增:
- backend/contracts/judgment.py:JudgeProfile/TaskValidation 契约(逐任务逐用途
  验收登记 + 冻结 checkpoint)。放在独立契约模块,因为规格要求的 tokenizer/预算
  字段会撞上 prescreen.py 既有的"契约层不得出现凭据字样"静态守卫;不放宽该守卫。
- backend/services/judgment_adapters.py:统一适配入口。准入(未登记任务/用途拒绝、
  runtime 须独立评测)、冻结权重(unresolved 与不一致均拒绝)、上下文预算
  (正文与选项分开计,超限返回 E-25 且零发送)、中文否定/条件例外/来源矛盾/未登记
  映射一律落 unknown(inconclusive),不硬选正面结果。
- backend/fixtures/judgment/:中文提炼依据题集、本地模型原始输出、Jev 厂商响应
  fixture(虚构内容,离线契约用)。
- backend/test_judgment_adapters.py:30 项测试。

修改(最小面):
- contracts/prescreen.py:新增 E-25 运行级错误码 JUDGMENT_CONTEXT_EXCEEDED;
  Judgment 增 actual_checkpoint_ref、JudgmentConfig 增 checkpoint_ref。
- services/judgment_provider.py:映射适配层的策略拒绝异常;归一化保留实际权重。

未做(受授权限制):没有 Jev 凭据与 Laya 权重,因此只交付离线 fixture 路径,
live 传输层仅留注入点,未接入硬件与端口探测。fixture 通过不证明中文效果合格。

验证:backend 全量 unittest 440 项 OK(原 410 项无回归)。

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-30 09:09:05 +08:00

88 lines
4.0 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""T08 专项提供方准入契约:逐任务/用途验收与冻结权重修订。
Owner:本文件(docs/v2/11-judgment-providers.md §8.1 的 profile 契约)。
`contracts/prescreen.py` 继续持有运行侧对象(Judgment / JudgmentConfig /
ScreeningRun);把 profile 契约独立出来有两个理由:
1. 11 §8.1 规定 profile 必须带 tokenizer / 预算字段,而 `prescreen.py` 有一条
静态底线(`test_prescreen_contracts.test_no_credentials_in_contract_module`:
契约模块不得出现凭据字样)。把预算字段放进独立模块,比放宽那条安全断言更合适。
2. T08 的准入规则(task_validations)与运行对象生命周期不同:配置可换修订,
运行不得被就地改写(BR-20)。
依赖方向:本模块**不导入** `services/`。
"""
from __future__ import annotations
from typing import Literal
from pydantic import Field, model_validator
from contracts.prescreen import ID_MAX, PATH_MAX, Contract
JudgmentTask = Literal[
"extraction_grounding",
"experience_applicability",
"trajectory_quality",
"evidence_support",
]
JudgmentUsage = Literal["offline", "shadow", "runtime"]
class TaskValidation(Contract):
"""逐任务、逐用途的验收登记(docs/v2/11-judgment-providers.md §8.1)。
一条登记只证明"该任务在该用途下通过过验收",不是模型整体合格,也不能
由"服务能连通"或"别的任务通过"顶替(AC-38)。offline 通过与 runtime
启用必须分开登记,否则离线质检可以直接当生产准入。
"""
task: JudgmentTask
rubric_version: str = Field(min_length=1, max_length=ID_MAX)
dataset_ref: str = Field(min_length=1, max_length=PATH_MAX)
report_ref: str = Field(min_length=1, max_length=PATH_MAX)
decision: Literal["passed", "failed", "unverified"]
allowed_usage: JudgmentUsage
# 独立评测来源:同模型自评或其生成标签不算独立评测(AC-41)
independent: bool = False
class JudgmentProfile(Contract):
"""专项提供方配置(T08)。只描述"能跑什么",不代表已经跑通。
`checkpoint_ref` 必须可复现:远程不可变版本不可得时填 "unresolved",
此时运行期不允许隐式换权重(AC-39)。
"""
id: str = Field(min_length=1, max_length=ID_MAX)
revision: int = Field(ge=1)
provider: Literal["jev", "local"]
engine: str = Field(min_length=1, max_length=ID_MAX)
model_version: str = Field(min_length=1, max_length=ID_MAX)
adapter_version: str = Field(min_length=1, max_length=ID_MAX)
prompt_version: str = Field(min_length=1, max_length=ID_MAX)
mapping_version: str = Field(min_length=1, max_length=ID_MAX)
checkpoint_ref: str = Field(min_length=1, max_length=ID_MAX)
tokenizer_revision: str = Field(min_length=1, max_length=ID_MAX)
input_token_limit: int = Field(gt=0)
option_token_limit: int = Field(gt=0)
language_scope: str = Field(min_length=1, max_length=ID_MAX)
deployment_mode: Literal["in_process", "internal_http", "external_http"]
# 外发授权:未批准时 jev 一律拒绝发送,且不据此降低本地要求(AC-16/19)
egress_approved: bool = False
readiness: Literal["ready", "unready"] = "unready"
task_validations: list[TaskValidation] = Field(default_factory=list, max_length=32)
is_default: bool = False
@model_validator(mode="after")
def _check_profile(self) -> JudgmentProfile:
if self.provider == "local" and self.deployment_mode == "external_http":
raise ValueError("本地模型不得走外部出口;外发只属于 jev 路径")
if self.provider == "jev" and self.deployment_mode != "external_http":
raise ValueError("jev 属外部服务,部署模式必须是 external_http")
seen = {(item.task, item.allowed_usage) for item in self.task_validations}
if len(seen) != len(self.task_validations):
raise ValueError("同一任务与用途只能登记一条验收结果,不得重复累计")
return self