from __future__ import annotations from enum import StrEnum from math import sqrt from pydantic import BaseModel, Field, model_validator class AssetModality(StrEnum): IMAGE = "image" VIDEO = "video" VOICE = "voice" class ShotArchetype(StrEnum): # image CHARACTER_SHEET = "character_sheet" TURNAROUND = "turnaround" EXPRESSION_PACK = "expression_pack" TRANSPARENT_CUTOUT = "transparent_cutout" BACKGROUND_PLATE = "background_plate" STORYBOARD_FRAME = "storyboard_frame" EDUCATIONAL_DIAGRAM = "educational_diagram" START_END_FRAME = "start_end_frame" # video CHARACTER_CLOSEUP = "character_closeup" CHARACTER_DIALOGUE = "character_dialogue" MULTI_CHARACTER = "multi_character" CINEMATIC_ESTABLISHING = "cinematic_establishing" SHORT_BROLL = "short_broll" CLARITY_VISUAL = "clarity_visual" MOTION_TRANSFER = "motion_transfer" # voice VOICE_NEUTRAL_DIALOGUE = "voice_neutral_dialogue" VOICE_EMOTIONAL_DIALOGUE = "voice_emotional_dialogue" VOICE_EDUCATIONAL_EXPLANATION = "voice_educational_explanation" VOICE_REACTION = "voice_reaction" class BenchmarkConfidence(StrEnum): INSUFFICIENT_SAMPLE = "insufficient_sample" PROVISIONAL = "provisional" DECISION_READY = "decision_ready" class ModelBenchmarkMetrics(BaseModel): model_id: str provider: str | None = None modality: AssetModality shot_archetype: ShotArchetype sample_count: int = Field(ge=0) visual_quality: float | None = Field(default=None, ge=0, le=100) identity_consistency: float | None = Field(default=None, ge=0, le=100) prompt_adherence: float = Field(ge=0, le=100) anatomy_consistency: float | None = Field(default=None, ge=0, le=100) text_diagram_accuracy: float | None = Field(default=None, ge=0, le=100) background_cleanliness: float | None = Field(default=None, ge=0, le=100) motion_quality: float | None = Field(default=None, ge=0, le=100) temporal_continuity: float | None = Field(default=None, ge=0, le=100) audio_lipsync_quality: float | None = Field(default=None, ge=0, le=100) voice_identity_similarity: float | None = Field(default=None, ge=0, le=100) pronunciation_accuracy: float | None = Field(default=None, ge=0, le=100) emotional_readability: float | None = Field(default=None, ge=0, le=100) naturalness: float | None = Field(default=None, ge=0, le=100) first_pass_acceptance_pct: float = Field(ge=0, le=100) provider_success_pct: float = Field(ge=0, le=100) rework_rate_pct: float = Field(ge=0, le=100) median_latency_sec: float = Field(gt=0) p95_latency_sec: float = Field(gt=0) credits_per_output_unit: float = Field(gt=0) credits_per_accepted_unit: float = Field(gt=0) @model_validator(mode="after") def validate_modality_metrics(self) -> ModelBenchmarkMetrics: if self.modality is AssetModality.IMAGE: if self.visual_quality is None or self.identity_consistency is None: raise ValueError("image benchmarks require visual_quality and identity_consistency") elif self.modality is AssetModality.VIDEO: required = ( self.visual_quality, self.identity_consistency, self.motion_quality, self.temporal_continuity, ) if any(value is None for value in required): raise ValueError( "video benchmarks require visual_quality, identity_consistency, motion_quality and temporal_continuity" ) elif self.modality is AssetModality.VOICE: required = ( self.voice_identity_similarity, self.pronunciation_accuracy, self.emotional_readability, self.naturalness, ) if any(value is None for value in required): raise ValueError( "voice benchmarks require identity, pronunciation, emotion and naturalness metrics" ) return self class ModelBenchmarkPolicy(BaseModel): quality_floor: float = Field(default=80, ge=0, le=100) identity_floor: float = Field(default=80, ge=0, le=100) pronunciation_floor: float = Field(default=0, ge=0, le=100) target_credits_per_accepted_unit: float = Field(default=3.0, gt=0) target_median_latency_sec: float = Field(default=120.0, gt=0) quality_weight: float = 0.30 identity_weight: float = 0.20 adherence_weight: float = 0.10 modality_quality_weight: float = 0.10 acceptance_weight: float = 0.10 cost_weight: float = 0.10 latency_weight: float = 0.05 reliability_weight: float = 0.05 class ModelBenchmarkResult(BaseModel): model_id: str modality: AssetModality shot_archetype: ShotArchetype raw_score: float confidence_adjusted_score: float confidence: BenchmarkConfidence confidence_factor: float hard_gate_pass: bool violations: tuple[str, ...] breakdown: dict[str, float] def default_policy(modality: AssetModality, archetype: ShotArchetype) -> ModelBenchmarkPolicy: if modality is AssetModality.IMAGE: if archetype in { ShotArchetype.CHARACTER_SHEET, ShotArchetype.TURNAROUND, ShotArchetype.EXPRESSION_PACK, ShotArchetype.TRANSPARENT_CUTOUT, }: return ModelBenchmarkPolicy( quality_floor=90, identity_floor=95, target_credits_per_accepted_unit=15.0, ) if archetype is ShotArchetype.EDUCATIONAL_DIAGRAM: return ModelBenchmarkPolicy( quality_floor=88, identity_floor=70, target_credits_per_accepted_unit=12.0, ) return ModelBenchmarkPolicy( quality_floor=82, identity_floor=75, target_credits_per_accepted_unit=10.0, ) if modality is AssetModality.VOICE: return ModelBenchmarkPolicy( quality_floor=90, identity_floor=95, pronunciation_floor=98, target_credits_per_accepted_unit=5.0, target_median_latency_sec=60.0, ) if archetype in {ShotArchetype.CHARACTER_CLOSEUP, ShotArchetype.CHARACTER_DIALOGUE}: return ModelBenchmarkPolicy( quality_floor=88, identity_floor=95, target_credits_per_accepted_unit=3.5 ) if archetype is ShotArchetype.MULTI_CHARACTER: return ModelBenchmarkPolicy( quality_floor=90, identity_floor=95, target_credits_per_accepted_unit=4.0 ) if archetype is ShotArchetype.CLARITY_VISUAL: return ModelBenchmarkPolicy( quality_floor=85, identity_floor=80, target_credits_per_accepted_unit=2.5 ) if archetype is ShotArchetype.CINEMATIC_ESTABLISHING: return ModelBenchmarkPolicy( quality_floor=82, identity_floor=80, target_credits_per_accepted_unit=3.5 ) if archetype is ShotArchetype.MOTION_TRANSFER: return ModelBenchmarkPolicy( quality_floor=85, identity_floor=90, target_credits_per_accepted_unit=3.0 ) return ModelBenchmarkPolicy( quality_floor=75, identity_floor=70, target_credits_per_accepted_unit=2.0 ) def _bounded_ratio_score(target: float, observed: float) -> float: return max(0.0, min(100.0, 100.0 * target / observed)) def _confidence(sample_count: int) -> tuple[BenchmarkConfidence, float]: if sample_count < 5: return BenchmarkConfidence.INSUFFICIENT_SAMPLE, min(0.35, sample_count / 15) if sample_count < 20: return BenchmarkConfidence.PROVISIONAL, min(0.85, sqrt(sample_count / 20)) return BenchmarkConfidence.DECISION_READY, 1.0 def _modality_scores(metrics: ModelBenchmarkMetrics) -> tuple[float, float, float]: if metrics.modality is AssetModality.VOICE: identity = float(metrics.voice_identity_similarity or 0) quality = (float(metrics.naturalness or 0) + float(metrics.emotional_readability or 0)) / 2 modality_quality = float(metrics.pronunciation_accuracy or 0) return quality, identity, modality_quality quality = float(metrics.visual_quality or 0) identity = float(metrics.identity_consistency or 0) if metrics.modality is AssetModality.VIDEO: modality_quality = ( float(metrics.motion_quality or 0) + float(metrics.temporal_continuity or 0) ) / 2 else: candidates = [ metrics.anatomy_consistency, metrics.text_diagram_accuracy, metrics.background_cleanliness, ] available = [float(value) for value in candidates if value is not None] modality_quality = sum(available) / len(available) if available else quality return quality, identity, modality_quality def score_model_benchmark( metrics: ModelBenchmarkMetrics, policy: ModelBenchmarkPolicy | None = None, ) -> ModelBenchmarkResult: policy = policy or default_policy(metrics.modality, metrics.shot_archetype) quality, identity, modality_quality = _modality_scores(metrics) violations: list[str] = [] if quality < policy.quality_floor: violations.append("quality_below_floor") if identity < policy.identity_floor: violations.append("identity_consistency_below_floor") if metrics.modality is AssetModality.VOICE: pronunciation = float(metrics.pronunciation_accuracy or 0) if pronunciation < policy.pronunciation_floor: violations.append("pronunciation_below_floor") adherence = metrics.prompt_adherence acceptance = metrics.first_pass_acceptance_pct cost = _bounded_ratio_score( policy.target_credits_per_accepted_unit, metrics.credits_per_accepted_unit, ) latency = _bounded_ratio_score(policy.target_median_latency_sec, metrics.median_latency_sec) reliability = metrics.provider_success_pct weighted = { "quality": quality * policy.quality_weight, "identity": identity * policy.identity_weight, "adherence": adherence * policy.adherence_weight, "modality_quality": modality_quality * policy.modality_quality_weight, "first_pass_acceptance": acceptance * policy.acceptance_weight, "cost_efficiency": cost * policy.cost_weight, "latency_efficiency": latency * policy.latency_weight, "provider_reliability": reliability * policy.reliability_weight, } raw_score = sum(weighted.values()) confidence, factor = _confidence(metrics.sample_count) confidence_adjusted = 50.0 + (raw_score - 50.0) * factor hard_gate_pass = not violations if not hard_gate_pass: confidence_adjusted = min(confidence_adjusted, 49.0) return ModelBenchmarkResult( model_id=metrics.model_id, modality=metrics.modality, shot_archetype=metrics.shot_archetype, raw_score=round(raw_score, 2), confidence_adjusted_score=round(confidence_adjusted, 2), confidence=confidence, confidence_factor=round(factor, 4), hard_gate_pass=hard_gate_pass, violations=tuple(violations), breakdown={key: round(value, 2) for key, value in weighted.items()}, )