告别手动写复盘：大模型根因分析报告自动生成方法详解-平芜编程栈

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

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

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

1.1 挑战一：数据异构性

数据来源	格式	访问方式	数据量
Prometheus	时序数值	HTTP API	时间序列×时间点
Elasticsearch	JSON文档	REST API	按条件过滤
Jaeger	Span树	gRPC API	按TraceID查询
CMDB	结构化记录	SQL	有限条数
Git	提交记录	Git Log	按时间范围

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

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 调参心得

参数	推荐值	原因
temperature	0.1	降低创造力，提高确定性
top_p	0.9	保留一些多样性但不过度
max_tokens	4000	够覆盖完整报告
presence_penalty	0	不需要避免重复，专业术语需要重复
frequency_penalty	0.2	轻微惩罚无意义的重复