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

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

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

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%。