Spaces:
Sleeping
Sleeping
| """ | |
| Metrics counters for streaming intent tracking. | |
| Tracks escalation frequency, gray zone hits, decision timing, and near-misses. | |
| Thread-safe for use in concurrent request handling. | |
| Near-miss tracking: | |
| - Cases where escalation probability exceeded threshold but never committed | |
| - Critical for operational monitoring and safety validation | |
| """ | |
| import threading | |
| import time | |
| import numpy as np | |
| from dataclasses import dataclass, field | |
| from typing import Dict, Any, List, Optional, Tuple | |
| from collections import defaultdict | |
| class MetricsCounter: | |
| """ | |
| Thread-safe metrics tracking for streaming intent router. | |
| Tracks: | |
| - Total steps processed | |
| - Escalation count and rate | |
| - Gray zone hits (neither escalate nor commit) | |
| - Average steps to decision | |
| - Intent commitment distribution | |
| - Time-to-escalation distribution (percentiles) | |
| - Near-miss rate and peak probabilities | |
| """ | |
| total_steps: int = 0 | |
| escalation_count: int = 0 | |
| gray_zone_count: int = 0 | |
| commitment_count: int = 0 | |
| commitment_by_intent: Dict[str, int] = field(default_factory=lambda: defaultdict(int)) | |
| steps_to_escalation: List[int] = field(default_factory=list) | |
| steps_to_commitment: List[int] = field(default_factory=list) | |
| _current_session_steps: int = 0 | |
| _session_count: int = 0 | |
| _lock: threading.Lock = field(default_factory=threading.Lock) | |
| # Near-miss tracking | |
| near_miss_count: int = 0 | |
| near_miss_peak_probs: List[float] = field(default_factory=list) | |
| # Time-to-escalation in milliseconds (for distribution analysis) | |
| escalation_latencies_ms: List[float] = field(default_factory=list) | |
| _session_start_time_ms: float = 0.0 | |
| def record_step(self) -> None: | |
| """Record a processing step (thread-safe).""" | |
| with self._lock: | |
| self.total_steps += 1 | |
| self._current_session_steps += 1 | |
| def record_escalation(self) -> None: | |
| """Record an escalation event (thread-safe).""" | |
| with self._lock: | |
| self.escalation_count += 1 | |
| self.steps_to_escalation.append(self._current_session_steps) | |
| # Record time-to-escalation | |
| if self._session_start_time_ms > 0: | |
| latency_ms = time.time() * 1000 - self._session_start_time_ms | |
| self.escalation_latencies_ms.append(latency_ms) | |
| def record_commitment(self, intent: str) -> None: | |
| """Record an intent commitment (thread-safe).""" | |
| with self._lock: | |
| self.commitment_count += 1 | |
| self.commitment_by_intent[intent] += 1 | |
| self.steps_to_commitment.append(self._current_session_steps) | |
| def record_gray_zone(self) -> None: | |
| """Record a gray zone hit (no decision made) (thread-safe).""" | |
| with self._lock: | |
| self.gray_zone_count += 1 | |
| def record_near_miss(self, peak_prob: float) -> None: | |
| """ | |
| Record a near-miss event (thread-safe). | |
| A near-miss is when escalation probability exceeded threshold | |
| at some point but escalation was never triggered. | |
| Args: | |
| peak_prob: Peak escalation probability observed in session. | |
| """ | |
| with self._lock: | |
| self.near_miss_count += 1 | |
| self.near_miss_peak_probs.append(peak_prob) | |
| def start_session(self) -> None: | |
| """Start a new tracking session (thread-safe).""" | |
| with self._lock: | |
| self._current_session_steps = 0 | |
| self._session_count += 1 | |
| self._session_start_time_ms = time.time() * 1000 | |
| def reset(self) -> None: | |
| """Reset all counters (thread-safe).""" | |
| with self._lock: | |
| self.total_steps = 0 | |
| self.escalation_count = 0 | |
| self.gray_zone_count = 0 | |
| self.commitment_count = 0 | |
| self.commitment_by_intent = defaultdict(int) | |
| self.steps_to_escalation = [] | |
| self.steps_to_commitment = [] | |
| self._current_session_steps = 0 | |
| self._session_count = 0 | |
| # Reset near-miss tracking | |
| self.near_miss_count = 0 | |
| self.near_miss_peak_probs = [] | |
| # Reset time-to-escalation tracking | |
| self.escalation_latencies_ms = [] | |
| self._session_start_time_ms = 0.0 | |
| def _compute_percentiles(self, data: List[float], percentiles: Optional[List[int]] = None) -> Dict[str, float]: | |
| """Compute percentiles for a list of values.""" | |
| if percentiles is None: | |
| percentiles = [50, 90, 95, 99] | |
| if not data: | |
| return {f"p{p}": 0.0 for p in percentiles} | |
| arr = np.array(data) | |
| return {f"p{p}": float(np.percentile(arr, p)) for p in percentiles} | |
| def get_summary(self) -> Dict[str, Any]: | |
| """Get metrics summary (thread-safe).""" | |
| with self._lock: | |
| total_decisions = self.escalation_count + self.commitment_count | |
| avg_steps_to_escalation = ( | |
| sum(self.steps_to_escalation) / len(self.steps_to_escalation) | |
| if self.steps_to_escalation else 0.0 | |
| ) | |
| avg_steps_to_commitment = ( | |
| sum(self.steps_to_commitment) / len(self.steps_to_commitment) | |
| if self.steps_to_commitment else 0.0 | |
| ) | |
| # Time-to-escalation distribution | |
| escalation_latency_distribution = self._compute_percentiles( | |
| self.escalation_latencies_ms | |
| ) | |
| # Near-miss statistics | |
| near_miss_rate = ( | |
| self.near_miss_count / self._session_count | |
| if self._session_count > 0 else 0.0 | |
| ) | |
| avg_near_miss_peak = ( | |
| sum(self.near_miss_peak_probs) / len(self.near_miss_peak_probs) | |
| if self.near_miss_peak_probs else 0.0 | |
| ) | |
| return { | |
| "total_steps": self.total_steps, | |
| "total_sessions": self._session_count, | |
| "escalation_count": self.escalation_count, | |
| "escalation_rate": ( | |
| self.escalation_count / total_decisions if total_decisions > 0 else 0.0 | |
| ), | |
| "commitment_count": self.commitment_count, | |
| "commitment_by_intent": dict(self.commitment_by_intent), | |
| "gray_zone_count": self.gray_zone_count, | |
| "gray_zone_rate": ( | |
| self.gray_zone_count / self.total_steps if self.total_steps > 0 else 0.0 | |
| ), | |
| "avg_steps_to_escalation": avg_steps_to_escalation, | |
| "avg_steps_to_commitment": avg_steps_to_commitment, | |
| "avg_steps_to_decision": ( | |
| (sum(self.steps_to_escalation) + sum(self.steps_to_commitment)) / | |
| max(1, len(self.steps_to_escalation) + len(self.steps_to_commitment)) | |
| ), | |
| # Time-to-escalation distribution (ms) | |
| "escalation_latency_ms": escalation_latency_distribution, | |
| # Near-miss metrics | |
| "near_miss_count": self.near_miss_count, | |
| "near_miss_rate": near_miss_rate, | |
| "near_miss_avg_peak_prob": avg_near_miss_peak, | |
| } | |
| def __repr__(self) -> str: | |
| summary = self.get_summary() | |
| return ( | |
| f"MetricsCounter(" | |
| f"steps={summary['total_steps']}, " | |
| f"escalations={summary['escalation_count']}, " | |
| f"commitments={summary['commitment_count']}, " | |
| f"gray_zone={summary['gray_zone_count']})" | |
| ) | |