1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162
| """ 座舱 AI Agent 安全决策框架 对比单模态 vs 多模态安全策略 """
from dataclasses import dataclass from enum import Enum from typing import Optional
class RiskLevel(Enum): SAFE = 0 LOW_RISK = 1 MEDIUM_RISK = 2 HIGH_RISK = 3 CRITICAL = 4
@dataclass class VehicleContext: speed_kmh: float time_of_day: str road_type: str weather: str
@dataclass class DriverContext: attention_level: float gaze_target: str voice_command: str voice_confidence: float emotion: str
class SafetyGatekeeper: """ AI Agent 安全门控:在执行前评估风险 核心原则:高速行驶 + 危险操作 = 拒绝 """ DANGEROUS_COMMANDS = [ 'turn_off_headlights', 'open_door', 'fold_mirror', 'disable_adas', 'change_drive_mode', ] def evaluate(self, driver: DriverContext, vehicle: VehicleContext) -> tuple: """ 评估操作风险 Returns: (RiskLevel, should_execute, reason) """ risk = RiskLevel.SAFE if vehicle.speed_kmh > 60: risk = RiskLevel(risk.value + 1) if vehicle.time_of_day == 'night': risk = RiskLevel(risk.value + 1) if vehicle.road_type == 'highway': risk = RiskLevel(risk.value + 1) if driver.attention_level < 0.5: risk = RiskLevel(risk.value + 1) if driver.voice_confidence < 0.8: risk = RiskLevel(risk.value + 1) is_dangerous = any(cmd in driver.voice_command for cmd in self.DANGEROUS_COMMANDS) if is_dangerous: risk = RiskLevel(risk.value + 2) if risk.value >= RiskLevel.HIGH_RISK.value: return (risk, False, "操作风险过高,已拒绝执行") elif risk.value >= RiskLevel.MEDIUM_RISK.value: return (risk, False, "请确认操作意图") else: return (risk, True, "操作安全,执行中") def safe_execute(self, driver: DriverContext, vehicle: VehicleContext) -> dict: """ 安全执行流程 """ risk, should_exec, reason = self.evaluate(driver, vehicle) if not should_exec: response = { 'action': 'reject', 'reason': reason, 'risk_level': risk.name, 'voice_feedback': self._generate_feedback(driver, vehicle, reason), 'visual_alert': True if risk.value >= 3 else False, 'haptic_feedback': True if risk.value >= 4 else False } return response return { 'action': 'execute', 'risk_level': risk.name, 'reason': reason } def _generate_feedback(self, driver, vehicle, reason): """生成自然语言反馈""" if 'headlight' in driver.voice_command and vehicle.time_of_day == 'night': return "夜间高速行驶中,关闭前大灯存在安全风险,操作已拒绝。如需调整灯光,请停车后操作。" return f"当前场景下操作存在风险:{reason}。请在安全停车后重试。"
if __name__ == "__main__": gatekeeper = SafetyGatekeeper() accident_driver = DriverContext( attention_level=0.7, gaze_target='road', voice_command='turn_off_headlights', voice_confidence=0.75, emotion='neutral' ) accident_vehicle = VehicleContext( speed_kmh=110, time_of_day='night', road_type='highway', weather='clear' ) result = gatekeeper.safe_execute(accident_driver, accident_vehicle) print("=== 事故场景(单模态会执行,多模态拒绝)===") print(f"风险等级: {result['risk_level']}") print(f"执行决策: {result['action']}") print(f"原因: {result['reason']}") if 'voice_feedback' in result: print(f"语音反馈: {result['voice_feedback']}") normal_driver = DriverContext( attention_level=0.9, gaze_target='road', voice_command='turn_off_headlights', voice_confidence=0.95, emotion='neutral' ) normal_vehicle = VehicleContext( speed_kmh=0, time_of_day='day', road_type='parking', weather='clear' ) result = gatekeeper.safe_execute(normal_driver, normal_vehicle) print("\n=== 正常场景(停车场白天)===") print(f"风险等级: {result['risk_level']}") print(f"执行决策: {result['action']}") print(f"原因: {result['reason']}")
|