┌─────────────────────────────────────────────────────────────┐
│                      云端参数服务器                          │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │ 全局模型聚合 │  │ 差分隐私引擎 │  │ 设备选调度器 │          │
│  │ (FedAvg/FedProx)│  │ (Gaussian DP) │  │ (资源感知)   │          │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘          │
└─────────┼────────────────┼────────────────┼──────────────────┘
          │ 加密信道        │ 隐私预算分配    │ 下发全局模型
          ▼                ▼                ▼
┌─────────────────────────────────────────────────────────────┐
│                      边缘网关 / 接入层                        │
│  ┌─────────────────────────────────────────────────────┐    │
│  │           安全聚合协议 (SecAgg / Bonawitz et al.)    │    │
│  │  防止单点窃取单端梯度；容忍 Dropout 设备不影响聚合    │    │
│  └─────────────────────────────────────────────────────┘    │
└────────────────────────────────┬────────────────────────────┘
                                 │ 加密梯度上传
        ┌────────────────────────┼────────────────────────┐
        ▼                        ▼                        ▼
┌───────────────┐        ┌───────────────┐        ┌───────────────┐
│  终端设备 A   │        │  终端设备 B   │  ...   │  终端设备 N   │
│ (会议室/笔记本)│        │  (居家办公)   │        │  (移动端/户外)│
│ ┌───────────┐ │        │ ┌───────────┐ │        │ ┌───────────┐ │
│ │本地训练器 │ │        │ │本地训练器 │ │        │ │本地训练器 │ │
│ │(MobileNetV3│ │        │ │(MobileNetV3│ │        │ │(MobileNetV3│ │
│ │ + Attention)│ │        │ │ + Attention)│ │        │ │ + Attention)│ │
│ └───────────┘ │        │ └───────────┘ │        │ └───────────┘ │
│ ┌───────────┐ │        │ ┌───────────┐ │        │ ┌───────────┐ │
│ │隐私预算核算│ │        │ │隐私预算核算│ │        │ │隐私预算核算│ │
│ │(RDP Account)│ │        │ │(RDP Account)│ │        │ │(RDP Account)│ │
│ └───────────┘ │        │ └───────────┘ │        │ └───────────┘ │
└───────────────┘        └───────────────┘        └───────────────┘

# 伪代码：RDP 账户器核心逻辑
class RDPAccountant:
    def __init__(self, noise_multiplier: float, sample_rate: float):
        self.sigma = noise_multiplier
        self.q = sample_rate
        self.orders = [1 + x / 10. for x in range(1, 100)] + list(range(12, 64))
        self.rdp = np.zeros_like(self.orders, dtype=float)
    
    def step(self):
        # 单步 RDP 计算 (基于 Gaussian Mechanism)
        for i, alpha in enumerate(self.orders):
            self.rdp[i] += self._compute_rdp_gaussian(alpha)
    
    def get_epsilon(self, delta: float) -> float:
        # RDP -> (ε, δ)-DP 转换
        eps = self.rdp - np.log(delta) / (self.orders - 1)
        return float(np.min(eps))
    
    def _compute_rdp_gaussian(self, alpha: float) -> float:
        # 采样高斯机制 RDP 解析式 (Mironov 2019)
        return self.q**2 * alpha / (2 * self.sigma**2)  # 简化式，实际含高阶项

graph LR
    A[代码提交] --> B{单元测试/静态扫描}
    B --> C[构建训练镜像<br/>Base: Ubuntu 22.04 + PyTorch 2.3 + TFLite]
    C --> D[模拟器集成测试<br/>100 虚拟客户端/非IID模拟]
    D --> E{指标达标?<br/>Macro-F1>90% / ε<12.5}
    E -- 否 --> F[阻断/告警]
    E -- 是 --> G[Canary 灰度发布<br/>内网 50 台真机]
    G --> H[隐私审计自动化<br/>RDP Accountant 回放日志]
    H --> I{审计通过?}
    I -- 否 --> F
    I -- 是 --> J[全量发布<br/>签名分发/版本锁定]
    J --> K[线上监控看板<br/>Drift/Performance/Privacy]

# features/v1/mel_spectrogram.py (随模型 v1.2.0 发布，语义化版本绑定)
class MelSpectrogramExtractor:
    # 硬编码超参，禁止运行时配置
    SAMPLE_RATE = 16000
    N_FFT = 512
    HOP_LENGTH = 160      # 10ms 帧移
    WIN_LENGTH = 400      # 25ms 窗长
    N_MELS = 40
    F_MIN = 20.0
    F_MAX = 8000.0
    POWER = 2.0
    NORM = "slaney"       # 关键：必须与训练端 librosa/torchaudio 行为完全一致
    MEL_SCALE = "htk"
    
    def __init__(self):
        # 预计算 mel 基矩阵，避免重复计算，保证位级一致
        self.mel_basis = self._build_mel_basis() 
        self.window = torch.hann_window(self.WIN_LENGTH, periodic=True)
    
    @torch.no_grad()
    def forward(self, waveform: torch.Tensor) -> torch.Tensor:
        # waveform: [1, T], int16 或 float32 [-1, 1]
        spec = torch.stft(waveform, self.N_FFT, self.HOP_LENGTH, self.WIN_LENGTH,
                          self.window, return_complex=True, center=True, pad_mode="reflect")
        power = spec.abs().pow(self.POWER)  # [1, F, T]
        mel = torch.matmul(self.mel_basis.to(power.device), power)  # [1, 40, T]
        log_mel = torch.log(torch.clamp(mel, min=1e-10))
        # 逐通道 CMVN (Cepstral Mean Variance Normalization)
        mean = log_mel.mean(dim=-1, keepdim=True)
        std = log_mel.std(dim=-1, keepdim=True).clamp_min(1e-5)
        return (log_mel - mean) / std  # [1, 40, T]

Audio Callback (10ms 周期, 最高优先级)
│
├─ 1. 数据搬移 & 预处理 (0.3 ms)
│   └─ RingBuffer -> Float32 -> MelExtractor (40x100)
│
├─ 2. 噪声分类推理 (目标 < 4 ms)
│   ├─ TFLite/NNAPI/CoreML Invoke
│   ├─ 输出: 12-dim Logits -> Softmax -> Top-3
│   └─ 平滑滤波: EMA (α=0.3) + 最小持续时长 300ms 防抖
│
├─ 3. 参数映射查表 (0.05 ms)
│   └─ 读取 JSON -> 结构体 -> 原子写入共享内存
│
├─ 4. 下游模块消费 (并行/流水)
│   ├─ VAD (读取平滑后类别概率)
│   ├─ ANS (读取 overdrive/alpha_noise)
│   ├─ AEC (读取 filter_len/stepsize)
│   └─ AGC (读取 target_level/limiter)
│
└─ 5. 埋点上报 (异步线程, 非实时)
    └─ 类别分布/推理耗时/参数生效延迟

# server/robust_aggregator.py
class RobustAggregator:
    def __init__(self, config):
        self.cos_sim_threshold = config.get("cos_sim_threshold", 0.1)  # 方向一致性
        self.norm_bound = config.get("norm_bound", 3.0)                # 范数界 (Median + 3*MAD)
        self.loss_val_threshold = config.get("loss_val_threshold", 2.5) # 验证集 Loss 异常倍数
        self.history = []  # 存储最近 10 轮全局模型用于趋势判断
    
    def filter_and_aggregate(self, client_updates: List[ClientUpdate], global_model: nn.Module) -> Dict:
        """
        client_updates: [{client_id, delta_params (dict), num_samples, local_loss, val_loss, grad_norm}]
        return: {filtered_updates, aggregated_delta, rejected_ids, stats}
        """
        # 1. 范数过滤 (剔除梯度爆炸/投毒放大)
        norms = [u.grad_norm for u in client_updates]
        median_norm = np.median(norms)
        mad = np.median(np.abs(norms - median_norm))
        upper_bound = median_norm + 3 * 1.4826 * mad
        norm_pass = [u for u in client_updates if u.grad_norm <= upper_bound]
        
        # 2. 余弦相似度过滤 (剔除方向对抗/后门)
        # 计算每个更新与"临时平均方向"的夹角
        temp_avg = self._weighted_average(norm_pass)
        cos_sims = []
        for u in norm_pass:
            flat_u = torch.cat([v.flatten() for v in u.delta_params.values()])
            flat_avg = torch.cat([v.flatten() for v in temp_avg.values()])
            cos_sims.append(F.cosine_similarity(flat_u.unsqueeze(0), flat_avg.unsqueeze(0)).item())
        
        cos_pass = [u for u, sim in zip(norm_pass, cos_sims) if sim >= self.cos_sim_threshold]
        
        # 3. 验证集 Loss 过滤 (剔除标签噪声/数据分布严重偏移)
        # 服务端持有小规模干净验证集 (500 样本，覆盖 12 类)
        val_losses = []
        for u in cos_pass:
            # 临时应用更新评估
            self._apply_delta(global_model, u.delta_params, weight=1.0)
            loss = self._eval_on_clean_val(global_model)
            self._revert_delta(global_model, u.delta_params) # 恢复
            val_losses.append(loss)
        
        median_loss = np.median(val_losses)
        loss_pass = [u for u, loss in zip(cos_pass, val_losses) if loss <= median_loss * self.loss_val_threshold]
        
        # 最终加权聚合 (FedAvg 加权)
        final_delta = self._weighted_average(loss_pass)
        
        # 记录统计
        stats = {
            "total": len(client_updates),
            "norm_rejected": len(client_updates) - len(norm_pass),
            "cos_rejected": len(norm_pass) - len(cos_pass),
            "loss_rejected": len(cos_pass) - len(loss_pass),
            "final_participants": len(loss_pass),
            "median_cos_sim": float(np.median(cos_sims)) if cos_sims else 0.0
        }
        return {"aggregated_delta": final_delta, "rejected_ids": [u.client_id for u in client_updates if u not in loss_pass], "stats": stats}