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 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209
| import torch import torch.nn as nn import numpy as np from dataclasses import dataclass from enum import Enum
class ConversationStrategy(Enum): """对话策略类型""" SMALL_TALK = "闲聊" HAZARD_ALERT = "危险提醒" TRIVIA = "知识问答" MEMORY = "回忆对话" OPEN_ENDED = "开放话题"
@dataclass class DriverState: """驾驶员状态""" fatigue_level: float alertness: float microsleep: bool takeover_readiness: float eyes_on_road: bool conversation_engagement: float
class LLMConversationAgent: """ LLM对话代理:根据疲劳状态动态调整对话策略 架构: 1. 疲劳状态监测(DMS输入) 2. 对话策略选择(基于状态) 3. LLM生成对话(上下文相关) 4. 效果评估(警觉性反馈) """ def __init__(self): self.strategy_history = [] self.fatigue_threshold = 0.4 self.microsleep_threshold = 0.6 self.llm_config = { 'model': '车载7B参数LLM', 'max_tokens': 50, 'temperature': 0.7, 'voice_output': True, } def select_strategy(self, state: DriverState) -> ConversationStrategy: """ 根据驾驶员状态选择对话策略 规则: - 轻度疲劳:闲聊(轻松提升) - 中度疲劳:知识问答(需要思考) - 重度疲劳:危险提醒(紧急唤醒) - 微睡眠:强刺激唤醒 """ if state.microsleep: return ConversationStrategy.HAZARD_ALERT elif state.fatigue_level > self.microsleep_threshold: return ConversationStrategy.TRIVIA elif state.fatigue_level > self.fatigue_threshold: return ConversationStrategy.SMALL_TALK else: return None def generate_response(self, strategy: ConversationStrategy, state: DriverState, conversation_context: list) -> str: """ 生成对话响应 简化版:实际用LLM生成 """ responses = { ConversationStrategy.SMALL_TALK: [ "今天天气不错,注意到外面的山了吗?", "你平时周末喜欢做什么?", "这段路风景不错,经常走吗?", ], ConversationStrategy.TRIVIA: [ "你知道吗?蜂鸟是唯一能倒退飞行的鸟。", "快速算一下:17乘以3等于多少?", "说出三种红色的水果。", ], ConversationStrategy.HAZARD_ALERT: [ "前方有施工路段,请注意!", "你刚才闭眼超过2秒,需要休息吗?", "前方500米有交叉路口,请确认路况!", ], ConversationStrategy.MEMORY: [ "还记得上次走这段路的情况吗?", "回忆一下你今天的行程安排。", ], } pool = responses.get(strategy, ["你好,感觉怎么样?"]) return np.random.choice(pool) def assess_effect(self, pre_state: DriverState, post_state: DriverState) -> dict: """评估对话效果""" return { 'alertness_change': post_state.alertness - pre_state.alertness, 'fatigue_change': post_state.fatigue_level - pre_state.fatigue_level, 'engagement': post_state.conversation_engagement, 'microsleep_resolved': pre_state.microsleep and not post_state.microsleep, }
class FatigueInterventionSystem: """ 完整疲劳干预系统 架构: 1. DMS检测疲劳 2. LLM对话代理干预 3. 效果评估+策略调整 """ def __init__(self): self.agent = LLMConversationAgent() self.dms_state = DriverState( fatigue_level=0.0, alertness=1.0, microsleep=False, takeover_readiness=1.0, eyes_on_road=True, conversation_engagement=0.0, ) def update_dms(self, fatigue: float, alertness: float, microsleep: bool, eyes_on_road: bool): """更新DMS状态""" self.dms_state.fatigue_level = fatigue self.dms_state.alertness = alertness self.dms_state.microsleep = microsleep self.dms_state.eyes_on_road = eyes_on_road def intervene(self) -> dict: """执行干预""" strategy = self.agent.select_strategy(self.dms_state) if strategy is None: return {'action': 'none', 'reason': '正常状态'} pre_state = DriverState(**self.dms_state.__dict__) response = self.agent.generate_response( strategy, self.dms_state, [] ) self.dms_state.alertness = min(1.0, self.dms_state.alertness + 0.15) self.dms_state.fatigue_level = max(0, self.dms_state.fatigue_level - 0.10) if self.dms_state.microsleep: self.dms_state.microsleep = False self.dms_state.alertness = 0.7 self.dms_state.conversation_engagement = 0.6 effect = self.agent.assess_effect(pre_state, self.dms_state) return { 'action': 'conversation', 'strategy': strategy.value, 'response': response, 'effect': effect, }
if __name__ == "__main__": system = FatigueInterventionSystem() print("=== 30分钟L3驾驶疲劳干预模拟 ===\n") for minute in range(30): system.update_dms( fatigue=min(1.0, 0.02 * minute + np.random.randn() * 0.03), alertness=max(0.1, 1.0 - 0.025 * minute), microsleep=(minute > 20 and np.random.rand() < 0.15), eyes_on_road=(minute <= 15 or np.random.rand() > 0.2) ) result = system.intervene() if result['action'] != 'none' and minute % 5 == 0: print(f"[{minute:2d}min] 疲劳={system.dms_state.fatigue_level:.2f} " f"警觉={system.dms_state.alertness:.2f} " f"微睡眠={'是' if system.dms_state.microsleep else '否'}") print(f" → 策略: {result.get('strategy', '无')}") print(f" → 对话: {result.get('response', '无')}") if 'effect' in result: e = result['effect'] print(f" → 效果: 警觉+{e['alertness_change']:.2f} " f"疲劳{e['fatigue_change']:.2f}") print()
|