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
| """ When2Talk: 车内代理沟通时机决策系统
论文核心方法: 情境敏感 (CS) 策略
决策维度: 1. 事件后果 (Event Consequence) 2. 乘客活动 (Passenger Activity) 3. 信息持续价值 (Continuing Information Value) 4. 确认需求 (Confirmation Need) """
import numpy as np from dataclasses import dataclass from typing import Optional, List from enum import Enum
class TalkAction(Enum): IMMEDIATE = "立即沟通" DELAYED = "延迟沟通" SILENT = "保持沉默"
@dataclass class CabinEvent: """座舱事件""" event_type: str consequence: float passenger_busy: float info_value_decay: float needs_confirmation: bool @dataclass class TalkDecision: """沟通决策""" action: TalkAction delay_seconds: float reason: str confidence: float
class When2TalkAgent: """ 车内代理沟通时机决策器 论文 CS 策略实现 """ EVENT_WEIGHTS = { 'emergency_brake': {'consequence': 0.95, 'decay': 0.9, 'confirm': True}, 'lane_change': {'consequence': 0.6, 'decay': 0.5, 'confirm': False}, 'traffic_light': {'consequence': 0.4, 'decay': 0.3, 'confirm': False}, 'scenic_point': {'consequence': 0.1, 'decay': 0.1, 'confirm': False}, 'speed_limit': {'consequence': 0.3, 'decay': 0.4, 'confirm': False}, 'construction': {'consequence': 0.5, 'decay': 0.6, 'confirm': False}, 'accident_ahead': {'consequence': 0.8, 'decay': 0.7, 'confirm': True}, } def __init__(self, dispositional_trust: float = 0.5): """ Args: dispositional_trust: 用户基线信任度 (0-1) 高信任 → 偏好减少沟通 低信任 → 偏好即时沟通 """ self.trust = dispositional_trust def decide(self, event: CabinEvent, passenger_activity: float) -> TalkDecision: """ 决策: Immediate / Delayed / Silent Args: event: 座舱事件 passenger_activity: 乘客当前活动占用度 0-1 0 = 空闲/睡觉 0.5 = 轻度活动 (听音乐) 1.0 = 深度活动 (电话/阅读) Returns: decision: TalkDecision """ consequence_score = event.consequence conflict_score = passenger_activity * (1 - consequence_score) urgency = event.info_value_decay * (1 - self.trust) confirm_score = 1.0 if event.needs_confirmation else 0.0 talk_score = (consequence_score * 0.4 + urgency * 0.3 + confirm_score * 0.2 + (1 - conflict_score) * 0.1) silence_score = conflict_score * 0.5 + self.trust * 0.3 + (1 - event.consequence) * 0.2 if talk_score > 0.7: action = TalkAction.IMMEDIATE delay = 0 reason = f"高后果({event.consequence:.1f}) + 紧迫({urgency:.1f})" elif talk_score > 0.4: action = TalkAction.DELAYED delay = 5.0 * (1 - talk_score) reason = f"中等重要性, 等待乘客活动结束" else: action = TalkAction.SILENT delay = float('inf') reason = f"低后果 + 乘客忙碌({passenger_activity:.1f})" confidence = abs(talk_score - silence_score) / (talk_score + silence_score + 1e-8) return TalkDecision( action=action, delay_seconds=delay, reason=reason, confidence=confidence ) def batch_decide(self, events: List[CabinEvent], activities: List[float]) -> List[TalkDecision]: """批量决策""" return [self.decide(e, a) for e, a in zip(events, activities)]
if __name__ == "__main__": agent = When2TalkAgent(dispositional_trust=0.5) scenarios = [ ("紧急制动", CabinEvent("emergency_brake", 0.95, 0.8, 0.9, True), 0.3), ("变道", CabinEvent("lane_change", 0.6, 0.4, 0.5, False), 0.8), ("风景点", CabinEvent("scenic_point", 0.1, 0.2, 0.1, False), 0.5), ("前方事故", CabinEvent("accident_ahead", 0.8, 0.6, 0.7, True), 0.7), ("限速变化", CabinEvent("speed_limit", 0.3, 0.5, 0.4, False), 0.6), ] print("=== When2Talk 决策测试 ===") print(f"{'场景':<12} {'后果':<8} {'乘客忙':<8} {'行动':<10} {'延迟':<8} {'原因'}") for name, event, activity in scenarios: decision = agent.decide(event, activity) print(f"{name:<12} {event.consequence:<8.1f} {activity:<8.1f} " f"{decision.action.value:<10} {decision.delay_seconds:<8.1f} {decision.reason}") print("\n=== 信任度对沟通频率的影响 ===") for trust in [0.2, 0.5, 0.8]: agent_t = When2TalkAgent(dispositional_trust=trust) events = [CabinEvent("lane_change", 0.6, 0.4, 0.5, False) for _ in range(10)] activities = [0.5] * 10 decisions = agent_t.batch_decide(events, activities) immediate = sum(1 for d in decisions if d.action == TalkAction.IMMEDIATE) delayed = sum(1 for d in decisions if d.action == TalkAction.DELAYED) silent = sum(1 for d in decisions if d.action == TalkAction.SILENT) print(f"信任度 {trust}: 立即={immediate}, 延迟={delayed}, 沉默={silent}") print("\n=== 论文核心发现 ===") print(f"{'策略':<15} {'沟通频率':<12} {'中断感':<12} {'信任度':<12}") print(f"{'事件触发 (ET)':<15} {'100%':<12} {'高':<12} {'= CS':<12}") print(f"{'情境敏感 (CS)':<15} {'~40%':<12} {'低':<12} {'= ET':<12}") print(f"\n→ CS 减少沟通 60%, 不降低信任") print(f"→ 四维决策: 后果/活动/价值/确认")
|