告警与事件响应

当治理系统检测到异常时，需要快速告警并执行标准化的事件响应流程。

告警与响应流程

告警引擎

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timezone
from typing import Callable
class Severity(Enum):
P0_CRITICAL = 0
P1_HIGH = 1
P2_MEDIUM = 2
P3_LOW = 3
class AlertType(Enum):
JAILBREAK_BURST = "越狱攻击突增"
HIGH_BLOCK_RATE = "拦截率异常升高"
DATA_EXFILTRATION = "数据泄露尝试"
MODEL_DEGRADATION = "模型质量下降"
COST_SPIKE = "成本突增"
LATENCY_SPIKE = "延迟飙升"
@dataclass
class Alert:
alert_type: AlertType
severity: Severity
message: str
timestamp: str = ""
source: str = ""
metric_value: float = 0.0
threshold: float = 0.0
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now(timezone.utc).isoformat()
@dataclass
class AlertRule:
"""告警规则定义"""
name: str
alert_type: AlertType
severity: Severity
condition: Callable[[float], bool]
message_template: str
cooldown_seconds: int = 300   # 同类告警冷却期
class AlertEngine:
"""告警引擎——基于规则的实时告警"""
def __init__(self):
self.rules: list[AlertRule] = []
self.alerts: list[Alert] = []
self.last_fired: dict[str, float] = {}
self.handlers: dict[Severity, list[Callable[[Alert], None]]] = {
s: [] for s in Severity
}
def add_rule(self, rule: AlertRule) -> None:
self.rules.append(rule)
def register_handler(
self, severity: Severity, handler: Callable[[Alert], None]
) -> None:
"""注册告警处理器（按严重级别）"""
self.handlers[severity].append(handler)
def evaluate(self, metric_name: str, value: float) -> list[Alert]:
"""评估指标值，触发匹配的告警规则"""
triggered = []
now = datetime.now(timezone.utc).timestamp()
for rule in self.rules:
if rule.name != metric_name:
continue
if not rule.condition(value):
continue
# 冷却期检查
last = self.last_fired.get(rule.name, 0)
if now - last < rule.cooldown_seconds:
continue
alert = Alert(
alert_type=rule.alert_type,
severity=rule.severity,
message=rule.message_template.format(value=value),
source=metric_name,
metric_value=value,
threshold=0,
)
self.alerts.append(alert)
self.last_fired[rule.name] = now
triggered.append(alert)
# 调用对应级别的处理器
for handler in self.handlers[rule.severity]:
handler(alert)
return triggered
# 配置示例
engine = AlertEngine()
engine.add_rule(AlertRule(
name="block_rate",
alert_type=AlertType.HIGH_BLOCK_RATE,
severity=Severity.P1_HIGH,
condition=lambda v: v > 0.15,
message_template="拦截率达到 {value:.1%}，超过 15% 阈值",
cooldown_seconds=600,
))
engine.add_rule(AlertRule(
name="jailbreak_count_5min",
alert_type=AlertType.JAILBREAK_BURST,
severity=Severity.P0_CRITICAL,
condition=lambda v: v > 50,
message_template="5分钟内越狱尝试 {value:.0f} 次，触发紧急告警",
cooldown_seconds=300,
))

事件响应流程

from dataclasses import dataclass
from enum import Enum
class IncidentStatus(Enum):
OPENED = "opened"
INVESTIGATING = "investigating"
MITIGATING = "mitigating"
RESOLVED = "resolved"
POST_MORTEM = "post_mortem"
@dataclass
class Incident:
id: str
alert: Alert
status: IncidentStatus
assignee: str = ""
timeline: list[str] = field(default_factory=list)
root_cause: str = ""
resolution: str = ""
def update_status(self, new_status: IncidentStatus, note: str = "") -> None:
self.status = new_status
ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
self.timeline.append(f"[{ts}] {new_status.value}: {note}")
class IncidentManager:
"""事件管理器"""
PLAYBOOKS: dict[AlertType, list[str]] = {
AlertType.JAILBREAK_BURST: [
"1. 启用严格模式（提高检测灵敏度）",
"2. 临时限制可疑用户",
"3. 检查攻击模式是否为新型",
"4. 更新防御规则",
"5. 通知安全团队复查",
],
AlertType.DATA_EXFILTRATION: [
"1. 立即阻断相关会话",
"2. 评估泄露范围",
"3. 通知数据保护官（DPO）",
"4. 启动数据泄露应急预案",
"5. 72小时内向监管机构报告（GDPR）",
],
AlertType.HIGH_BLOCK_RATE: [
"1. 检查是否为误判导致的批量拦截",
"2. 分析被拦截请求的模式",
"3. 如果是误判，临时降低阈值",
"4. 更新关键词库或模型",
],
}
def __init__(self):
self.incidents: dict[str, Incident] = {}
self._counter = 0
def create_incident(self, alert: Alert) -> Incident:
self._counter += 1
incident_id = f"INC-{self._counter:04d}"
incident = Incident(
id=incident_id,
alert=alert,
status=IncidentStatus.OPENED,
)
incident.update_status(
IncidentStatus.OPENED,
f"由告警 {alert.alert_type.value} 自动创建"
)
self.incidents[incident_id] = incident
return incident
def get_playbook(self, alert_type: AlertType) -> list[str]:
"""获取标准化响应流程"""
return self.PLAYBOOKS.get(alert_type, ["无标准流程，请人工评估"])

告警级别定义

事件响应成熟度

本章小结

• 告警分四级（P0-P3）——不同级别对应不同响应时间和通知方式
• 规则引擎驱动告警——可配置条件 + 冷却期防止告警风暴
• 标准化 Playbook——每种告警类型有预定义的响应流程
• 事件生命周期管理——打开 → 调查 → 缓解 → 解决 → 复盘
• 自动化升级——P0/P1 自动触发熔断和限流

下一章：规则引擎设计