研究AIOps已有1年多,目前手里有不少可落地的方案了,而且目前我们已经有了自研的AIOps平台,欢迎关注并与我链接(看留言区)。
最近有不少朋友向我请教用AI如何做告警降噪?
我的观点:告警降噪 90% 的工作是规则、分组、抑制,剩下 10% 才轮到 AI。
今天这篇文章跟大家分享下我的方案四板斧,前三板斧跟AI没关系,全是工程事。

第一板斧:把 PromQL 和阈值写对
告警吵,一半是因为规则写得烂。三个最常见的毛病:没做平滑、没加 for、阈值拍脑袋。
拿 CPU 告警举例。磁盘使用率是"水位",慢慢爬,很少剧烈波动,设个阈值基本够用。但 CPU 是"流量",一秒一个值,瞬时冲到 90% 再掉下来是常态。直接拿原始值判断,误报能把人烦死。
看一段我改过的 CPU 告警:
groups:- name: node_alertsrules:- alert: CpuUsageCriticalexpr: avg by (instance) ((1 - rate(node_cpu_seconds_total{mode="idle"}[1m]) - rate(node_cpu_seconds_total{mode="iowait"}[1m])) * 100) > 95for: 1mlabels:severity: criticalannotations:summary: "{{ $labels.instance }} CPU 使用率超过 95%"- alert: CpuUsageHighexpr: avg by (instance) ((1 - rate(node_cpu_seconds_total{mode="idle"}[5m]) - rate(node_cpu_seconds_total{mode="iowait"}[5m])) * 100) > 85for: 10mlabels:severity: warningannotations:summary: "{{ $labels.instance }} CPU 使用率持续超过 85%"
三个关键点:
1. rate 窗口,级别越高窗口越短。
CPU 使用率是计数器派生的瞬时值,一秒一个样,必须用窗口算平均把尖峰抹平。但窗口不是越大越好:warning 用 5 分钟窗口看趋势,critical 用 1 分钟窗口抓突变。级别越高,窗口越短。
2. 排除 iowait,别把等待算成忙。
node_cpu_seconds_total 按 CPU 模式拆成多个序列,idle 是空闲占比。但直接 100 减 idle,会把 iowait 也当成 CPU 忙。磁盘慢的时候,CPU 大量时间在等 IO,iowait 能到 30%,这时候报"CPU 80%"就是误报。所以公式里把 idle 和 iowait 一起减掉,剩下的才是真忙。
3. for 分级,不是一刀切。
有人会问:CPU 真爆了,业务都挂了,还等 10 分钟?问得好。所以 for 要按级别分级,看上面两条规则:critical 用 1 分钟 for,冲到 95% 一分钟就告警,快速响应;warning 用 10 分钟 for,85% 持续十分钟才告警,过滤抖动。
一套规则两条腿走路:fast 负责抓真故障,slow 负责挡假报警。哪个都不能少。
相对阈值。
负载告警别用绝对值,用 load1 除以核数:
- alert: LoadHighexpr: node_load1 / count(node_cpu_seconds_total{mode="idle"}) > 0.8for: 15mlabels:severity: warningannotations:summary: "{{ $labels.instance }} 负载达到核数的 80%"
16 核和 4 核的机器,用同一个绝对值判断负载本来就不科学。除以核数之后,所有机器一套规则,不用每台单独调。
第二板斧:Alertmanager 分组 + 抑制
规则写对之后,还有一半告警是"一个故障炸出几十条"。靠 Alertmanager 解决。
分组配置:
route:receiver: defaultgroup_by: ['alertname', 'cluster'] ## 同一个告警名、同一个集群的告警,合并成一条通知。group_wait: 30s ## 等 30 秒,把同一批故障的告警攒一起再发,不一条条炸。group_interval: 5mrepeat_interval: 4h ## 告警持续期间,4 小时才重复提醒一次。不然你睡到半夜,手机每隔 5 分钟响一次。
抑制规则,这是降噪威力最大的一块:
inhibit_rules:# 主机挂了,抑制这台机器上所有非 critical 告警- source_matchers:- 'alertname="NodeDown"'target_matchers:- 'severity!="critical"'equal: ['instance']
解释一下:一台机器 down 了,node_exporter 抓不到数据,内存、磁盘、进程、服务健康检查全部一起告警,几十条。但你真正需要处理的只有一条"这台机器挂了"。这条抑制规则的效果:NodeDown 一旦触发,这台 instance 上所有非 critical 的告警全部闭嘴。
就这么一条规则,最狠的告警风暴直接按住。
第三板斧:用标签建依赖,把告警聚成事件
再往前走一步,是事件关联。
一个核心服务挂了,下游几十个服务跟着告警。你要做的不是去点几十条,而是让它们聚成一个事件。
实现方式不复杂,先给所有告警打上统一的 cluster 和 service 标签,然后在抑制规则里做上下游抑制:
# 上游 service 挂了,抑制下游所有依赖它的告警- source_matchers:- 'alertname="ServiceDown"'- 'severity="critical"'target_matchers:- 'severity!="critical"'equal: ['service']
注意这里的 equal 用的是 service,不是 cluster。如果 equal 写成 cluster,同一集群里 A 服务挂了,B 服务的告警也会被一起误抑制。只有当你确认"集群内服务强依赖、挂了会一起挂",才用 cluster。
值日的人看到的就不再是"50 条告警",而是"1 个事件:XX 服务异常,连带 40 个下游告警已抑制"。
这一步做完,告警量基本能降八九成。
第四板斧:AI 聚类,最后才上
前三板斧做完,告警量已经降了八九成。剩下的漏网之鱼,才轮到 AI。
AI 干两件事:把长得不一样但本质同一个根因的告警聚成一类,再对每个事件给根因判断和处理建议。
但这里有个关键问题:告警怎么给到 AI?
事件驱动:让 Alertmanager 在告警触发时,通过 webhook 主动推给一个分析服务,告警一到就分析。数据流是这样的:
Alertmanager 触发 → webhook POST 到分析服务 → 格式化 → 调 LLM → 返回聚类结果第一步,配 Alertmanager。 加一个专门的 webhook receiver,把告警路由到 AI 分析:
route:receiver: defaultgroup_by: ['alertname', 'cluster']group_wait: 30sgroup_interval: 5mrepeat_interval: 4hroutes:- matchers:- severity="critical"receiver: ai-analysisreceivers:- name: defaultwebhook_configs:- url: http://your-ops/webhook/dingtalk- name: ai-analysiswebhook_configs:- url: http://your-service:8080/webhook/alertssend_resolved: false
注意这里的路由只匹配 severity="critical",warning 不会喂给 AI。这是故意的:AI 分析要烧 token,warning 级别的漏网之鱼靠人工和报表兜底就行,不值得每次花这个钱。真正需要快速响应的,只有 critical。
这里有个巧妙的点:Alertmanager 的 group_wait 已经帮我们把同一批告警攒在一起了。它等 30 秒,把同组的告警聚成一批,一次性 POST 给 webhook。所以你的分析服务拿到的,天然就是"一个故障的一批告警",不用自己再做聚合。
第二步,写分析服务的 webhook。 收到 POST,格式化,调 LLM,把结果发到值班群:
from flask import Flask, requestimport requestsapp = Flask(__name__)def send_to_dingtalk(content):# 发到钉钉值班群,换成你们自己的机器人地址requests.post("https://oapi.dingtalk.com/robot/send?access_token=你的token",json={"msgtype": "text", "text": {"content": content}},)@app.post("/webhook/alerts")def handle_alerts():data = request.jsonfiring = [a for a in data["alerts"] if a["status"] == "firing"]if not firing:return "ok", 200# 格式化告警成文本lines = []for i, a in enumerate(firing, 1):labels = a["labels"]lines.append(f"告警{i}:{labels.get('alertname')} | "f"级别:{labels.get('severity')} | "f"实例:{labels.get('instance')} | "f"集群:{labels.get('cluster')} | "f"服务:{labels.get('service')} | "f"开始:{a.get('startsAt')} | "f"描述:{a.get('annotations', {}).get('summary', '')}")user_prompt = f"当前有 {len(firing)} 条告警正在触发:\n" + "\n".join(lines)# 调 LLMresult = requests.post("https://api.deepseek.com/v1/chat/completions",headers={"Authorization": "Bearer 你的key"},json={"model": "deepseek-chat","messages": [{"role": "system", "content": SYSTEM_PROMPT},{"role": "user", "content": user_prompt},],"temperature": 0.2,},)analysis = result.json()["choices"][0]["message"]["content"]send_to_dingtalk(analysis)return "ok", 200if __name__ == "__main__":app.run(host="0.0.0.0", port=8080)
关键点:一定要把 labels 里的集群、服务、实例带上。 AI 聚类靠的就是这些上下文。没有集群和服务标签,AI 分不清哪些告警在同一条链路上,聚类就是瞎分。
System Prompt 是核心:
你是一个资深 SRE 告警分析师。我会给你一批当前正在触发的告警,每条包含告警名、级别、实例、集群、服务、开始时间和描述。请执行以下分析:1. 把这批告警聚类成事件,判断哪些告警属于同一个根因。2. 对每个事件,给出最可能的根因判断,并说明推理依据。3. 给出处理建议,按优先级排序。4. 用 JSON 格式输出,包含字段:event_id、root_cause、alerts、severity、action。
AI 返回的 JSON 长这样:
{"events": [{"event_id": "event-1","root_cause": "api 服务所在宿主机 CPU 饱和,导致健康检查超时","alerts": ["CpuUsageHigh", "ServiceDown", "HealthCheckFailed"],"severity": "critical","action": "先排查宿主机 CPU 占用进程,或对 api 服务扩容"}]}
这下值班的人看到的不是几十条告警,而是一个结论:根因是什么、先处理什么。
注意边界:AI 聚类基于的是 labels 和描述文本,它只能做"归类"和"根因推测",不能替代真实故障的最终判断。 但它能把几十条告警收敛成一个事件加一个建议,这就够了。
这里还有个天生短板:webhook 是按组触发的,同一批 group 内的告警能聚在一起,但跨组的关联它看不到。
比如磁盘慢导致 MySQL 慢查询超时,再导致订单服务告警,三组告警会分三次 POST 进来,各分析各的,聚不成一个根因。要跨组聚类,得定期全量拉一遍所有告警做二次聚合,代价是回到轮询。这就是事件驱动和完整聚类的取舍,先知道它做不到什么,才知道怎么用它。
告警降噪做到最后,你会发现 90% 是工程问题,AI 只负责最后 10%。先把规则写对,把分组配好,把抑制建起来。这些事不做,AI 救不了你。
最后介绍下我新上的《AIOps极简入门》小课,这套AIOps智能体落地案例集是我大模型课程里AIOps章节中的10个案例,现在单独把它拆出来售卖!
虽然只有10个小案例,但价值巨大!
