Spaces:
Running
Running
File size: 2,638 Bytes
df97e68 | 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 | """
Separates RewardModel fields into reward signal r_t and cost signals c_t.
Phase 1-3 : costs logged only (diagnostic).
Phase 4 : costs drive Lagrangian multiplier updates.
Thresholds:
d_sla = 0.15 (max 15% SLA breach rate)
d_fairness = 0.20 (max 0.20 fairness gap)
d_escalation = 0.10 (max 10% wasted escalation)
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import List
import numpy as np
THRESHOLD_SLA = 0.15
THRESHOLD_FAIRNESS = 0.20
THRESHOLD_ESCALATION = 0.10
@dataclass
class CostRecord:
step: int
c_sla: float
c_fairness: float
c_escalation: float
c_invalid: float
c_idle: float
@dataclass
class EpisodeCostSummary:
mean_c_sla: float
mean_c_fairness: float
mean_c_escalation: float
sla_violated: bool
fairness_violated: bool
escalation_violated: bool
total_steps: int
class CostTracker:
"""Accumulates per-step cost signals across an episode."""
def __init__(self) -> None:
self._records: List[CostRecord] = []
self._step = 0
def reset(self) -> None:
self._records.clear()
self._step = 0
def record(self, reward_breakdown: dict) -> CostRecord:
rec = CostRecord(
step=self._step,
c_sla=abs(float(reward_breakdown.get("sla_penalty", 0.0))),
c_fairness=abs(float(reward_breakdown.get("fairness_penalty", 0.0))),
c_escalation=abs(float(reward_breakdown.get("invalid_action_penalty", 0.0))),
c_invalid=abs(float(reward_breakdown.get("invalid_action_penalty", 0.0))),
c_idle=abs(float(reward_breakdown.get("idle_capacity_penalty", 0.0))),
)
self._records.append(rec)
self._step += 1
return rec
def summarise(self) -> EpisodeCostSummary:
if not self._records:
return EpisodeCostSummary(0.0, 0.0, 0.0, False, False, False, 0)
mean_sla = float(np.mean([r.c_sla for r in self._records]))
mean_fair = float(np.mean([r.c_fairness for r in self._records]))
mean_esc = float(np.mean([r.c_escalation for r in self._records]))
return EpisodeCostSummary(
mean_c_sla=mean_sla,
mean_c_fairness=mean_fair,
mean_c_escalation=mean_esc,
sla_violated=(mean_sla > THRESHOLD_SLA),
fairness_violated=(mean_fair > THRESHOLD_FAIRNESS),
escalation_violated=(mean_esc > THRESHOLD_ESCALATION),
total_steps=len(self._records),
)
|