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
| """
论文：FedQoS: QoS-Aware Federated Learning for Multimodal In-Cabin Interaction
arXiv: 2609.20123
核心方法：异步事件触发联邦学习 + 两阶段门控
"""
import torch
import torch.nn as nn
import numpy as np
from typing import Optional, Dict, Tuple
class FedQoSGate:
"""FedQoS两阶段门控"""
def __init__(self,
buffer_threshold: float = 0.3,
energy_threshold: float = 0.2,
qos_latency_max: float = 100.0,
qos_energy_max: float = 0.5,
novelty_threshold: float = 0.05):
self.buffer_threshold = buffer_threshold
self.energy_threshold = energy_threshold
self.qos_latency_max = qos_latency_max
self.qos_energy_max = qos_energy_max
self.novelty_threshold = novelty_threshold
def training_gate(self, buffer_level: float,
energy_level: float) -> Tuple[bool, str]:
"""
阶段1: 资源感知训练门控
仅当传感缓冲区和能量储备满足安全阈值时才初始化本地训练
"""
if buffer_level >= self.buffer_threshold and energy_level >= self.energy_threshold:
return True, "TRAIN: 资源充足"
elif buffer_level < self.buffer_threshold:
return False, f"SKIP: 缓冲区不足 ({buffer_level:.2f} < {self.buffer_threshold})"
else:
return False, f"SKIP: 能量不足 ({energy_level:.2f} < {self.energy_threshold})"
def transmission_gate(self,
model_update: Dict[str, torch.Tensor],
global_params: Dict[str, torch.Tensor],
latency: float,
energy_cost: float) -> Tuple[bool, float, str]:
"""
阶段2: QoS感知传输门控
基于模型新颖性+延迟+能量的效率分数决定是否上传
"""
novelty = 0.0
count = 0
for key in model_update:
if key in global_params:
diff = (model_update[key] - global_params[key]).norm().item()
norm = global_params[key].norm().item() + 1e-8
novelty += diff / norm
count += 1
novelty = novelty / max(count, 1)
latency_penalty = min(latency / self.qos_latency_max, 1.0)
energy_penalty = min(energy_cost / self.qos_energy_max, 1.0)
efficiency_score = novelty / (1.0 + latency_penalty + energy_penalty)
if efficiency_score > self.novelty_threshold:
return True, efficiency_score, f"UPLOAD: 效率={efficiency_score:.4f}"
else:
return False, efficiency_score, f"SKIP: 效率不足 ({efficiency_score:.4f} < {self.novelty_threshold})"
class StalenessAwareProximalTerm(nn.Module):
"""
过期感知近端项
动态调整全局锚强度，基于更新年龄
"""
def __init__(self, mu_max: float = 0.1, tau_max: int = 10):
super().__init__()
self.mu_max = mu_max
self.tau_max = tau_max
def compute_proximal_loss(self,
local_params: Dict[str, torch.Tensor],
global_params: Dict[str, torch.Tensor],
staleness: int) -> torch.Tensor:
"""
计算过期感知近端损失
staleness越大 → mu越大 → 更强的全局锚定
"""
mu = self.mu_max * min(staleness / self.tau_max, 1.0)
proximal_loss = 0.0
for key in local_params:
if key in global_params:
diff = local_params[key] - global_params[key]
proximal_loss += mu * 0.5 * (diff ** 2).sum()
return proximal_loss
if __name__ == "__main__":
gate = FedQoSGate()
print("=== FedQoS门控测试 ===\n")
print("阶段1: 训练门控")
cases = [
(0.5, 0.8, "资源充足"),
(0.1, 0.8, "缓冲区不足"),
(0.5, 0.1, "能量不足"),
]
for buf, eng, desc in cases:
train, reason = gate.training_gate(buf, eng)
print(f" {desc}: buffer={buf}, energy={eng} → {reason}")
print("\n阶段2: 传输门控")
local = {"layer1": torch.randn(10, 5), "layer2": torch.randn(5, 3)}
global_p = {"layer1": torch.randn(10, 5) * 0.1, "layer2": torch.randn(5, 3) * 0.1}
upload, score, reason = gate.transmission_gate(local, global_p, 50.0, 0.2)
print(f" 延迟50ms, 能量0.2J: {reason}")
upload, score, reason = gate.transmission_gate(local, global_p, 200.0, 0.8)
print(f" 延迟200ms, 能量0.8J: {reason}")
print("\n=== 论文关键结果 ===")
print("通信开销降低: 76.7%")
print("延迟成本降低: 26.0%")
print("精度损失: 仅边际 (接近FedAvg)")
print("QoS违规: 大幅减少")
|
