┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│  音频采集阵列 │────▶│  声源定位/VAD │────▶│                 │
└─────────────┘     └──────────────┘     │                 │
                                           │  多模态注意力融合  │◀── 语义/上下文特征
┌─────────────┐     ┌──────────────┐     │  (核心计算模块)   │
│  视频采集阵列 │────▶│  人脸/人体检测 │────▶│                 │
└─────────────┘     └──────────────┘     │                 │
                                           └────────┬────────┘
                                                    ▼
                                         ┌────────────────────┐
                                         │  决策与调度引擎     │
                                         │  - 发言人锁定       │
                                         │  - 构图策略生成     │
                                         │  - 码率/分辨率分配  │
                                         └────────┬───────────┘
                                                  ▼
                                         ┌────────────────────┐
                                         │  视频流合成与推流   │
                                         └────────────────────┘

# 伪代码：音频特征提取
class AudioEncoder(nn.Module):
    def __init__(self, dim=256):
        super().__init__()
        self.frontend = nn.Sequential(
            nn.Conv2d(1, 64, kernel_size=3, stride=2, padding=1),
            nn.BatchNorm2d(64), nn.ReLU(),
            # ... ResNet-18 backbone ...
        )
        self.temporal = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(d_model=dim, nhead=4, batch_first=True),
            num_layers=3
        )
        self.proj = nn.Linear(dim, dim)

    def forward(self, mel_spec):  # [B, 1, T, F]
        x = self.frontend(mel_spec).flatten(2).transpose(1, 2)  # [B, T', dim]
        x = self.temporal(x)
        return self.proj(x[:, -1])  # 取最后一帧作为全局表征 [B, dim]

def compute_viewport(attn_scores, bboxes, frame_shape, config):
    """
    attn_scores: [N] 归一化概率
    bboxes: [N, 4] 归一化坐标 (x1, y1, x2, y2)
    返回: 裁剪区域 (cx, cy, w, h) 归一化坐标
    """
    # 1. 选取 Top-K 候选人 (K=1 或 2)
    topk_idx = attn_scores.topk(min(2, len(attn_scores))).indices
    
    # 2. 加权融合目标框
    weights = attn_scores[topk_idx] / attn_scores[topk_idx].sum()
    target_box = (weights[:, None] * bboxes[topk_idx]).sum(0)
    
    # 3. 平滑跟踪 (EMA)
    smoothed_box = ema_update(prev_box, target_box, alpha=0.3)
    
    # 4. 构图规则：保留头部空间、遵循三分法
    cx, cy, w, h = smoothed_box
    cy -= 0.08 * h  # 向上偏移留白
    w = h = max(w, h) * 1.3  # 正方形裁剪，扩展 30%
    
    # 5. 边界裁剪
    return clamp_viewport(cx, cy, w, h, margin=0.05)

graph LR
    A[网络探测: RTT/丢包/带宽] --> B{带宽预算}
    B -->|充裕 > 8Mbps| C[4K 主流 + 1080p 辅流]
    B -->|中等 2-8Mbps| D[1080p 主流 + 720p 辅流 + SVC 基础层]
    B -->|受限 < 2Mbps| E[720p 单流 + 关键帧优先 + FEC]
    C --> F[构图策略: 多画面并行]
    D --> G[构图策略: 主画面+小窗]
    E --> H[构图策略: 单人聚焦+降帧率]

graph LR
    subgraph 终端/边缘 SoC
        A[摄像头/采集卡] -->|V4L2 Buffer (DMA-BUF FD)| B[硬解码/ISP]
        B -->|DMA-BUF FD| C[预处理 Kernel (NEON/CUDA/RKNPU)]
        C -->|DMA-BUF FD| D[推理 Engine (TensorRT/RKNN/ONNX Runtime)]
        D -->|DMA-BUF FD| E[后处理/合成 Kernel]
        E -->|DMA-BUF FD| F[硬编码器 (H.264/H.265/AV1 SVC)]
    end
    F -->|RTP Packets| G[网络发送]

# 伪代码：动态批次调度器核心逻辑
class SLOBatcher:
    def __init__(self, max_batch=8, slo_ms=30, model_profile: Dict[int, float]):
        self.queue = PriorityQueue()  # (deadline_ts, stream_id, frame_ptr)
        self.model_profile = model_profile  # {batch_size: latency_ms}
        self.max_batch = max_batch

    def push(self, frame, stream_id, arrival_ts):
        deadline = arrival_ts + self.slo_ms
        self.queue.put((deadline, stream_id, frame))
        self._try_schedule()

    def _try_schedule(self):
        now = time.monotonic()
        # 1. 收集已到期或即将到期的帧
        batch = []
        while not self.queue.empty() and len(batch) < self.max_batch:
            deadline, sid, frame = self.queue.queue[0]
            if deadline <= now + 2:  # 2ms 调度开销余量
                batch.append(self.queue.get()[2])
            else:
                break
        
        if not batch: return

        # 2. 选择最优批次大小 (吞吐/延迟 Pareto 前沿)
        optimal_bs = max(batch_size for batch_size, lat in self.model_profile.items() 
                         if lat <= self.slo_ms and batch_size <= len(batch))
        
        # 3. 执行推理 (零拷贝)
        infer_batch(batch[:optimal_bs])
        # 4. 剩余帧重入队或标记丢帧
        for f in batch[optimal_bs:]:
            self.queue.put((time.monotonic() + self.slo_ms, f.stream_id, f))

BL:  IDR ---- P ---- P ---- P ---- IDR ...
EL1:        P(ref BL) ---- P ---- P(ref BL) ...
EL2:                 P(ref EL1) ---- P ...

// 核心状态机 (简化)
enum JitterState { FAST_START, STEADY, SLOW_DRAIN, REBUFFER };

struct AdaptiveJitterBuffer {
    int64_t target_delay_ms;      // 当前目标延迟 (40-300ms)
    int64_t min_delay_ms = 40;    // 硬件解码最小延迟
    int64_t max_delay_ms = 500;   // 用户可感知上限
    double  loss_rate_ewma;       // 指数加权丢包率
    double  delay_gradient;       // 网络延迟趋势
    
    void on_packet_arrival(Packet* pkt) {
        update_estimators(pkt);
        adjust_target_delay();
        push_to_decode_queue(pkt);
    }
    
    void adjust_target_delay() {
        if (state == FAST_START) {
            target_delay_ms = min(max_delay_ms, target_delay_ms * 1.5);
            if (queue_depth > 3) state = STEADY;
        } else if (state == STEADY) {
            // 基于梯度与丢包率的 PID 控制
            double error = (loss_rate_ewma - 0.001) * 1000 + delay_gradient * 10;
            target_delay_ms = clamp(target_delay_ms + error, min_delay_ms, max_delay_ms);
            
            if (loss_rate_ewma > 0.05) state = SLOW_DRAIN; // 主动排队吸收抖动
            if (queue_empty) state = REBUFFER;
        }
        // ... SLOW_DRAIN/REBUFFER 逻辑
    }
};

┌──────────────┐     ┌──────────────┐     ┌─────────────────┐     ┌──────────────┐
│  线上推理日志 │────▶│  难例自动挖掘 │────▶│  人工/大模型标注 │────▶│  数据版本管理 │
│  (采样 1%-5%) │     │  (不确定性/分布│     │  (半自动化管线)  │     │  (DVC/LakeFS) │
└──────────────┘     │   偏移检测)   │     └─────────────────┘     └──────┬───────┘
                     └──────────────┘                                    │
                            ▲                                           │
                            │                                           ▼
                     ┌──────┴──────┐     ┌──────────────┐     ┌─────────────────┐
                     │  影子模式验证 │◀────│  自动化训练   │◀────│  特征存储/样本  │
                     │  (A/B 测试)   │     │  (Kubeflow)   │     │  管理 (Feast)   │
                     └─────────────┘     └──────────────┘     └─────────────────┘

graph TD
    A[原始音视频流] --> B{本地感知引擎}
    B -->|人脸框/关键点/Embedding| C[特征向量 (512-d)]
    B -->|VAD/声源角度| D[音频元数据]
    C --> E[联邦学习本地更新]
    D --> E
    E -->|加密梯度/模型增量| F[安全聚合服务器 (TEE/MPC)]
    F -->|全局模型| G[终端分发]
    
    A -.->|仅在云端录制模式| H[加密转码/存储]
    H --> I[访问控制 (RBAC/ABAC) + 审计日志]

// 典型 Span 结构 (OpenTelemetry Semantic Conventions)
{
  "trace_id": "a1b2c3d4...",
  "span_id": "e5f6g7h8...",
  "name": "inference.attention_fusion",
  "kind": "INTERNAL",
  "attributes": {
    "net.host.name": "edge-gpu-03",
    "net.host.port": 8080,
    "ai.model.name": "cross_attn_v3.2",
    "ai.model.version": "20241015-abc123",
    "ai.inference.batch_size": 4,
    "ai.inference.input_shape": "[4, 256]",
    "ai.inference.latency_ms": 18.5,
    "meeting.id": "m_987654",
    "stream.id": "s_12345",
    "user.id": "u_556677"
  },
  "events": [
    {"name": "cache_hit", "attributes": {"cache.layer": "L2", "key": "speaker_embed_001"}},
    {"name": "fallback_triggered", "attributes": {"reason": "low_confidence", "fallback_to": "visual_only"}}
  ]
}

# 规则示例：会议加入失败率飙升
- alert: MeetingJoinFailureSpike
  expr: |
    sum(rate(meeting_join_failure_total[5m])) by (region, sdk_version)
    /
    sum(rate(meeting_join_attempt_total[5m])) by (region, sdk_version)
    > 0.05
  for: 3m
  labels:
    severity: critical
    domain: signaling
  annotations:
    summary: "区域 {{ $labels.region }} SDK {{ $labels.sdk_version }} 入会失败率 > 5%"
    runbook: "https://wiki.example.com/rca/join_failure"
    # 自动关联查询
    related_queries:
      - "signaling_server_error_rate{region='{{ $labels.region }}'}"
      - "ice_candidate_failure{region='{{ $labels.region }}'}"
      - "cdn_edge_health{region='{{ $labels.region }}'}"
      - "certificate_expiry{region='{{ $labels.region }}'}"