一条 P0 告警被晾了 8 分钟，就因为 AI 分类错了

# 原始告警
告警名称: MySQLDown | 服务: order-db | 环境: production
摘要: MySQL 主库连接失败，已持续 2 分钟

# AI 分类结果（错误）
分类: infrastructure  ← ❌ 应该分类到 database
置信度: 0.52         ← ⚠️ 阈值设得太低

# 路由结果（错误）
路由到: 基础设施值班群 ← ❌ 应该路由到 DBA 组
响应时间: 8 分 12 秒  ← ❌ P0 要求 < 5 分钟

# 检查分类器对 MySQLDown 的匹配
alert_text = "告警名称: MySQLDown 服务: order-db 摘要: MySQL 主库连接失败"

for cat, prototypes in CATEGORY_PROTOTYPES.items():
    proto_embs = category_embeddings[cat]
    a_emb = model.encode([alert_text])
    sim = np.max(np.dot(proto_embs, a_emb.T))
print(f"{cat}: 最高相似度 {sim:.4f}")

# 输出:
# database:  0.48  ← 低，因为没包含"MySQL主库"原型
# network:   0.31
# application: 0.35
# infrastructure: 0.52  ← 最高！因为"连接失败"像网络问题

# 修复 1：增强数据库分类原型
CATEGORY_PROTOTYPES["database"] = [
"MySQL 连接失败", "MySQLDown", "主库宕机",
"数据库不可用", "connection refused",
"replication lag", "主从延迟",
]

# 修复 2：置信度拒绝策略
defclassify_with_fallback(alert_text, threshold=0.65):
    results = classify_alert(alert_text, top_k=2)

if results[0]["max_similarity"] < threshold:
return {"category": "pending_review", "fallback": True}

# P0/P1 告警即使分类低也通知所有相关群
if alert.severity in ("P0", "P1") and results[0]["max_similarity"] < 0.75:
return {"category": results[0]["category"], "notify_all": True}

return results[0]

# 修复 3：P0 告警广播通知
P0_BROADCAST_GROUPS = ["database", "infrastructure", "application"]

修复后 MySQLDown 分类:
- database 相似度: 0.81 ✅
- 分类准确率: 从 72% → 94%
- P0 误路由率: 从 8% → 0.5%

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("shibing624/text2vec-base-chinese")

# 定义分类原型
CATEGORY_PROTOTYPES = {
"database": [
"MySQL 数据库连接失败", "Redis 缓存命中率过低",
"MongoDB 副本集主从切换", "数据库慢查询数量超过阈值",
    ],
"network": [
"CDN 节点不可达", "负载均衡后端健康检查失败",
"防火墙规则导致连接被拒", "DNS 解析超时",
    ],
"application": [
"Spring Boot 服务不可用", "HTTP 接口错误率超过阈值",
"应用 OOM 内存溢出", "接口响应时间超时",
    ],
"infrastructure": [
"Kubernetes Pod CrashLoopBackOff", "服务器 CPU 使用率过高",
"磁盘 IO 等待时间过长", "节点 NotReady 状态",
    ],
}

# 预计算原型向量
category_embeddings = {}
for cat, descriptions in CATEGORY_PROTOTYPES.items():
    category_embeddings[cat] = model.encode(descriptions)

defclassify_alert(alert_text: str) -> dict:
"""对告警进行语义分类"""
    query_embedding = model.encode([alert_text])

    best_cat, best_score = None, 0
for cat, proto_embeddings in category_embeddings.items():
        similarities = np.dot(proto_embeddings, query_embedding.T).flatten()
        max_sim = float(np.max(similarities))
if max_sim > best_score:
            best_score = max_sim
            best_cat = cat

return {"category": best_cat, "confidence": round(best_score, 4)}

# 测试
test_alert = "告警名称: MySQLDown | 服务: order-db | 环境: production | 摘要: MySQL 主库连接失败"
result = classify_alert(test_alert)
print(f"分类: {result['category']} (置信度: {result['confidence']})")

# 新增中间件分类
CATEGORY_PROTOTYPES["middleware"] = [
"Kafka 消息积压", "RabbitMQ 队列堆积",
"RocketMQ 消费延迟", "ZooKeeper 会话过期",
]

# 低于阈值归为未分类
defclassify_with_threshold(alert_text: str, threshold: float = 0.45):
    results = classify_alert(alert_text)
if results["confidence"] < threshold:
return {"category": "uncategorized"}
return results

defget_oncall_person(team: str) -> dict:
"""从企业通讯录获取当前值班人员"""
    resp = requests.get(
f"http://oncall-api.internal/api/v1/schedules",
        params={"team": team, "current": "true"}
    )
    data = resp.json()
return {
"name": data["oncall"]["name"],
"phone": data["oncall"]["phone"],
"dingtalk_id": data["oncall"]["dingtalk_id"],
    }

routes:
-match:
ai_category:database
ai_severity:P0
receiver:db-team-urgent
group_wait:10s
repeat_interval:5m

-match:
ai_category:database
receiver:db-team-normal
group_wait:30s
repeat_interval:30m

-match:
ai_category:application
ai_severity:P0
receiver:app-team-urgent
group_wait:10s
repeat_interval:5m