告别手动写复盘：大模型根因分析报告自动生成方法详解

告别手动写复盘：大模型根因分析报告自动生成方法详解

一、报告生成的三个核心挑战

在搭建自动报告系统之前，先直面三个难题：

1.1 挑战一：数据异构性

把这些异构数据整合成一个统一的结构化上下文，是首要挑战。

1.2 挑战二：信息过载 vs 信息不足

故障时间窗口内的数据量巨大，但LLM的上下文窗口有限。需要在"保留关键证据"和"控制token数量"之间找平衡。

1.3 挑战三：幻觉控制

LLM生成报告时，可能会"编造"根因或数据——这是运维场景最不能接受的。

二、完整的技术方案

2.1 数据提取层：多源数据采集与摘要

# data_extractor.py — 多源数据提取器
class MultiSourceDataExtractor:
"""从多个数据源提取故障相关的数据"""

def __init__(self):
self.sources = {
'prometheus': PrometheusExtractor(),
'elasticsearch': ElasticsearchExtractor(),
'jaeger': JaegerExtractor(),
'cmdb': CMDBExtractor(),
'git': GitExtractor()
}

def extract_summary(self, service: str, start_time: datetime,
end_time: datetime) -> dict:
"""提取摘要数据（控制token量）"""

# 每个数据源都返回摘要而非全量数据
results = {}

# Prometheus：只返回异常指标的摘要
metrics = self.sources['prometheus'].get_anomaly_metrics(
service, start_time, end_time
)
results['metrics'] = {
'total_metrics_checked': metrics['total'],
'anomaly_count': len(metrics['anomalies']),
'top_anomalies': metrics['anomalies'][:5], # 只取前5个
'summary': f"检查了{metrics['total']}个指标，发现{len(metrics['anomalies'])}个异常"
}

# Elasticsearch：只返回聚类后的错误模式
logs = self.sources['elasticsearch'].get_error_patterns(
service, start_time, end_time
)
results['logs'] = {
'total_errors': logs['total'],
'patterns': logs['patterns'][:10], # 只取前10种模式
'summary': f"发现{logs['total']}条错误日志，聚合为{len(logs['patterns'])}种模式"
}

# Jaeger：只返回慢Trace的摘要
traces = self.sources['jaeger'].get_slow_traces(
service, start_time, end_time
)
results['traces'] = {
'total_slow': traces['total'],
'top_spans': traces['top_spans'][:5],
'summary': f"检测到{traces['total']}条慢Trace，最慢的Span在{traces['slowest_service']}"
}

# CMDB：返回故障前2小时的变更
changes = self.sources['cmdb'].get_changes(
service, start_time – timedelta(hours=2), start_time
)
results['changes'] = {
'total': len(changes),
'changes': changes[:5],
'summary': f"故障前2小时内有{len(changes)}次变更"
}

return results

2.2 提示词工程层：结构化Prompt设计

# prompt_engineer.py — 结构化提示词工程
class ReportPromptBuilder:
"""构建结构化的报告生成提示词"""

def build(self, context: dict) -> list:
"""构建多轮对话的提示词"""

# System Prompt：角色定义与约束
system_prompt = """你是AI-SRE，一个严谨的运维故障分析助手。你的核心原则：
1. 只基于提供的数据做分析，不编造不存在的信息
2. 如果证据不足，明确标注"不确定"而非强行给出结论
3. 输出格式严格执行Markdown结构
4. 所有数据引用必须标注来源

分析框架：采用5Why分析法逐层深入
"""

# User Prompt：上下文数据与指令
user_prompt = f"""请基于以下故障数据生成根因分析报告。

## 故障信息
– 服务名称：{context['service_name']}
– 故障时间窗口：{context['time_window']}
– 告警级别：{context['severity']}

## 数据摘要
### 指标异常
{json.dumps(context['data']['metrics'], indent=2, ensure_ascii=False)}

### 错误日志
{json.dumps(context['data']['logs'], indent=2, ensure_ascii=False)}

### 链路追踪
{json.dumps(context['data']['traces'], indent=2, ensure_ascii=False)}

### 最近变更
{json.dumps(context['data']['changes'], indent=2, ensure_ascii=False)}

## 输出要求
请按以下结构生成报告，每节需要引用数据来源：

# 故障根因分析报告

## 一、故障概览
[基本信息：故障编号、时间、影响范围、严重级别]

## 二、影响评估
[受影响请求数、错误率变化、SLA影响]

## 三、故障时间线
| 时间 | 事件 | 数据来源 |
|——|——|———|

## 四、根因分析
### 4.1 现象描述
[故障在监控数据中的具体表现]
### 4.2 直接原因
[通过数据直接证实的触发因素]
### 4.3 根本原因
[通过5Why分析法追溯的底层原因]

## 五、修复措施
– 临时止血：[具体操作]
– 长期修复：[建议方案]

## 六、改进建议
### 6.1 可观测性改进
### 6.2 流程改进
### 6.3 架构改进

⚠️ **重要：对于没有数据支持的推断，请明确标注"推测"字样**
"""

return [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': user_prompt}
]

2.3 报告验证层：严谨性审查

# report_validator.py — 报告验证器
class ReportValidator:
"""验证报告的数据准确性和逻辑合理性"""

def validate(self, report: str, raw_data: dict) -> dict:
"""验证报告，返回验证结果"""
issues = []

# 1. 检查数字一致性
numbers_in_report = self._extract_numbers(report)
numbers_in_data = self._extract_numbers_from_data(raw_data)

for num in numbers_in_report:
if num['value'] not in numbers_in_data:
issues.append({
'type': 'data_inconsistency',
'detail': f"报告中的数值'{num['value']}'在原始数据中未找到",
'location': num['context']
})

# 2. 检查是否有"可能""大概"等模糊表述
vague_patterns = ['可能', '大概', '也许', 'perhaps', 'maybe', 'possibly']
for pattern in vague_patterns:
matches = re.finditer(pattern, report)
for match in matches:
issues.append({
'type': 'vague_statement',
'detail': f"使用了模糊表述'{pattern}'",
'location': report[max(0, match.start()-50):match.end()+50]
})

# 3. 检查是否所有数据来源都有标注
data_sources = ['Prometheus', 'Elasticsearch', 'Jaeger', 'CMDB', 'Git']
for source in data_sources:
if source not in report:
issues.append({
'type': 'missing_source',
'detail': f"报告中未引用{source}数据",
})

return {
'is_valid': len(issues) == 0,
'issues': issues,
'report_sections': self._extract_sections(report)
}

2.4 完整生成流水线

# report_pipeline.py — 完整的报告生成流水线
class ReportGenerationPipeline:
"""报告生成流水线"""

async def generate(self, alert_event: dict) -> dict:
"""完整的报告生成流程"""
pipeline_start = time.time()

try:
# Step 1: 数据提取（异步并发）
async with aiohttp.ClientSession() as session:
context = await self._extract_data(alert_event, session)

# Step 2: 构建提示词
prompt_messages = self.prompt_builder.build(context)

# Step 3: 调用LLM生成报告
report = await self._call_llm(prompt_messages)

# Step 4: 验证报告
validation = self.validator.validate(report, context['data'])

# Step 5: 如果有问题，修正
if not validation['is_valid']:
report = await self._refine_report(report, validation['issues'])

pipeline_duration = time.time() – pipeline_start

return {
'report': report,
'validation': validation,
'pipeline_stats': {
'duration_seconds': pipeline_duration,
'data_points_processed': self._count_data_points(context),
'model': self.model_name,
'tokens_used': self._estimate_tokens(prompt_messages, report)
}
}

except Exception as e:
return {
'error': str(e),
'partial_report': report if 'report' in locals() else None,
'pipeline_stats': {
'failed_at': time.time() – pipeline_start
}
}

三、实践经验与效果

3.1 调参心得

3.2 效果数据

四、总结

自动化根因分析报告生成的价值不只是"省了写报告的时间"。它让故障分析从"事后回忆"变成了"事中记录"——告警触发的同时，所有关键数据已经被系统自动捕获和关联，不会因为人的记忆偏差而遗漏细节。

我给团队定了一个原则：AI负责写初稿，人负责审核和深度分析。这个搭配让我们的复盘质量提升了42%，而复盘成本下降了80%。