AI生成运维告警通知，故障信息格式化推送
传统运维告警往往只推一段原始日志，值班人员需要手动翻看上下文才能判断严重程度。
本文介绍如何用AI将原始告警格式化成结构清晰的通知，并通过Webhook推送到钉钉或企业微信。
适合刚接触自动化运维、希望减少告警噪音的零基础读者。
整体思路与前置准备
核心流程分三步：监控系统触发Webhook，中间脚本调用大模型API整理告警内容，最后将格式化结果推送到IM工具。
你需要准备：
- 一台能访问外网的Linux服务器（Ubuntu 20.04+或CentOS 7+）。
- Python 3.8以上环境，安装
requests库：pip3 install requests。 - 一个大模型API Key（如OpenAI、通义千问、DeepSeek等，以官方文档为准）。
- 一个钉钉或企业微信群机器人Webhook地址。
- 监控系统告警Webhook配置权限（如Prometheus Alertmanager、Zabbix）。
验证点：在服务器上执行python3 -c "import requests; print(requests.__version__)"，能输出版本号即可。
编写告警格式化脚本
在/opt/ai-alert/目录下创建format_alert.py，核心逻辑是接收告警JSON，拼接提示词，调用大模型API，再推送。
import json, requests, sys
LLM_API_KEY = "你的API_KEY"
LLM_URL = "https://api.example.com/v1/chat/completions" # 以实际服务商为准
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
def format_with_ai(raw_alert):
prompt = f"""请将以下运维告警整理成简洁通知，包含：故障级别、影响服务、发生时间、可能原因、建议操作。用中文输出，不超过150字。
告警原文：{raw_alert}"""
headers = {"Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json"}
data = {"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3}
resp = requests.post(LLM_URL, headers=headers, json=data, timeout=15)
return resp.json()["choices"][0]["message"]["content"]
def send_to_dingtalk(text):
payload = {"msgtype": "text", "text": {"content": text}}
requests.post(DINGTALK_WEBHOOK, json=payload, timeout=10)
if __name__ == "__main__":
raw = sys.stdin.read()
formatted = format_with_ai(raw)
send_to_dingtalk(formatted)关键点：LLM_URL和model字段需替换为你实际使用的服务商参数，temperature设为0.3可让输出更稳定。
接入监控系统并测试推送
以Prometheus Alertmanager为例，编辑alertmanager.yml，添加Webhook接收器：
receivers:
- name: 'ai-webhook'
webhook_configs:
- url: 'http://你的服务器IP:5000/alert'
send_resolved: true然后写一个Flask接收端receiver.py：
from flask import Flask, request
import subprocess
app = Flask(__name__)
@app.route('/alert', methods=['POST'])
def alert():
raw = request.get_data(as_text=True)
subprocess.run(['python3', '/opt/ai-alert/format_alert.py'], input=raw, text=True)
return 'ok', 200
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)启动接收端：nohup python3 receiver.py &。
手动触发一条告警测试，观察钉钉群是否收到整理后的消息。
验证方式：
在Alertmanager界面点击“Silence”创建一个测试告警，
或直接curl -X POST -d '{"alert":。
"CPU过高"}' http:
//localhost:
5000/alert
避坑与稳定性建议
- 大模型API超时：设置
timeout=15，并在脚本中加try...except，超时后降级为推送原始告警，避免告警丢失。 - 敏感信息泄露：告警原文可能包含IP、密码，调用外部API前建议先脱敏，或使用私有化部署模型。
- 频率限制：大模型API通常有QPS限制，可在脚本中加
time.sleep(1)或使用队列缓冲。 - 推送失败重试：钉钉机器人有频率限制，建议捕获异常后记录日志，不要无限重试。
- 提示词稳定性：不同模型对同一提示词输出差异较大，建议固定模型版本，并保存几次输出做对比。
效果检查与后续优化
完成配置后，人为制造一次真实告警（如停掉一个测试服务），确认钉钉收到格式统一的通知，字段包含级别、服务名、时间、原因和建议。
如果输出内容过长，可调整提示词中的字数限制；
如果分类不准，可在提示词中给出示例。
常见疑问
问：没有公网服务器能跑吗？
答：可以，只要服务器能访问大模型API和钉钉Webhook即可，内网机器需配置代理。
问：必须用Python吗？
答：不是，Shell、Go都可以，核心是HTTP请求和JSON处理。
问：大模型API收费吗？
答：多数服务商有免费额度或按量计费，具体以官方定价为准。
这套方案的价值在于把零散告警变成可读通知，减少值班人员判断成本。
建议先在测试环境跑通，再逐步接入生产告警。
