前言
前面写过使用夜莺作为 Prometheus 告警引擎的文章，在此基础上接入 DeepSeek 进行故障分析，给出可能引起故障的原因和排查方法，大大降低平均故障处理时间（MTTR）。
工作流程：Prometheus周期采集Exporter的指标并存储在本地，夜莺周期查询Prometheus中的指标是否符合故障告警规则，在产生故障后将故障信息发送给 DeepSeek ，DeepSeek通过分析给出的故障原因和处理建议，拼接原告警信息和DeepSeek分析结果一并发给用户。对于没有使用夜莺的小伙伴可以直接使用webhook来执行脚本。
「流程图」
1. 新建通知模板
告警通知 --> 通知模板 --> 新增, 新增aiops模板
#### {{if .IsRecovered}}<font color="#008800">? {{.RuleName}}恢复</font>{{else}}<font color="#FF0000">? {{.RuleName}}告警</font>{{end}}
---
**级别状态**：{{if .IsRecovered}}<font color="#008800">S{{.Severity}}</font>{{else}}<font color="#FF0000">S{{.Severity}}</font>{{end}}
{{if eq (index .TagsMap "job") "web_status"}}
**归属公司**：{{index .TagsMap "company"}}
**项目名称**：{{index .TagsMap "project_cn"}}
**系统名称**：{{index .TagsMap "name"}}
{{if .IsRecovered}}**恢复内容**：{{index .TagsMap "name"}} 当前已恢复正常！
{{else}}**告警内容**：{{index .TagsMap "name"}} 当前无法访问！
{{end}}
**系统地址**：[{{index .TagsMap "instance"}}]({{index .TagsMap "instance"}})
{{end}}
{{if .IsRecovered}}**触发时间**：{{timeformat .FirstTriggerTime}}
**恢复时间**：{{timeformat .LastEvalTime}}{{else}}**触发时间**：{{timeformat .FirstTriggerTime}}{{end}}
2. 新建通知媒介
告警通知 --> 通知设置 --> 通知媒介 --> 添加
名称： aiops标识： aiops
3. 配置通知脚本
告警通知 --> 通知设置 --> 通知脚本 --> 使用脚本
#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys
import json
import requests
class Sender(object):
@classmethod
def send_email(cls, payload):
# already done in go code
pass
@classmethod
def send_wecom(cls, payload):
# already done in go code
pass
# 钉钉机器人
DINGTALK_URL = "https://oapi.dingtalk.com/robot/send?access_token=XXXXXXXXXX"
# DeepSeek key
DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"
DEEPSEEK_KEY = "sk-XXXXXXXXXX"
@classmethod
def call_deepseek(cls, message):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {cls.DEEPSEEK_KEY}"
}
data = {
"model": "deepseek-chat",
"messages": [{
"role": "user",
"content": f"""
告警信息：{message}
你是一名运维领域的专家，请分析告警信息给出可能原因、处理建议和紧急程度
排版要求：AI故障分析标题为蓝色h4大小，语言简洁突出重点
"""
}]
}
try:
response = requests.post(cls.DEEPSEEK_URL, headers=headers, json=data)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
except Exception as e:
print(f"Deepseek API error: {str(e)}")
return"无法获取处理建议"
@classmethod
def send_dingtalk(cls, payload):
original_message = payload.get('tpls').get("dingtalk", "dingtalk not found")
analysis = cls.call_deepseek(original_message)
final_message = f"""{original_message}
---
{analysis}
"""
headers = {
"Content-Type": "application/json;charset=utf-8"
}
body = {
"msgtype": "markdown",
"markdown": {
"title": "告警通知",
"text": final_message
}
}
response = requests.post(cls.DINGTALK_URL, headers=headers, data=json.dumps(body))
print(f"notify_dingtalk: status_code={response.status_code} response_text={response.text}")
@classmethod
def send_mm(cls, payload):
# already done in go code
pass
@classmethod
def send_sms(cls, payload):
pass
@classmethod
def send_voice(cls, payload):
pass
def main():
payload = json.load(sys.stdin)
with open(".payload", 'w') as f:
f.write(json.dumps(payload, indent=4))
for ch in payload.get('event').get('notify_channels'):
send_func_name = "send_{}".format(ch.strip())
ifnot hasattr(Sender, send_func_name):
print("function: {} not found", send_func_name)
continue
send_func = getattr(Sender, send_func_name)
send_func(payload)
def hello():
print("hello nightingale")
if __name__ == "__main__":
if len(sys.argv) == 1:
main()
elif sys.argv[1] == "hello":
hello()
else:
print("I am confused")
DeepSeek 分析回答需要一定时间，超时时间建议配置 120s。
4. 告警规则启用 AIOps 通知方式
告警管理 --> 告警规则 ，原有的规则勾上aiops通知媒介
5. 告警测试
原告警信息
接入大模型后的告警信息
在这里只是简单的将告警信息发送给 DeepSeek 大模型进行分析然后将分析的结果发送给我们，分析的结果一般没办法完全符合我们的预期。下一步需要进行「RAG」(检索增强生成)，通过检索外部知识库的方式，将知识库中的相关内容作为提示输入给大模型，从而给出更加符合我们预期的分析结果。在AIOps建设的过程中，运维知识库的建设将非常重要，「RAG」的结果符合我们的预期后，再结合工作流、AI Agent的方式进行告警收敛、故障自愈等操作。相关开源工具 --- dify
