Spaces:
Running
Running
File size: 6,907 Bytes
f492127 25e135f f492127 25e135f f492127 25e135f f492127 25e135f f492127 25e135f f492127 25e135f f492127 25e135f f492127 25e135f f492127 | 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 | """
HaramGuard — ReflectionAgent
==============================
AISA Layer : Reasoning + Governance
Design Pattern : Reflection (Self-Critique & Correction)
Responsibilities:
- Observe every RiskResult produced by RiskAgent
- Critique the assessment for four systematic biases:
1. Chronic LOW under-reporting (20+ consecutive LOW with large crowd)
2. Rising trend ignored (trend=rising but risk=LOW with n>20 persons)
3. Count-risk mismatch (80+ persons but risk=LOW)
4. Over-estimation (HIGH risk but only <15 persons detected)
- Correct: override risk_level and risk_score when bias is found
- Log: persist every reflection to DB for evaluation / auditability
Flow per frame:
Observe → Critique → Correct → Log
"""
import numpy as np
from collections import deque
from core.models import FrameResult, RiskResult
import config
class ReflectionAgent:
BIAS_WINDOW = config.REFLECTION_BIAS_WINDOW # 10
CROWD_LOW_THRESH = config.REFLECTION_CROWD_LOW_THRESH # 15
HIGH_CROWD_THRESH = config.REFLECTION_HIGH_CROWD_THRESH # 50
CRITICAL_CROWD_THRESH = 70 # 70+ persons always HIGH (single-camera calibration)
def __init__(self):
self.name = 'ReflectionAgent'
self.aisa_layer = 'Reasoning (Reflection)'
self._history = deque(maxlen=self.BIAS_WINDOW)
self._reflection_log = []
print('🪞 [ReflectionAgent] Ready — 4 bias detectors active')
def reflect(self, rr: RiskResult, fr: FrameResult) -> dict:
"""
Four-step reflection loop:
Step 1 — Observe : record latest assessment in history window
Step 2 — Critique : check for each known bias pattern
Step 3 — Correct : compute corrected_level / corrected_score
Step 4 — Log : append to internal log (also saved to DB by pipeline)
Returns reflection dict — pipeline applies corrections to rr before
passing it downstream to OperationsAgent.
"""
self._history.append({
'risk_level': rr.risk_level,
'risk_score': rr.risk_score,
'person_count': fr.person_count,
'trend': rr.trend,
})
critique = []
bias_detected = False
corrected_level = rr.risk_level
corrected_score = rr.risk_score
# Bias 1: Chronic LOW under-reporting
if len(self._history) >= self.BIAS_WINDOW:
all_low = all(h['risk_level'] == 'LOW' for h in self._history)
avg_crowd = np.mean([h['person_count'] for h in self._history])
if all_low and avg_crowd > self.CROWD_LOW_THRESH:
bias_detected = True
corrected_level = 'MEDIUM'
corrected_score = max(rr.risk_score, 0.30) # push into MEDIUM zone (>0.20)
critique.append(
f'CHRONIC_LOW_BIAS: {self.BIAS_WINDOW} consecutive LOW '
f'with avg crowd={avg_crowd:.0f} persons. Upgraded to MEDIUM.'
)
# Bias 2: Rising trend ignored
if rr.trend == 'rising' and rr.risk_level == 'LOW' and fr.person_count > 20:
bias_detected = True
corrected_level = 'MEDIUM'
corrected_score = max(corrected_score, 0.25) # push into MEDIUM zone (>0.20)
critique.append(
f'RISING_TREND_IGNORED: trend=rising, persons={fr.person_count}, '
f'but risk=LOW. Upgraded to MEDIUM.'
)
# Bias 3: Count-risk mismatch (enhanced with critical threshold)
if fr.person_count >= self.CRITICAL_CROWD_THRESH and rr.risk_level == 'LOW':
# 70+ persons with LOW risk → CRITICAL inconsistency → upgrade to HIGH
bias_detected = True
corrected_level = 'HIGH'
corrected_score = max(corrected_score, 0.85) # push into HIGH zone (>0.80)
critique.append(
f'CRITICAL_COUNT_RISK_MISMATCH: {fr.person_count} persons but risk=LOW. '
f'This is a severe inconsistency. '
f'Upgraded to HIGH (corrected_score={corrected_score:.3f}).'
)
elif fr.person_count > self.HIGH_CROWD_THRESH and rr.risk_level == 'LOW':
# 50+ persons with LOW risk → upgrade to MEDIUM
bias_detected = True
corrected_level = 'MEDIUM'
corrected_score = max(corrected_score, 0.30) # push into MEDIUM zone (>0.20)
critique.append(
f'COUNT_RISK_MISMATCH: {fr.person_count} persons but risk=LOW. '
f'Thresholds may need recalibration. Upgraded to MEDIUM.'
)
# Bias 4: Over-estimation — HIGH risk but very few persons (downward correction)
if rr.risk_level == 'HIGH' and fr.person_count < config.REFLECTION_OVER_EST_THRESH:
bias_detected = True
corrected_level = 'MEDIUM'
corrected_score = min(corrected_score, 0.50) # pull down to MEDIUM zone
critique.append(
f'OVER_ESTIMATION: HIGH risk but only {fr.person_count} persons. '
f'Downgraded to MEDIUM — possible empty-frame or detection artifact.'
)
if not critique:
critique.append('OK: assessment consistent, no bias detected.')
reflection = {
'frame_id': rr.frame_id,
'original_level': rr.risk_level,
'original_score': rr.risk_score,
'corrected_level': corrected_level,
'corrected_score': round(corrected_score, 4),
'bias_detected': bias_detected,
'critique': ' | '.join(critique),
'person_count': fr.person_count,
}
self._reflection_log.append(reflection)
if bias_detected:
print(
f' 🪞 [ReflectionAgent] Frame {rr.frame_id}: '
f'{rr.risk_level}({rr.risk_score:.3f}) → '
f'{corrected_level}({corrected_score:.3f})'
)
print(f' {critique[0][:90]}')
return reflection
def get_summary(self) -> dict:
"""Summary stats — used by dashboard and evaluation section."""
total = len(self._reflection_log)
biased = sum(1 for r in self._reflection_log if r['bias_detected'])
return {
'total_reflections': total,
'bias_events': biased,
'bias_rate_pct': round(biased / total * 100, 1) if total > 0 else 0,
'corrections': {
'LOW->MEDIUM': sum(1 for r in self._reflection_log
if r['original_level'] == 'LOW'
and r['corrected_level'] == 'MEDIUM'),
},
}
|