fab65e7d7a
Some checks are pending
CD Pipeline / build-and-deploy (push) Has started running
根因：find_by_fingerprint 的 PENDING 匹配條件無時間上限， 2026-04-12 建立的 3 筆 PENDING approval records（hit=77/30/17） 持續吃掉所有同指紋告警，造成 2+ 小時 Telegram 靜音。 修正（approval_db.py）： - PENDING_TTL_HOURS = 24：PENDING 記錄逾 24h 不再收斂新告警 - 原本：OR(status=PENDING, created_at>=30min前) - 修正：OR(PENDING AND created_at>=24h前, created_at>=30min前) 緊急修復：kubectl exec 直接將 7 筆過期 PENDING 記錄設為 expired， 即時恢復 Telegram 告警流（不等部署）。 Phase 6 AI 自我治理閉環（ADR-087）： - feat(db): 新增 ai_governance_events 表 + 3 個 index（base.py + models.py） - feat(svc): ai_slo_calculator.py — 7d 滾動 SLO（success/override/false_neg） - feat(svc): trust_drift_detector.py — Playbook 信任度極端偏態偵測 - feat(job): kb_rot_cleaner.py — K8s API/Prom metric/老舊 incident_case 腐爛清理 - feat(svc): decision_manager.py — 自我降級守衛（SLO 違反 → 提高門檻/保守模式） 2026-04-15 ogt + Claude Sonnet 4.6（亞太） Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
244 lines
10 KiB
Python
244 lines
10 KiB
Python
"""
|
||
AWOOOI AIOps Phase 6 — Trust Drift Detector（信任度漂移偵測器）
|
||
===============================================================
|
||
職責：偵測 Playbook trust_score 分布的兩種極端偏態：
|
||
|
||
極端 A「盲目樂觀」：> 70% Playbook trust_score > 0.9
|
||
→ 可能是 PostExecutionVerifier 失效，或 RAG 資料被污染，讓所有 AI 都以為「我很棒」
|
||
→ 真正的好系統不會所有 Playbook 都高分
|
||
|
||
極端 B「學習鎖死」：> 70% Playbook trust_score < 0.3
|
||
→ 可能是 EWMA 計算出錯，或所有執行都被誤判失敗，讓 AI 對自己完全沒信心
|
||
→ 學習機制可能卡死
|
||
|
||
設計原則：
|
||
1. 只讀 DB，不修改任何數據
|
||
2. 違反 → 寫 trust_drift 事件到 ai_governance_events
|
||
3. 樣本不足（< 10 個 approved Playbook）→ 跳過偵測，不告警
|
||
|
||
ADR-087: AI 自我治理閉環
|
||
2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 6 初始建立
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
|
||
import structlog
|
||
from sqlalchemy import func, select
|
||
|
||
from src.db.base import get_session_factory
|
||
from src.db.models import AiGovernanceEvent, PlaybookRecord
|
||
from src.utils.timezone import now_taipei
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 偵測閾值（MASTER §3.6，修改需 ADR-087 更新）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
DRIFT_HIGH_THRESHOLD: float = 0.9 # trust_score > 此值算「過高」
|
||
DRIFT_LOW_THRESHOLD: float = 0.3 # trust_score < 此值算「過低」
|
||
DRIFT_RATIO_TRIGGER: float = 0.70 # 超過 70% Playbook 落在極端 → 觸發警報
|
||
DRIFT_MIN_SAMPLES: int = 10 # 最少 approved Playbook 數量
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Data Types
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
@dataclass
|
||
class TrustDistribution:
|
||
"""Playbook 信任度分布快照"""
|
||
total: int
|
||
high_count: int # trust_score > 0.9
|
||
low_count: int # trust_score < 0.3
|
||
mid_count: int # 0.3 <= trust_score <= 0.9（正常區間）
|
||
high_ratio: float
|
||
low_ratio: float
|
||
mean_trust: float
|
||
drift_type: str | None # "optimism_bias" / "confidence_collapse" / None
|
||
drift_detected: bool
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"total": self.total,
|
||
"high_count": self.high_count,
|
||
"low_count": self.low_count,
|
||
"mid_count": self.mid_count,
|
||
"high_ratio": round(self.high_ratio, 4),
|
||
"low_ratio": round(self.low_ratio, 4),
|
||
"mean_trust": round(self.mean_trust, 4),
|
||
"drift_type": self.drift_type,
|
||
"drift_detected": self.drift_detected,
|
||
"thresholds": {
|
||
"high": DRIFT_HIGH_THRESHOLD,
|
||
"low": DRIFT_LOW_THRESHOLD,
|
||
"ratio_trigger": DRIFT_RATIO_TRIGGER,
|
||
"min_samples": DRIFT_MIN_SAMPLES,
|
||
},
|
||
}
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Main Service
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TrustDriftDetector:
|
||
"""
|
||
信任度漂移偵測器
|
||
|
||
Usage:
|
||
detector = TrustDriftDetector()
|
||
dist = await detector.detect()
|
||
if dist.drift_detected:
|
||
await detector.save_drift_event(dist)
|
||
"""
|
||
|
||
async def detect(self) -> TrustDistribution:
|
||
"""
|
||
讀取所有 approved Playbook，計算信任度分布，偵測漂移。
|
||
|
||
Returns:
|
||
TrustDistribution（樣本不足時 drift_detected=False）
|
||
"""
|
||
try:
|
||
async with get_session_factory()() as session:
|
||
# 只計算 approved 狀態的 Playbook
|
||
total_q = await session.execute(
|
||
select(func.count()).where(
|
||
PlaybookRecord.status == "approved"
|
||
)
|
||
)
|
||
total: int = total_q.scalar() or 0
|
||
|
||
if total < DRIFT_MIN_SAMPLES:
|
||
logger.info(
|
||
"trust_drift_skip_insufficient_samples",
|
||
total=total,
|
||
required=DRIFT_MIN_SAMPLES,
|
||
)
|
||
return TrustDistribution(
|
||
total=total,
|
||
high_count=0, low_count=0, mid_count=0,
|
||
high_ratio=0.0, low_ratio=0.0, mean_trust=0.0,
|
||
drift_type=None, drift_detected=False,
|
||
)
|
||
|
||
high_q = await session.execute(
|
||
select(func.count()).where(
|
||
PlaybookRecord.status == "approved",
|
||
PlaybookRecord.trust_score > DRIFT_HIGH_THRESHOLD,
|
||
)
|
||
)
|
||
high_count: int = high_q.scalar() or 0
|
||
|
||
low_q = await session.execute(
|
||
select(func.count()).where(
|
||
PlaybookRecord.status == "approved",
|
||
PlaybookRecord.trust_score < DRIFT_LOW_THRESHOLD,
|
||
)
|
||
)
|
||
low_count: int = low_q.scalar() or 0
|
||
|
||
mean_q = await session.execute(
|
||
select(func.avg(PlaybookRecord.trust_score)).where(
|
||
PlaybookRecord.status == "approved"
|
||
)
|
||
)
|
||
mean_trust: float = float(mean_q.scalar() or 0.0)
|
||
|
||
mid_count = total - high_count - low_count
|
||
high_ratio = high_count / total
|
||
low_ratio = low_count / total
|
||
|
||
# 偵測漂移類型
|
||
drift_type = None
|
||
if high_ratio >= DRIFT_RATIO_TRIGGER:
|
||
drift_type = "optimism_bias" # 所有 Playbook 都覺得自己很好 → 可疑
|
||
elif low_ratio >= DRIFT_RATIO_TRIGGER:
|
||
drift_type = "confidence_collapse" # AI 對自己完全沒信心 → 學習卡死
|
||
|
||
dist = TrustDistribution(
|
||
total=total,
|
||
high_count=high_count,
|
||
low_count=low_count,
|
||
mid_count=mid_count,
|
||
high_ratio=high_ratio,
|
||
low_ratio=low_ratio,
|
||
mean_trust=mean_trust,
|
||
drift_type=drift_type,
|
||
drift_detected=drift_type is not None,
|
||
)
|
||
|
||
if dist.drift_detected:
|
||
logger.warning(
|
||
"trust_drift_detected",
|
||
drift_type=drift_type,
|
||
high_ratio=round(high_ratio, 3),
|
||
low_ratio=round(low_ratio, 3),
|
||
mean_trust=round(mean_trust, 3),
|
||
total=total,
|
||
)
|
||
else:
|
||
logger.info(
|
||
"trust_drift_ok",
|
||
mean_trust=round(mean_trust, 3),
|
||
total=total,
|
||
high_ratio=round(high_ratio, 3),
|
||
)
|
||
|
||
return dist
|
||
|
||
except Exception as e:
|
||
logger.error("trust_drift_detect_error", error=str(e))
|
||
# 保守：偵測失敗 → 不告警（不知道比亂告警好）
|
||
return TrustDistribution(
|
||
total=0,
|
||
high_count=0, low_count=0, mid_count=0,
|
||
high_ratio=0.0, low_ratio=0.0, mean_trust=0.0,
|
||
drift_type=None, drift_detected=False,
|
||
)
|
||
|
||
async def save_drift_event(self, dist: TrustDistribution) -> None:
|
||
"""將信任度漂移事件寫入 ai_governance_events。"""
|
||
try:
|
||
async with get_session_factory()() as session:
|
||
event = AiGovernanceEvent(
|
||
event_type="trust_drift",
|
||
details={
|
||
**dist.to_dict(),
|
||
"detected_at": now_taipei().isoformat(),
|
||
},
|
||
resolved=False,
|
||
)
|
||
session.add(event)
|
||
await session.commit()
|
||
logger.warning(
|
||
"trust_drift_event_saved",
|
||
drift_type=dist.drift_type,
|
||
)
|
||
except Exception as e:
|
||
logger.error("trust_drift_event_save_error", error=str(e))
|
||
|
||
async def run(self) -> TrustDistribution:
|
||
"""完整執行：偵測 → 如有漂移則寫事件。"""
|
||
dist = await self.detect()
|
||
if dist.drift_detected:
|
||
await self.save_drift_event(dist)
|
||
return dist
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Singleton
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_detector: TrustDriftDetector | None = None
|
||
|
||
|
||
def get_trust_drift_detector() -> TrustDriftDetector:
|
||
global _detector
|
||
if _detector is None:
|
||
_detector = TrustDriftDetector()
|
||
return _detector
|