服务机器人7×24小时运行，异常不可避免——关键是如何优雅降级和自动恢复：
1. 快速检测：看门狗+健康检查
2. 优雅降级：核心功能优先，非核心关闭
3. 自动恢复：分级恢复策略
4. 安全兜底：无法恢复时安全停车
5. 根因分析：记录现场，远程诊断class ExceptionHierarchy:
"""异常层级体系"""
def __init__(self):
self.exceptions = {
"NavigationError": {
"StuckError": {"severity": "critical", "recovery": "脱困程序"},
"LostError": {"severity": "critical", "recovery": "重新定位"},
"PathBlockedError": {"severity": "warning", "recovery": "重规划路径"},
"ElevatorTimeoutError": {"severity": "warning", "recovery": "重新呼叫"},
},
"HardwareError": {
"BatteryLowError": {"severity": "critical", "recovery": "返回充电"},
"MotorError": {"severity": "critical", "recovery": "停止运动+通知"},
"SensorError": {"severity": "warning", "recovery": "降级运行"},
"NetworkError": {"severity": "warning", "recovery": "切换离线模式"},
},
"TaskError": {
"GraspFailError": {"severity": "warning", "recovery": "重试抓取"},
"PersonNotFoundError": {"severity": "info", "recovery": "等待+通知"},
"DeliveryFailError": {"severity": "warning", "recovery": "退回+通知"},
},
"InteractionError": {
"ASRTimeoutError": {"severity": "info", "recovery": "提示重说"},
"NLUConfusionError": {"severity": "info", "recovery": "换种说法"},
"FaceNotFoundError": {"severity": "info", "recovery": "请求注册"},
},
}
def get_recovery(self, exception_name):
"""获取恢复策略"""
for category, exceptions in self.exceptions.items():
if exception_name in exceptions:
return exceptions[exception_name]
return {"severity": "unknown", "recovery": "记录并忽略"}
def list_all(self):
for cat, excs in self.exceptions.items():
print(f"\n🔹 {cat}:")
for name, info in excs.items():
icon = {"critical":"🔴","warning":"🟡","info":"ℹ️"}.get(info["severity"],"❓")
print(f" {icon} {name}: {info['severity']} → {info['recovery']}")
eh = ExceptionHierarchy()
print("异常层级体系")
print("=" * 55)
eh.list_all()
print("\n✅ 异常体系验证通过")class RecoveryManager:
"""异常恢复管理器"""
def __init__(self):
self.recovery_strategies = {
"StuckError": [
{"step": 1, "action": "尝试后退0.5m", "success_rate": 0.6},
{"step": 2, "action": "原地旋转360°", "success_rate": 0.3},
{"step": 3, "action": "重规划路径绕行", "success_rate": 0.8},
{"step": 4, "action": "通知远程协助", "success_rate": 0.95},
],
"BatteryLowError": [
{"step": 1, "action": "降低运行速度", "success_rate": 1.0},
{"step": 2, "action": "取消低优先级任务", "success_rate": 1.0},
{"step": 3, "action": "返回充电桩", "success_rate": 0.9},
],
"NetworkError": [
{"step": 1, "action": "重试连接(3次)", "success_rate": 0.7},
{"step": 2, "action": "切换到4G备用", "success_rate": 0.8},
{"step": 3, "action": "进入离线模式", "success_rate": 1.0},
],
}
self.recovery_history = []
def recover(self, error_type, context=None):
"""执行恢复流程"""
strategies = self.recovery_strategies.get(error_type, [])
if not strategies:
return {"success": False, "reason": f"无恢复策略: {error_type}"}
print(f"\n📋 恢复 {error_type}:")
for strategy in strategies:
print(f" 步骤{strategy['step']}: {strategy['action']} (成功率{strategy['success_rate']:.0%})")
self.recovery_history.append({"error": error_type, "step": strategy["step"]})
# 假设按概率成功
import random
if random.random() < strategy["success_rate"]:
print(f" ✅ 成功!")
return {"success": True, "step": strategy["step"], "action": strategy["action"]}
else:
print(f" ❌ 失败, 尝试下一步")
return {"success": False, "reason": "所有恢复策略均失败"}
def get_stats(self):
total = len(self.recovery_history)
by_error = {}
for entry in self.recovery_history:
by_error.setdefault(entry["error"], 0)
by_error[entry["error"]] += 1
return {"total_steps": total, "by_error": by_error}
import random
random.seed(42)
rm = RecoveryManager()
print("异常恢复管理器")
print("=" * 55)
errors = ["StuckError", "BatteryLowError", "NetworkError"]
for err in errors:
result = rm.recover(err)
stats = rm.get_stats()
print(f"\n恢复统计: 总步骤{stats['total_steps']}")
print("✅ 恢复管理验证通过")class WatchdogTimer:
"""看门狗定时器"""
def __init__(self):
self.timers = {}
self.timeouts = {
"navigation": 120, # 导航超时2分钟
"elevator": 90, # 电梯交互超时90秒
"task": 600, # 任务超时10分钟
"heartbeat": 30, # 心跳超时30秒
"sensor": 5, # 传感器数据超时5秒
}
self.critical_processes = ["navigation", "sensor", "heartbeat"]
self.log = []
def register(self, process, timeout=None):
t = timeout or self.timeouts.get(process, 60)
self.timers[process] = {"timeout": t, "last_kick": 0, "expired": False}
def kick(self, process, current_time):
"""喂狗"""
if process in self.timers:
self.timers[process]["last_kick"] = current_time
self.timers[process]["expired"] = False
def check(self, current_time):
"""检查所有看门狗"""
expired = []
for process, timer in self.timers.items():
elapsed = current_time - timer["last_kick"]
if elapsed > timer["timeout"] and not timer["expired"]:
timer["expired"] = True
severity = "CRITICAL" if process in self.critical_processes else "WARNING"
expired.append({"process": process, "elapsed": elapsed, "timeout": timer["timeout"], "severity": severity})
self.log.append(f"[{severity}] {process} 看门狗超时 ({elapsed}s > {timer['timeout']}s)")
return expired
wd = WatchdogTimer()
wd.register("navigation", 120)
wd.register("elevator", 90)
wd.register("heartbeat", 30)
wd.register("sensor", 5)
print("看门狗定时器")
print("=" * 55)
# 模拟运行
for t in range(150):
wd.kick("sensor", t) # 传感器正常
wd.kick("heartbeat", t) # 心跳正常
if t < 50:
wd.kick("navigation", t) # 导航前50秒正常
# elevator一直没kick
if t % 30 == 0:
expired = wd.check(t)
if expired:
for e in expired:
icon = "🔴" if e["severity"] == "CRITICAL" else "🟡"
print(f" {icon} t={t}: {e['process']} 超时 ({e['elapsed']}s > {e['timeout']}s)")
print(f"\n日志: {len(wd.log)}条")
print("✅ 看门狗验证通过")| 模式 | 描述 | 应用 |
|---|---|---|
| 断路器 | 连续失败后断开 | 外部服务调用 |
| 重试+退避 | 指数退避重试 | 网络请求 |
| 降级模式 | 核心功能优先 | 资源不足时 |
| 冗余备份 | 主备切换 | 关键传感器 |
| 看门狗 | 超时检测+重启 | 进程监控 |
实现断路器模式：连续3次调用失败后断开，30秒后半开尝试，成功则恢复。
设计安全停车流程：所有恢复策略失败后，机器人安全停车→锁轮→通知远程→等待人工干预。
实现异常预测：基于历史模式，预测即将发生的异常（如电量趋势预测低电量），提前预防。