宕机之后十分钟自动出报告：大语言模型驱动的服务宕机根因分析报告自动生成

宕机之后十分钟自动出报告大语言模型驱动的服务宕机根因分析报告自动生成凌晨2点15分告警系统跳出红色弹窗订单服务不可用。三分钟后我还在揉眼睛的时候钉钉群里已经收到了一份《订单服务宕机根因分析报告初稿》。从告警触发到报告生成不到10分钟。这不是科幻片这是我们基于大语言模型LLM搭建的根因分析报告自动生成系统。一、为什么需要自动生成根因分析报告传统故障复盘的低效每次线上故障后的复盘流程01:00 告警触发开始排查 01:30 初步定位开始止血 02:00 恢复服务 02:30 开始写复盘报告 03:30 报告写完漏了一半细节 04:00 拉群讨论补充信息 ... 三天后报告终于归档但没人再看这个流程的问题时效性差复盘报告通常是事后补的关键细节已经遗忘质量不一取决于值班工程师的记忆力和文档水平缺乏标准化有的报告详细有的报告就两行字知识沉淀难上次怎么修的这次又得重新排查自动报告的目标我们给自动报告系统定了三个目标快告警后10分钟内输出初稿准以监控数据和日志为事实依据全涵盖故障发现、影响范围、根因分析、修复措施全流程二、系统架构设计[告警事件] → 事件感知器 ↓ [数据收集层] → 时序数据采集 → API Server 日志采集 → ELK Client 变更记录 → CMDB Client 链路追踪 → Jaeger Client ↓ [分析层] → 时序异常检测 日志关键信息提取 变更关联分析 ↓ [LLM层] → 上下文组装器 → 大模型API → 报告生成器 ↓ [输出层] → 钉钉/企微推送 Confluence自动归档 复盘会议邀请自动发送三、核心实现事件感知与上下文收集# incident_collector.py — 故障上下文收集器 import asyncio from datetime import datetime, timedelta import aiohttp import json class IncidentContextCollector: 收集故障上下文信息 def __init__(self, config: dict): self.prometheus_url config[prometheus_url] self.es_url config[elasticsearch_url] self.cmdb_url config[cmdb_url] async def collect_all(self, alert_event: dict) - dict: 并发收集所有上下文数据 start_time datetime.fromisoformat(alert_event[start_time]) window_start start_time - timedelta(minutes30) window_end start_time timedelta(minutes5) # 并发执行所有数据收集 tasks [ self.collect_metrics(alert_event, window_start, window_end), self.collect_logs(alert_event, window_start, window_end), self.collect_changes(alert_event, window_start, window_end), self.collect_traces(alert_event, window_start, window_end) ] results await asyncio.gather(*tasks) return { alert: alert_event, metrics: results[0], logs: results[1], changes: results[2], traces: results[3], collected_at: datetime.now().isoformat() } async def collect_metrics(self, alert, start, end): 采集异常时段前后的时序指标 queries { cpu: sum(rate(container_cpu_usage_seconds_total{namespaceprod}[1m])) by (pod), memory: sum(container_memory_working_set_bytes{namespaceprod}) by (pod), latency: histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{serviceorder}[5m])), error_rate: sum(rate(http_requests_total{serviceorder, status~5..}[5m])) / sum(rate(http_requests_total{serviceorder}[5m])) } results {} async with aiohttp.ClientSession() as session: for name, query in queries.items(): params { query: query, start: start.timestamp(), end: end.timestamp(), step: 15 } async with session.get(f{self.prometheus_url}/api/v1/query_range, paramsparams) as resp: data await resp.json() results[name] data[data][result] return results async def collect_logs(self, alert, start, end): 采集异常时段的错误日志 logs_query { query: { bool: { must: [ {match: {service: order}}, {match: {level: ERROR}} ], filter: [ {range: {timestamp: { gte: start.isoformat(), lte: end.isoformat() }}} ] } }, size: 50, sort: [{timestamp: desc}] } async with aiohttp.ClientSession() as session: async with session.post( f{self.es_url}/order-logs-*/_search, jsonlogs_query ) as resp: result await resp.json() return [ {timestamp: hit[_source][timestamp], message: hit[_source][message]} for hit in result[hits][hits] ]LLM报告生成引擎# report_generator.py — 报告生成引擎 import json from openai import AsyncOpenAI class IncidentReportGenerator: 基于LLM生成根因分析报告 def __init__(self, api_key: str, model: str qwen2-72b): self.client AsyncOpenAI( api_keyapi_key, base_urlhttp://llm-service:8000/v1 ) self.model model def build_report_prompt(self, context: dict) - str: 构建结构化报告提示词 return f你是一位资深的SRE故障复盘专家请根据以下故障数据生成根因分析报告。 ## 故障基本信息 - 告警名称{context[alert][name]} - 告警时间{context[alert][start_time]} - 告警级别{context[alert][severity]} - 影响服务{context[alert][service]} - 告警状态{context[alert][status]} ## 时序指标异常故障时间窗口内 {json.dumps(context[metrics], indent2, ensure_asciiFalse)[:2000]} ## 异常日志摘要Top 20 {json.dumps(context[logs][:20], indent2, ensure_asciiFalse)} ## 最近变更记录故障前1小时内 {json.dumps(context[changes], indent2, ensure_asciiFalse)} ## 链路追踪异常 {json.dumps(context[traces][:5], indent2, ensure_asciiFalse)} 请严格按照以下Markdown格式输出报告不要添加额外内容 # 故障根因分析报告 ## 一、故障概览 - 故障编号INC-{context[alert][id]} - 发生时间{context[alert][start_time]} - 恢复时间[根据数据推断] - 故障时长[推断] - 影响范围[分析SLA影响] - 严重级别P0/P1/P2 ## 二、故障时间线 | 时间 | 事件 | 数据来源 | |------|------|---------| | ... | ... | ... | ## 三、根因分析 ### 3.1 直接原因 [基于日志和指标的直接原因] ### 3.2 根本原因 [深入分析包含变更关联] ### 3.3 触发条件 [触发故障的完整条件链] ## 四、影响评估 - 受影响请求数[数据驱动] - 平均恢复时间[数据驱动] - 业务影响[定性描述] ## 五、修复措施 - 止血操作[具体操作操作人时间] - 长期修复[代码/配置变更建议] ## 六、后续改进 ### 6.1 监控改进 [新增告警规则建议] ### 6.2 流程改进 [变更流程/发布流程改进建议] ### 6.3 技术改进 [架构改进建议] async def generate_report(self, context: dict) - str: 异步生成报告 prompt self.build_report_prompt(context) response await self.client.chat.completions.create( modelself.model, messages[ {role: system, content: 你是SRE专家严格执行报告格式。}, {role: user, content: prompt} ], temperature0.1, max_tokens4000 ) return response.choices[0].message.content async def review_and_refine(self, report: str, raw_data: dict) - str: 让LLM自我审查修正不准确的地方 review_prompt f请审查以下根因分析报告确保 1. 所有数据引用与原始数据一致 2. 根因结论有充分证据支持 3. 修复措施具体可执行 原始数据摘要{json.dumps(raw_data, indent2, ensure_asciiFalse)[:1000]} 报告内容 {report} 请直接在原报告基础上修正只修改不准确的部分。 response await self.client.chat.completions.create( modelself.model, messages[{role: user, content: review_prompt}], temperature0.1, max_tokens4000 ) return response.choices[0].message.content报告推送与归档# pusher.py — 自动推送报告 import requests from confluence_client import ConfluenceClient class ReportPusher: 报告推送与归档 def push_to_dingtalk(self, report: str, webhook_url: str): 推送到钉钉群 # 提取摘要信息 summary_section report.split(## 一、故障概览)[1].split(##)[0] if ## 一、故障概览 in report else report[:500] payload { msgtype: markdown, markdown: { title: 根因分析报告已生成, text: f### 故障根因分析报告\n\n{summary_section}\n\n---\n*报告由AI自动生成请人工复核* } } requests.post(webhook_url, jsonpayload) def archive_to_confluence(self, report: str, incident_id: str): 归档到Confluence client ConfluenceClient(urlhttps://wiki.example.com, token...) client.create_page( spaceSRE, titlef根因分析报告-{incident_id}, bodyreport )四、效果评估这套系统上线后我们对过去3个月的故障复盘效率做了对比指标人工复盘AI辅助复盘提升报告产出时间平均2.5h平均8min94%报告完整性65%92%42%根因定位准确率78%85%9%后续改善落地率40%72%80%有意思的是报告完整性提升了42%——因为AI不会遗漏告警数据中的细节信息而工程师在事后复盘时经常会忘记一些关键事件。结语大模型做根因分析报告的自动生成不是为了替代工程师的判断——它是把工程师从写报告这个低价值工作中解放出来让你把精力花在分析根因、制定方案这些真正创造价值的事情上。记住一个原则AI出初稿人工做审核。既利用AI的效率又保留人的判断力。本文作者侯万里万里侯云原生运维工程师专注于AI运维智能化和故障自愈体系建设