Spaces:
Runtime error
Runtime error
File size: 4,641 Bytes
91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b 4d97278 91bd31b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | import random
from dataclasses import dataclass
from typing import List
from enum import Enum
TASK_TYPES = ["summarization", "qa", "coding", "translation", "classification"]
INPUT_LENGTHS = ["short", "medium", "long"]
class DriftType(str, Enum):
PROMPT_TEMPLATE_CHANGE = "prompt_template_change"
QUANTIZATION_APPLIED = "quantization_applied"
SAFETY_FILTER_MISCONFIG = "safety_filter_misconfig"
DATA_CONTAMINATION = "data_contamination"
CONTEXT_WINDOW_BUG = "context_window_bug"
INFRA_LATENCY = "infra_latency"
ROUTER_BUG = "router_bug"
DRIFT_CATALOG = {
DriftType.PROMPT_TEMPLATE_CHANGE: {
"description": "Upstream prompt template was silently modified",
"affected_task_types": ["all"],
"affected_input_lengths": ["all"],
"quality_delta": -0.18,
"remediation": "rollback_prompt_template",
},
DriftType.QUANTIZATION_APPLIED: {
"description": "4-bit quantization applied to reduce serving cost",
"affected_task_types": ["coding", "qa"],
"affected_input_lengths": ["long"],
"quality_delta": -0.25,
"remediation": "revert_quantization",
},
DriftType.SAFETY_FILTER_MISCONFIG: {
"description": "Safety filter misconfigured — false positives on sensitive keywords",
"affected_task_types": ["all"],
"affected_input_lengths": ["all"],
"quality_delta": -0.12,
"affected_probability": 0.3,
"remediation": "recalibrate_safety_filter",
},
DriftType.DATA_CONTAMINATION: {
"description": "Fine-tuning batch contaminated with low-quality summarization data",
"affected_task_types": ["summarization"],
"affected_input_lengths": ["all"],
"quality_delta": -0.30,
"remediation": "rollback_finetune_checkpoint",
},
DriftType.CONTEXT_WINDOW_BUG: {
"description": "Context window handling bug truncates long inputs incorrectly",
"affected_task_types": ["all"],
"affected_input_lengths": ["long"],
"quality_delta": -0.22,
"remediation": "patch_context_window_handler",
},
DriftType.INFRA_LATENCY: {
"description": "Serving infrastructure latency spike causing timeout-induced quality drops",
"affected_task_types": ["coding", "summarization"],
"affected_input_lengths": ["medium", "long"],
"quality_delta": -0.15,
"remediation": "scale_serving_infra",
},
DriftType.ROUTER_BUG: {
"description": "Traffic router sending requests to a stale model checkpoint",
"affected_task_types": ["all"],
"affected_input_lengths": ["all"],
"quality_delta": -0.20,
"affected_probability": 0.4,
"remediation": "fix_traffic_routing",
},
}
@dataclass
class SimulatedOutput:
sample_id: str
task_type: str
input_length: str
base_quality: float
final_quality: float
drift_applied: List[str]
class DriftSimulator:
def __init__(self, seed: int = 42):
self.rng = random.Random(seed)
def generate_pool(
self,
n: int = 200,
active_drifts: List[DriftType] | None = None,
) -> List[SimulatedOutput]:
pool = []
active_drifts = active_drifts or []
for i in range(n):
task = self.rng.choice(TASK_TYPES)
length = self.rng.choice(INPUT_LENGTHS)
base_q = round(self.rng.gauss(0.78, 0.08), 3)
base_q = max(0.0, min(1.0, base_q))
final_q = base_q
applied = []
for drift in active_drifts:
spec = DRIFT_CATALOG[drift]
task_match = (
spec["affected_task_types"] == ["all"]
or task in spec["affected_task_types"]
)
len_match = (
spec.get("affected_input_lengths") == ["all"]
or length in spec.get("affected_input_lengths", ["all"])
)
prob_match = self.rng.random() < spec.get("affected_probability", 1.0)
if task_match and len_match and prob_match:
final_q = max(0.0, final_q + spec["quality_delta"])
applied.append(drift.value)
pool.append(
SimulatedOutput(
sample_id=f"sample_{i:04d}",
task_type=task,
input_length=length,
base_quality=base_q,
final_quality=round(final_q, 3),
drift_applied=applied,
)
)
return pool |