239 lines
10 KiB
Python
239 lines
10 KiB
Python
"""독립 human-labeled G7 voice-gain 종료 증거 계약.
|
|
|
|
이 계약은 repository synthetic benchmark와 의도적으로 분리되어 있다. 원음과
|
|
축어록을 받지 않고, 동의·분할·모델·라벨링 provenance와 비식별 수치 관측만 받는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
AllianceAxis = Literal["goal", "task", "bond"]
|
|
PredictionStatus = Literal["observed", "missing", "error"]
|
|
ParticipantSplit = Literal["calibration", "held_out"]
|
|
Sha256 = str
|
|
|
|
|
|
class G7EvidenceProvenance(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
protocol_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
consent_protocol_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
dataset_manifest_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
split_manifest_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
labeling_protocol_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
analysis_plan_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
registered_at: datetime
|
|
held_out_labels_opened_at: datetime
|
|
|
|
@model_validator(mode="after")
|
|
def require_preregistered_analysis(self) -> "G7EvidenceProvenance":
|
|
if self.registered_at >= self.held_out_labels_opened_at:
|
|
raise ValueError("analysis protocol must precede held-out label access")
|
|
return self
|
|
|
|
|
|
class G7ModelProvenance(BaseModel):
|
|
model_config = ConfigDict(
|
|
extra="forbid",
|
|
frozen=True,
|
|
protected_namespaces=(),
|
|
)
|
|
|
|
role: Literal["text_only_baseline", "voice_enabled_candidate"]
|
|
provider: str = Field(min_length=1, max_length=80)
|
|
model_id: str = Field(min_length=1, max_length=160)
|
|
model_version: str = Field(min_length=1, max_length=120)
|
|
artifact_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
configuration_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
|
|
|
|
class G7PowerPlan(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
primary_metric: Literal["paired_one_minus_mae_gain"] = (
|
|
"paired_one_minus_mae_gain"
|
|
)
|
|
clustering_unit: Literal["participant"] = "participant"
|
|
required_held_out_participants: int = Field(ge=1)
|
|
required_held_out_sessions: int = Field(ge=1)
|
|
required_paired_axis_observations: int = Field(ge=3)
|
|
alpha: Literal[0.05] = 0.05
|
|
target_power: float = Field(ge=0.8, lt=1.0)
|
|
minimally_detectable_gain: float = Field(gt=0.0, le=1.0)
|
|
planned_bootstrap_samples: Literal[10000] = 10000
|
|
|
|
|
|
class G7ParticipantAssignment(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
participant_key: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
|
|
split: ParticipantSplit
|
|
consent_receipt_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
|
|
|
|
class G7LabelerAttestation(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
labeler_key: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
|
|
blinded_to_model_condition: Literal[True] = True
|
|
blinded_to_other_labelers: Literal[True] = True
|
|
labeled_independently: Literal[True] = True
|
|
attestation_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
|
|
|
|
class G7HumanAxisLabel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
labeler_key: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
|
|
score: float = Field(ge=0.0, le=1.0)
|
|
category: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$")
|
|
|
|
|
|
class G7ReliabilityClaim(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
method: Literal["ICC(A,1)"] = "ICC(A,1)"
|
|
labeler_keys: tuple[str, ...] = Field(min_length=2)
|
|
reported_icc: float = Field(ge=-1.0, le=1.0)
|
|
reported_categorical_kappa: float = Field(ge=-1.0, le=1.0)
|
|
report_sha256: Sha256 = Field(pattern=r"^[a-f0-9]{64}$")
|
|
|
|
@model_validator(mode="after")
|
|
def require_unique_labelers(self) -> "G7ReliabilityClaim":
|
|
if len(set(self.labeler_keys)) != len(self.labeler_keys):
|
|
raise ValueError("reliability labeler keys must be unique")
|
|
return self
|
|
|
|
|
|
class G7PairedAxisObservation(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
observation_id: str = Field(pattern=r"^g7-human-observation-[A-Za-z0-9._:-]+$")
|
|
participant_key: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
|
|
session_key: str = Field(pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,127}$")
|
|
axis: AllianceAxis
|
|
text_only_status: PredictionStatus
|
|
text_only_score: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
voice_enabled_status: PredictionStatus
|
|
voice_enabled_score: float | None = Field(default=None, ge=0.0, le=1.0)
|
|
labels: tuple[G7HumanAxisLabel, ...] = Field(min_length=2)
|
|
raw_audio_included: Literal[False] = False
|
|
transcript_included: Literal[False] = False
|
|
|
|
@model_validator(mode="after")
|
|
def require_status_consistent_scores(self) -> "G7PairedAxisObservation":
|
|
pairs = (
|
|
(self.text_only_status, self.text_only_score, "text-only"),
|
|
(self.voice_enabled_status, self.voice_enabled_score, "voice-enabled"),
|
|
)
|
|
for status, score, label in pairs:
|
|
if status == "observed" and score is None:
|
|
raise ValueError(f"{label} observed status requires a score")
|
|
if status != "observed" and score is not None:
|
|
raise ValueError(f"{label} missing/error status must remain scoreless")
|
|
labeler_keys = [item.labeler_key for item in self.labels]
|
|
if len(set(labeler_keys)) != len(labeler_keys):
|
|
raise ValueError("observation labeler keys must be unique")
|
|
return self
|
|
|
|
|
|
class G7HumanVoiceGainEvidencePack(BaseModel):
|
|
"""합성 자료로 대체할 수 없는 독립 held-out G7 증거 묶음."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
schema_version: Literal["g7_human_voice_gain_v1"] = "g7_human_voice_gain_v1"
|
|
evidence_kind: Literal["independent_human_held_out_voice_gain"] = (
|
|
"independent_human_held_out_voice_gain"
|
|
)
|
|
data_classification: Literal["consented_deidentified_research_metrics"] = (
|
|
"consented_deidentified_research_metrics"
|
|
)
|
|
synthetic_pack: Literal[False] = False
|
|
clinical_claim_allowed: Literal[False] = False
|
|
provenance: G7EvidenceProvenance
|
|
text_only_model: G7ModelProvenance
|
|
voice_enabled_model: G7ModelProvenance
|
|
power_plan: G7PowerPlan
|
|
participants: tuple[G7ParticipantAssignment, ...] = Field(min_length=2)
|
|
labeler_attestations: tuple[G7LabelerAttestation, ...] = Field(min_length=2)
|
|
reliability: G7ReliabilityClaim
|
|
observations: tuple[G7PairedAxisObservation, ...] = Field(min_length=3)
|
|
|
|
@model_validator(mode="after")
|
|
def enforce_independent_holdout_boundaries(self) -> "G7HumanVoiceGainEvidencePack":
|
|
if self.text_only_model.role != "text_only_baseline":
|
|
raise ValueError("text-only model provenance role is invalid")
|
|
if self.voice_enabled_model.role != "voice_enabled_candidate":
|
|
raise ValueError("voice-enabled model provenance role is invalid")
|
|
baseline_identity = (
|
|
self.text_only_model.artifact_sha256,
|
|
self.text_only_model.configuration_sha256,
|
|
)
|
|
candidate_identity = (
|
|
self.voice_enabled_model.artifact_sha256,
|
|
self.voice_enabled_model.configuration_sha256,
|
|
)
|
|
if baseline_identity == candidate_identity:
|
|
raise ValueError("baseline and candidate model provenance must differ")
|
|
|
|
participant_keys = [item.participant_key for item in self.participants]
|
|
if len(set(participant_keys)) != len(participant_keys):
|
|
raise ValueError("participant assignments must be unique")
|
|
receipt_hashes = [item.consent_receipt_sha256 for item in self.participants]
|
|
if len(set(receipt_hashes)) != len(receipt_hashes):
|
|
raise ValueError("participant consent receipts must be unique")
|
|
split_by_participant = {
|
|
item.participant_key: item.split for item in self.participants
|
|
}
|
|
if "calibration" not in split_by_participant.values():
|
|
raise ValueError("participant-level calibration split is required")
|
|
if "held_out" not in split_by_participant.values():
|
|
raise ValueError("participant-level held-out split is required")
|
|
|
|
attested_labelers = {
|
|
item.labeler_key for item in self.labeler_attestations
|
|
}
|
|
if len(attested_labelers) != len(self.labeler_attestations):
|
|
raise ValueError("labeler attestations must be unique")
|
|
reliability_panel = set(self.reliability.labeler_keys)
|
|
if not reliability_panel.issubset(attested_labelers):
|
|
raise ValueError("reliability panel requires blind independent attestations")
|
|
|
|
observation_ids = [item.observation_id for item in self.observations]
|
|
if len(set(observation_ids)) != len(observation_ids):
|
|
raise ValueError("observation ids must be unique")
|
|
composite_keys = [
|
|
(item.participant_key, item.session_key, item.axis)
|
|
for item in self.observations
|
|
]
|
|
if len(set(composite_keys)) != len(composite_keys):
|
|
raise ValueError("paired participant/session/axis observations must be unique")
|
|
|
|
participant_by_session: dict[str, str] = {}
|
|
axes_by_session: dict[str, set[AllianceAxis]] = {}
|
|
for observation in self.observations:
|
|
if split_by_participant.get(observation.participant_key) != "held_out":
|
|
raise ValueError("evaluation observations must use held-out participants")
|
|
prior_participant = participant_by_session.setdefault(
|
|
observation.session_key,
|
|
observation.participant_key,
|
|
)
|
|
if prior_participant != observation.participant_key:
|
|
raise ValueError("a session cannot belong to multiple participants")
|
|
axes_by_session.setdefault(observation.session_key, set()).add(
|
|
observation.axis
|
|
)
|
|
row_labelers = {item.labeler_key for item in observation.labels}
|
|
if row_labelers != reliability_panel:
|
|
raise ValueError("every row must use the declared reliability panel")
|
|
|
|
required_axes: set[AllianceAxis] = {"goal", "task", "bond"}
|
|
if any(axes != required_axes for axes in axes_by_session.values()):
|
|
raise ValueError("every held-out session must cover goal, task, and bond")
|
|
return self
|