Spaces:
Running
Running
File size: 3,799 Bytes
6343479 | 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 | """Resource Pressure State Evaluator with Hysteresis.
Section 14:
- GREEN: Normal operation (< 70% RAM, normal queue)
- YELLOW: Reduce background work (70% - 80% RAM)
- ORANGE: Pause expensive background generation (80% - 88% RAM)
- RED: Unload non-critical models (88% - 94% RAM)
- CRITICAL: Preserve core state, pause generation, perform recovery (> 94% RAM)
Uses hysteresis thresholds to prevent rapid oscillation between states.
"""
from enum import Enum
from typing import Dict, Any
class PressureState(str, Enum):
GREEN = "GREEN"
YELLOW = "YELLOW"
ORANGE = "ORANGE"
RED = "RED"
CRITICAL = "CRITICAL"
class PressureEvaluator:
def __init__(self,
yellow_enter: float = 72.0, yellow_exit: float = 68.0,
orange_enter: float = 82.0, orange_exit: float = 78.0,
red_enter: float = 89.0, red_exit: float = 85.0,
critical_enter: float = 94.0, critical_exit: float = 91.0):
self.ye_enter = yellow_enter
self.ye_exit = yellow_exit
self.or_enter = orange_enter
self.or_exit = orange_exit
self.re_enter = red_enter
self.re_exit = red_exit
self.cr_enter = critical_enter
self.cr_exit = critical_exit
self.current_state = PressureState.GREEN
self.last_percent = 0.0
def evaluate(self, ram_percent: float) -> PressureState:
"""Evaluates pressure state using hysteresis to avoid state flutter."""
self.last_percent = float(ram_percent)
if self.current_state == PressureState.GREEN:
if ram_percent >= self.cr_enter:
self.current_state = PressureState.CRITICAL
elif ram_percent >= self.re_enter:
self.current_state = PressureState.RED
elif ram_percent >= self.or_enter:
self.current_state = PressureState.ORANGE
elif ram_percent >= self.ye_enter:
self.current_state = PressureState.YELLOW
elif self.current_state == PressureState.YELLOW:
if ram_percent >= self.cr_enter:
self.current_state = PressureState.CRITICAL
elif ram_percent >= self.re_enter:
self.current_state = PressureState.RED
elif ram_percent >= self.or_enter:
self.current_state = PressureState.ORANGE
elif ram_percent <= self.ye_exit:
self.current_state = PressureState.GREEN
elif self.current_state == PressureState.ORANGE:
if ram_percent >= self.cr_enter:
self.current_state = PressureState.CRITICAL
elif ram_percent >= self.re_enter:
self.current_state = PressureState.RED
elif ram_percent <= self.or_exit:
self.current_state = PressureState.YELLOW
elif self.current_state == PressureState.RED:
if ram_percent >= self.cr_enter:
self.current_state = PressureState.CRITICAL
elif ram_percent <= self.re_exit:
self.current_state = PressureState.ORANGE
elif self.current_state == PressureState.CRITICAL:
if ram_percent <= self.cr_exit:
self.current_state = PressureState.RED
return self.current_state
def status(self) -> Dict[str, Any]:
return {
"state": self.current_state.value,
"ram_percent": round(self.last_percent, 1),
"allow_background_ai": self.current_state in (PressureState.GREEN, PressureState.YELLOW),
"allow_speculative_work": self.current_state == PressureState.GREEN,
"recovery_required": self.current_state == PressureState.CRITICAL
}
|