AI 驱动的 SRE 值班排班优化：从轮值到智能调度
一、值班排班的经验主义困境：疲劳积累与技能错配
SRE 团队的值班排班通常采用轮值制——每人值班一周，依次轮换。但轮值制忽略了两个关键因素：一是疲劳积累，连续处理深夜告警的工程师，第二天的判断力显著下降；二是技能错配，数据库故障需要 DBA 技能，网络故障需要网络工程师，但轮值到谁就是谁，技能不匹配导致排障时间延长。
AI 驱动的排班优化，核心思路是：根据告警类型的历史分布、工程师的技能画像和疲劳状态，动态生成最优排班方案，最小化"技能错配率"和"疲劳风险"。
二、排班优化的架构设计与约束模型
排班优化是一个多约束优化问题：硬约束包括值班连续性（同一人不能连续值班超过 2 周）、时区覆盖（7×24 小时覆盖）和法定休息日；软约束包括技能匹配度最大化、疲劳风险最小化和偏好尊重。
flowchart TB A[排班输入] --> B[工程师技能画像] A --> C[告警历史分布] A --> D[约束条件] B --> E[技能匹配度评分] C --> F[告警类型预测] D --> G[硬约束: 连续性/覆盖/休息] D --> H[软约束: 疲劳/偏好/公平] E --> I[排班优化引擎] F --> I G --> I H --> I I --> J[最优排班方案] J --> K[技能匹配率: > 80%] J --> L[疲劳风险: < 20%] J --> M[公平性: 方差 < 10%]三、生产级实现：排班优化引擎
# schedule_optimizer.py — AI 驱动的 SRE 排班优化引擎 from dataclasses import dataclass, field from typing import List, Dict, Optional, Set from enum import Enum from datetime import date, timedelta import random class SkillType(Enum): DATABASE = "database" NETWORK = "network" KUBERNETES = "kubernetes" APPLICATION = "application" SECURITY = "security" @dataclass class Engineer: id: str name: str skills: Dict[SkillType, float] # 技能熟练度 0-1 timezone: str last_oncall_end: Optional[date] = None # 上次值班结束日期 oncall_count_this_month: int = 0 preference: Dict[str, bool] = field(default_factory=dict) # 偏好 @dataclass class OncallSlot: date: date shift: str # "day" / "night" required_skills: List[SkillType] # 该时段最可能需要的技能 @dataclass class OncallAssignment: slot: OncallSlot engineer: Engineer skill_match_score: float fatigue_risk: float class ScheduleOptimizer: """排班优化引擎：多约束优化排班方案""" def optimize( self, engineers: List[Engineer], slots: List[OncallSlot], constraints: Dict, ) -> List[OncallAssignment]: """生成最优排班方案""" assignments = [] assigned_dates: Dict[str, Set[date]] = {e.id: set() for e in engineers} for slot in slots: # 为每个时段选择最优工程师 best_engineer = self._select_best_engineer( slot, engineers, assigned_dates, constraints ) if best_engineer: skill_score = self._calculate_skill_match( best_engineer, slot ) fatigue_risk = self._calculate_fatigue_risk( best_engineer, assigned_dates[best_engineer.id] ) assignments.append(OncallAssignment( slot=slot, engineer=best_engineer, skill_match_score=skill_score, fatigue_risk=fatigue_risk, )) assigned_dates[best_engineer.id].add(slot.date) return assignments def _select_best_engineer( self, slot: OncallSlot, engineers: List[Engineer], assigned_dates: Dict[str, Set[date]], constraints: Dict, ) -> Optional[Engineer]: """为指定时段选择最优工程师""" candidates = [] for eng in engineers: score = 0.0 # 因子 1：技能匹配度（权重 40%） skill_score = self._calculate_skill_match(eng, slot) score += skill_score * 40 # 因子 2：疲劳风险（权重 30%） fatigue = self._calculate_fatigue_risk( eng, assigned_dates[eng.id] ) score += (1 - fatigue) * 30 # 疲劳越低，分数越高 # 因子 3：公平性（权重 20%） max_oncall = max(e.oncall_count_this_month for e in engineers) fairness = 1 - (eng.oncall_count_this_month / max(max_oncall, 1)) score += fairness * 20 # 因子 4：偏好（权重 10%） if eng.preference.get(f"no_{slot.shift}", False): score -= 50 # 强烈惩罚违反偏好 if eng.preference.get(f"prefer_{slot.shift}", False): score += 10 # 硬约束检查 if not self._check_hard_constraints(eng, slot, assigned_dates, constraints): continue candidates.append((eng, score)) if not candidates: return None # 选择得分最高的工程师 candidates.sort(key=lambda x: x[1], reverse=True) return candidates[0][0] def _calculate_skill_match( self, engineer: Engineer, slot: OncallSlot ) -> float: """计算技能匹配度""" if not slot.required_skills: return 0.5 total = 0.0 for skill in slot.required_skills: total += engineer.skills.get(skill, 0.0) return total / len(slot.required_skills) def _calculate_fatigue_risk( self, engineer: Engineer, assigned_dates: Set[date] ) -> float: """计算疲劳风险：连续值班天数越多，风险越高""" if not assigned_dates: return 0.0 # 计算最近 7 天的值班天数 recent = sum(1 for d in assigned_dates if (date.today() - d).days <= 7) return min(recent / 5.0, 1.0) # 5 天以上为最高风险 def _check_hard_constraints( self, engineer: Engineer, slot: OncallSlot, assigned_dates: Dict[str, Set[date]], constraints: Dict, ) -> bool: """检查硬约束""" # 约束 1：连续值班不超过 max_consecutive_days max_consecutive = constraints.get("max_consecutive_days", 5) consecutive = self._count_consecutive_days( assigned_dates[engineer.id], slot.date ) if consecutive >= max_consecutive: return False # 约束 2：上次值班结束距今天至少 min_rest_days 天 if engineer.last_oncall_end: rest_days = (slot.date - engineer.last_oncall_end).days if rest_days < constraints.get("min_rest_days", 2): return False # 约束 3：本月值班次数不超过 max_per_month max_per_month = constraints.get("max_per_month", 8) if engineer.oncall_count_this_month >= max_per_month: return False return True def _count_consecutive_days( self, assigned: Set[date], target: date ) -> int: """计算到目标日期为止的连续值班天数""" count = 0 check = target - timedelta(days=1) while check in assigned: count += 1 check -= timedelta(days=1) return count四、边界分析与架构权衡
AI 排班优化在生产落地中需要正视以下 Trade-off：
技能匹配的精度。工程师的技能画像需要定期更新，技能熟练度的评估存在主观性。建议从故障工单的"处理人"和"解决时间"数据中自动推断技能熟练度——处理某类故障越快，该技能的熟练度越高。
排班的灵活性。优化算法生成的排班方案可能无法覆盖突发情况（如工程师生病、紧急请假）。必须保留人工调整的入口，且调整后自动重新优化后续排班。
公平性的定义。不同工程师对"公平"的理解不同——有人认为值班次数相等就是公平，有人认为按技能分配更公平。建议将公平性定义为"值班负担的方差最小化"，综合考虑值班次数、夜班次数和疲劳程度。
适用边界：排班优化最适合 5 人以上的 SRE 团队。3-4 人的小团队排班空间有限，优化收益不大。
五、总结
AI 驱动的 SRE 排班优化，将值班安排从"简单轮值"推进到"智能调度"。核心模型：技能匹配度 × 疲劳风险 × 公平性 × 偏好的多因子评分，硬约束保证合规，软约束优化质量。落地建议：第一，从故障工单数据自动推断技能画像；第二，保留人工调整入口应对突发情况；第三，定期收集工程师反馈，优化评分权重。关键原则：排班优化的目标不是"完美排班"，而是"减少技能错配和疲劳积累"——每一次技能匹配的排班，都可能将故障恢复时间缩短 30%。
