关键词自动化报告 / 数据回归 / 基线比对 / 流水线守护## 背景自动化报告最怕的不是报错是静默错一条跑批流水线周而复始地生成周报、月报、BI 看板。它**不会报错**——但某天你发现这个月的智能数据中心工单数凭空少了 30 条。追查半天才知道上周改了部门归属的一个正则把一批工单从智能数据中心错划到了未匹配。这类问题的特征是**逻辑能跑、结果能出、但错了没人知道**。人工复核又跟不上自动化节奏。解法借鉴软件工程里的快照测试 / 基线比对把上一次**正确**的输出冻结成基线每次运行都和上一次 diff发现漂移就**亮红灯、拦住交付**而不是默默带着错误往下走。## 核心机制三件事构成一个闭环1. **freeze冻结基线**把当前正确的结果 CSV 复制成 baseline.csv。2. **compare对比**新结果和基线逐条 diff产出变更/新增/移除清单 部门分布变化 低置信度告警。3. **gate闸门**用**退出码**驱动上层流水线——无漂移继续有漂移 halt 等人工确认确认后再 freeze 推进。pythonimport csv, os, jsonfrom collections import CounterHERE os.path.dirname(os.path.abspath(__file__))def _read(path):读取 工单号 - {dept, strategy, handler} 的映射。d {}with open(path, r, encodingutf-8-sig, newline) as f:for r in csv.DictReader(f):oid (r.get(工单号) or ).strip()if oid:d[oid] {dept: r.get(部门, ),strategy: r.get(strategy, ),handler: r.get(处理人, )}return d, (strategy in (next(iter(d.values()), {})))def compare(baseline_path, current_path):base, _ _read(baseline_path)curr, has_strat _read(current_path)base_ids, curr_ids set(base), set(curr)# 跨周期护栏上周/本周工单完全不同直接 diff 会假阳性爆炸inter len(base_ids curr_ids)is_cross (inter / max(len(base_ids), len(curr_ids), 1)) 0.2changed, added, removed [], [], []if not is_cross:for oid in sorted(base_ids curr_ids):if base[oid][dept] ! curr[oid][dept]:changed.append((oid, base[oid][dept], curr[oid][dept]))added sorted(curr_ids - base_ids)removed sorted(base_ids - curr_ids)base_dist Counter(v[dept] for v in base.values())curr_dist Counter(v[dept] for v in curr.values())dist_lines [f{d}: {base_dist[d]} - {curr_dist[d]}for d in sorted(set(base_dist) | set(curr_dist))if base_dist[d] ! curr_dist[d]]return {is_cross: is_cross, changed: changed,added: added, removed: removed,dist_lines: dist_lines, has_strat: has_strat}def report(res):if res[is_cross]:print(ℹ️ 新周期首跑跳过工单级 diff基线由调用方自动推进。)return 0total len(res[changed]) len(res[added]) len(res[removed])for oid, old, new in res[changed]:print(f[变更] {oid}: {old} - {new})for line in res[dist_lines]:print(f[分布] {line})if total 0:print(✅ 无变化通过。)return 0print(f⚠️ 发现 {total} 处变化请人工复核后再交付。)return 2 # 2 有漂移haltdef main():base_p os.path.join(HERE, baseline.csv)cur_p os.path.join(HERE, current.csv)import sysif --freeze in sys.argv:import shutilshutil.copy(cur_p, base_p)print([OK] 基线已冻结。)return 0return report(compare(base_p, cur_p))if __name__ __main__:raise SystemExit(main())## 两个踩过的坑都是精华### 坑 1跨周期假阳性第一个版本直接 diff结果每周一跑都**炸出几千行变更**——因为上周的工单和本周的根本不是同一批。代码把上周全体当移除、本周全体当新增baseline 永远对齐不上流水线永远 halt。解法用**交集比例**判断是不是同一批数据。inter / max(n_base, n_curr) 0.2 就判定为新周期首跑跳过工单级 diff只让调用方把基线推进到本周。这一行救了整个自动化。### 坑 2用退出码当控制流别把有无漂移藏在 print 里。把它变成**进程退出码**让上层 _launcher.py 用 if guard_res 2: halt 直接分叉pythonguard_res run([python, regression_guard.py])if guard_res 0: # 无漂移freeze_baseline() # 自动推进基线elif guard_res 2: # 有漂移if input(确认无误并冻结? (y/N): ).lower() y:freeze_baseline() # 人工确认后才推进else:sys.exit(2) # 中止不生成报告**无漂移 → 自动推进有漂移 → 人工确认后才推进** 这个不对称设计很重要它既不让人工天天点确认又不让错误悄悄固化成新基线。## 适用场景任何周期性自动产出、错了难发现的东西都该装这道闸- 周报 / 月报 / 季报纸面数字- 数据看板的核心 KPI- 自动同步的台账、对账文件- 甚至模型推理的批量输出把输出当成一份**契约**这次和上次不一致必须有人签字。## 小结- 自动化报告最大的风险是**静默错**不是崩。- 用 **freeze → compare → gate** 三件套把上次的自己变成守门员。- **跨周期要用交集比例护栏**否则首跑即假阳性爆炸。- **退出码当控制流**无漂移自动推进、有漂移人工确认。