Spaces:
Sleeping
Sleeping
File size: 6,846 Bytes
5add236 baa72f5 5add236 baa72f5 5add236 baa72f5 5add236 baa72f5 5add236 baa72f5 5add236 baa72f5 5add236 | 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 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | """End-of-episode deterministic grader.
CRITICAL: All scores are clamped to the strictly open interval (0.001, 0.999).
Never returns exactly 0.0 or 1.0.
"""
from typing import Any, Dict, List, Set
from .config import (
GRADER_WEIGHTS, SCORE_MIN, SCORE_MAX,
Severity, severity_distance,
)
def _clamp(score: float) -> float:
"""Clamp score to strictly open interval (SCORE_MIN, SCORE_MAX)."""
if score <= 0.0:
return SCORE_MIN
if score >= 1.0:
return SCORE_MAX
# Also clamp values very close to boundaries
return max(SCORE_MIN, min(SCORE_MAX, score))
def _pair_set_f1(
agent_incidents: Dict[str, Set[str]],
true_incidents: Dict[str, List[str]],
) -> float:
"""Compute pair-set F1 over incident groupings."""
if not true_incidents:
return 1.0 # Vacuously correct
# True pairs
true_pairs = set()
for label, ids in true_incidents.items():
sids = sorted(ids)
for i in range(len(sids)):
for j in range(i + 1, len(sids)):
true_pairs.add((sids[i], sids[j]))
if not true_pairs:
return 1.0
# Agent pairs
agent_pairs = set()
for label, ids in agent_incidents.items():
sids = sorted(ids)
for i in range(len(sids)):
for j in range(i + 1, len(sids)):
agent_pairs.add((sids[i], sids[j]))
if not agent_pairs:
return 0.0
tp = len(true_pairs & agent_pairs)
precision = tp / len(agent_pairs) if agent_pairs else 0.0
recall = tp / len(true_pairs) if true_pairs else 0.0
if precision + recall == 0:
return 0.0
return 2 * precision * recall / (precision + recall)
def compute_grader_score(
task_id: str,
ground_truth: Dict[str, Dict],
agent_decisions: Dict[str, Dict],
agent_incidents: Dict[str, Set[str]],
true_incidents: Dict[str, List[str]],
skipped_alerts: Set[str],
stealth_root_service: str = None,
original_alert_ids: Set[str] = None,
) -> float:
"""Compute the final grader score for an episode.
Returns a score in the strictly open interval (0.001, 0.999).
"""
weights = GRADER_WEIGHTS[task_id]
# Only score original alerts (exclude cascade-spawned alerts)
if original_alert_ids:
scoreable_ids = original_alert_ids
else:
scoreable_ids = set(ground_truth.keys())
total_alerts = len(scoreable_ids)
if total_alerts == 0:
return SCORE_MIN
# Separate real alerts and false alarms
real_alert_ids = set()
false_alarm_ids = set()
for aid in scoreable_ids:
gt = ground_truth.get(aid, {})
if gt.get("is_false_alarm", False):
false_alarm_ids.add(aid)
else:
real_alert_ids.add(aid)
# ── Root cause accuracy ──
rc_correct = 0
for aid in scoreable_ids:
gt = ground_truth[aid]
if aid in skipped_alerts:
# Skipped = agent says false alarm
if gt["is_false_alarm"]:
rc_correct += 1
elif aid in agent_decisions:
dec = agent_decisions[aid]
if dec.get("root_cause") == gt["root_cause"]:
rc_correct += 1
# else: untriaged = incorrect
root_cause_acc = rc_correct / total_alerts
# ── Severity accuracy ──
sev_score_sum = 0.0
for aid in scoreable_ids:
gt = ground_truth[aid]
if aid in skipped_alerts:
if gt["is_false_alarm"]:
sev_score_sum += 1.0 # Correct skip
# else: wrong skip = 0
elif aid in agent_decisions:
dec = agent_decisions[aid]
try:
agent_sev = Severity(dec.get("severity", ""))
true_sev = Severity(gt["severity"])
dist = severity_distance(agent_sev, true_sev)
if dist == 0:
sev_score_sum += 1.0
elif dist == 1:
sev_score_sum += 0.15
except ValueError:
pass
# else: untriaged = 0
severity_acc = sev_score_sum / total_alerts
# ── Remediation accuracy ──
rem_correct = 0
for aid in scoreable_ids:
gt = ground_truth[aid]
if aid in skipped_alerts:
if gt["is_false_alarm"]:
rem_correct += 1 # dismiss is correct for FA
elif aid in agent_decisions:
dec = agent_decisions[aid]
if dec.get("remediation") == gt["remediation"]:
rem_correct += 1
remediation_acc = rem_correct / total_alerts
# ── Incident link F1 ──
incident_f1 = _pair_set_f1(agent_incidents, true_incidents)
# ── False alarm accuracy ──
# FA accuracy = correctly identified as FA (skipped) + correctly identified as real (triaged)
if false_alarm_ids or real_alert_ids:
fa_correct = 0
# Correctly skipped false alarms (agent correctly identified it as FA)
fa_correct += len(false_alarm_ids & skipped_alerts)
# Correctly triaged real alerts (agent correctly identified it as real, not skipped)
for aid in real_alert_ids:
if aid in agent_decisions:
fa_correct += 1
fa_total = len(false_alarm_ids) + len(real_alert_ids)
false_alarm_acc = fa_correct / fa_total if fa_total > 0 else 1.0
else:
false_alarm_acc = 1.0 # Vacuously correct
# ── Coverage multiplier ──
covered = (set(agent_decisions.keys()) | skipped_alerts) & scoreable_ids
coverage_count = len(covered)
coverage = coverage_count / total_alerts if total_alerts > 0 else 0.0
coverage_mult = coverage ** 1.5
# ── Weighted score ──
base_score = (
weights["root_cause_accuracy"] * root_cause_acc +
weights["severity_accuracy"] * severity_acc +
weights["remediation_accuracy"] * remediation_acc +
weights["incident_link_f1"] * incident_f1 +
weights["false_alarm_accuracy"] * false_alarm_acc
)
score = base_score * coverage_mult
# ── Stealth bonus (hard only) ──
if task_id == "hard" and stealth_root_service:
# Check if agent correctly identified the stealth root service
stealth_bonus_count = 0
for aid, gt in ground_truth.items():
if gt.get("is_stealth_root", False) and aid in agent_decisions:
dec = agent_decisions[aid]
if dec.get("root_cause") == gt["root_cause"]:
stealth_bonus_count += 1
# Apply bonus per correct stealth root (capped at 0.05)
score += min(stealth_bonus_count * 0.05, 0.05)
# ── CRITICAL: Clamp to strictly open interval (0.001, 0.999) ──
return _clamp(score)
|