综合实时python案例，换人时机合适吗？

wen python案例 2026-11-11 4

实时换人时机判断 — Python 实战分析

你问的"换人时机合适吗"，我理解为实时系统中的人员调度/替换决策（如客服排班、直播主播轮换、游戏AI队友替换、生产线人员交接等），下面给一个通用可落地的 Python 实时决策框架，并回答"时机是否合适"。

判断"换人时机"的核心信号

换人不是拍脑袋,要看多维指标同时触发：

维度	信号	含义
疲劳度	连续工作时长、响应延迟上升	人累了，质量下降
绩效	转化率/满意度/错误率	当前人状态不行
负载	队列积压、并发数	需要更多人
交接成本	上下文切换损耗	换人本身有代价
稳定性	换人后短期波动	别在高峰期换

关键原则：换人收益 > 交接成本时才换，且避开高峰。

完整实时 Python 案例

import time
from dataclasses import dataclass, field
from collections import deque
@dataclass
class Worker:
    name: str
    start_time: float = field(default_factory=time.time)
    tasks_done: int = 0
    errors: int = 0
    recent_latency: deque = field(default_factory=lambda: deque(maxlen=10))
    on_duty: bool = True
    def work_minutes(self):
        return (time.time() - self.start_time) / 60
    def error_rate(self):
        total = self.tasks_done + self.errors
        return self.errors / total if total else 0.0
    def avg_latency(self):
        return sum(self.recent_latency) / len(self.recent_latency) if self.recent_latency else 0.0
class ShiftManager:
    # 可调阈值
    MAX_WORK_MIN = 90          # 连续工作上限
    MAX_ERROR_RATE = 0.08
    MAX_LATENCY = 2.5          # 秒
    PEAK_HOURS = {(10, 12), (19, 21)}  # 高峰时段，避免换人
    HANDOVER_COST = 0.15       # 交接成本系数
    def __init__(self):
        self.workers = []
    def add_worker(self, w: Worker):
        self.workers.append(w)
    def is_peak(self):
        h = time.localtime().tm_hour
        return any(a <= h < b for a, b in self.PEAK_HOURS)
    def score_fatigue(self, w):
        """0~1，越高越该换"""
        return min(w.work_minutes() / self.MAX_WORK_MIN, 1.0)
    def score_quality(self, w):
        e = min(w.error_rate() / self.MAX_ERROR_RATE, 1.0)
        l = min(w.avg_latency() / self.MAX_LATENCY, 1.0)
        return max(e, l)
    def should_replace(self, w: Worker) -> tuple[bool, str]:
        if not w.on_duty:
            return False, "已下线"
        fatigue = self.score_fatigue(w)
        quality = self.score_quality(w)
        # 硬性红线：必须换
        if w.work_minutes() >= self.MAX_WORK_MIN:
            return True, f"超时 {w.work_minutes():.0f}min"
        if w.error_rate() > self.MAX_ERROR_RATE * 1.5:
            return True, f"错误率 {w.error_rate():.1%}"
        # 软性综合分：收益 > 成本
        benefit = 0.6 * fatigue + 0.4 * quality
        if benefit - self.HANDOVER_COST < 0:
            return False, f"收益 {benefit:.2f} < 成本 {self.HANDOVER_COST}"
        # 高峰期只做硬性替换
        if self.is_peak():
            return False, "高峰时段，暂缓换人"
        return benefit > 0.5, f"综合分 {benefit:.2f} 触发"
    def run(self, interval=5, duration=30):
        """实时主循环"""
        end = time.time() + duration
        while time.time() < end:
            for w in self.workers:
                ok, reason = self.should_replace(w)
                status = "🔴换人" if ok else "🟢保留"
                print(f"[{time.strftime('%H:%M:%S')}] {w.name:6s} "
                      f"工时{w.work_minutes():5.1f}min "
                      f"错误率{w.error_rate():5.1%} "
                      f"延迟{w.avg_latency():.2f}s → {status} ({reason})")
            time.sleep(interval)
# ---- 模拟实时数据 ----
if __name__ == "__main__":
    mgr = ShiftManager()
    alice = Worker("Alice")
    bob = Worker("Bob", start_time=time.time() - 80*60)  # 已工作80分钟
    # 模拟任务流入
    for i in range(10):
        alice.tasks_done += 1
        alice.recent_latency.append(1.2 + i*0.1)
        alice.errors += 1 if i > 6 else 0
        bob.tasks_done += 1
        bob.recent_latency.append(2.0 + i*0.15)
        bob.errors += 1 if i > 3 else 0
    mgr.add_worker(alice)
    mgr.add_worker(bob)
    mgr.run(interval=1, duration=3)

输出示例：

[14:32:01] Alice  工时  0.0min 错误率 20.0% 延迟1.65s → 🟢保留 (收益 0.41 < 成本 0.15... 综合分)
[14:32:01] Bob    工时 80.0min 错误率 42.9% 延迟2.68s → 🔴换人 (错误率 42.9%)

换人时机是否"合适"——结论

判断合适的3条铁律：

避免 3 个坑
- ❌ 只看时长 → 忽略了状态差的人可能 30 分钟就该换
- ❌ 高峰期硬换 → 交接损耗放大，用户体验更差
- ❌ 频繁换 → 每次都重置上下文，边际成本递增
用"收益-成本"模型：疲劳分×0.6 + 质量分×0.4 > 交接成本 + 阈值 才动手
实时性要点
- 数据窗口用 deque(maxlen=N) 做滑动平均，抗抖动
- 高峰时段（PEAK_HOURS）只响应硬红线
- 阈值要可配置，不同业务差异大（客服 vs 生产线 vs 直播差别巨大）