┌─────────────────────────────────────────────────────────────┐
│                    LLM-Driven RTC Anomaly Pipeline          │
├─────────────────────────────────────────────────────────────┤
│  L4: 闭环验证与持续进化层  (Feedback Loop & Continuous Evo)  │
│  ├─ A/B 测试框架  ├─ 人工标注回流  ├─ 模型蒸馏与量化部署       │
├─────────────────────────────────────────────────────────────┤
│  L3: 在线推理与决策层      (Online Inference & Decision)      │
│  ├─ 流式特征聚合  ├─ 多模型集成投票  ├─ 降级熔断策略          │
├─────────────────────────────────────────────────────────────┤
│  L2: 模型训练与微调层      (Training & Fine-tuning)           │
│  ├─ 预训练适配  ├─ 指令微调 (SFT)  ├─ RLHF 偏好对齐           │
├─────────────────────────────────────────────────────────────┤
│  L1: 数据采集与特征工程层  (Data Collection & Feature Eng)    │
│  ├─ 多源日志汇聚  ├─ 会话级拼接  ├─ 结构化/非结构化双轨特征    │
└─────────────────────────────────────────────────────────────┘

# 伪代码：统一日志 Schema (Apache Iceberg / Hudi)
@dataclass
class SignalingEvent:
    # 标识维度
    call_id: str
    conference_id: str
    participant_id: str
    timestamp_ms: int
    sequence_id: int          # 单会话内单调递增
    
    # 协议维度
    protocol: Literal["SIP", "WebRTC", "WHIP", "WHEP", "XMPP"]
    message_type: str         # INVITE, 200 OK, CANDIDATE, SDP_OFFER...
    direction: Literal["inbound", "outbound", "internal"]
    
    # 结构化载荷
    payload_json: Dict        # 解析后的 SDP/JSON/Protobuf
    raw_bytes: bytes          # 原始报文（保留用于溯源）
    
    # 网络/传输维度
    src_ip: str
    dst_ip: str
    transport: Literal["UDP", "TCP", "TLS", "QUIC"]
    rtt_ms: Optional[int]
    retrans_count: int
    
    # 业务标签（后续回填）
    is_anomaly: Optional[bool] = None
    root_cause_label: Optional[str] = None

sequenceDiagram
    participant Client A
    participant Signaling GW
    participant Media Server
    participant Client B
    
    Note over Client A,Client B: 完整会话切片窗口 = [Join, Leave] ∪ [Rejoin, Releave]*
    
    Client A->>Signaling GW: JOIN (Offer)
    Signaling GW->>Media Server: ALLOCATE
    Media Server-->>Signaling GW: ANSWER
    Signaling GW-->>Client A: 200 OK (Answer)
    Client A->>Signaling GW: CANDIDATE (Trickle ICE)
    ...
    Client A->>Signaling GW: LEAVE (BYE)

def slice_session(events: List[SignalingEvent], max_len: int = 4096) -> List[SessionSlice]:
    """
    滑动窗口 + 关键锚点对齐：
    1. 以 JOIN/REJOIN/LEAVE 为硬边界
    2. 窗口内按 sequence_id 排序
    3. 超过 max_len tokens 时，保留首尾各 30% + 中间异常密度最高 40%
    """

{
  "instruction": "以下是某 WebRTC 会议的信令交互片段，请判断是否存在异常，给出根因分析与修复建议。",
  "input": "[SESSION: conf_7x9k2m]n1. Client->GW: INVITE (Offer: VP8/90000, opus/48000)n2. GW->MS: ALLOCATE (mid=audio)n3. MS->GW: 200 OK (Answer: opus/48000, ptime=20)n4. GW->Client: 200 OK (Answer)n5. Client->GW: CANDIDATE (host, 192.168.1.5:50000, udp)n6. GW->Client: CANDIDATE (srflx, 203.0.113.10:60000, udp)n7. [DTLS HANDSHAKE TIMEOUT @ 5000ms]n8. Client->GW: REINVITE (Offer: VP8/90000, opus/48000, ICE_RESTART)n...",
  "output": {
    "anomaly_detected": true,
    "anomaly_span": [7],
    "root_cause": "DTLS 握手超时导致 ICE 连接失败。典型原因：对端防火墙/NAT 设备丢弃 UDP 端口 50000-60000 范围的 DTLS 包，或对端媒体引擎版本过低不支持 DTLS 1.2。",
    "severity": "P1",
    "suggested_actions": [
      "建议客户端启用 ICE-TCP / TURN-TLS 回退",
      "媒体服务器侧开启 DTLS 1.3 支持并降级兼容 1.2",
      "排查对端网络出口 ACL 策略，放行 UDP 49152-65535"
    ],
    "confidence": 0.93
  }
}

# Flink SQL 定义：会话级滚动聚合 (每 500ms 触发一次推理)
CREATE TABLE rtc_signaling_enriched (
    call_id STRING,
    conference_id STRING,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    -- 结构化聚合特征
    msg_count BIGINT,
    retrans_ratio DOUBLE,
    avg_rtt_ms DOUBLE,
    sdp_renego_count INT,
    ice_state_transitions STRING,  -- JSON: ["CHECKING","CONNECTED","FAILED"]
    error_code_histogram MAP<STRING, INT>,
    -- 非结构化语料（截断前 2k chars）
    raw_log_concat STRING,
    -- 主键
    PRIMARY KEY (call_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'rtc-session-features',
    ...
);

flowchart TD
    A[会话特征向量] --> B{规则引擎快速过滤}
    B -- 明显已知模式 --> C[直接命中规则库nLatency < 5ms]
    B -- 疑似/未知 --> D[LLM 推理队列]
    D --> E[Qwen2-7B 主模型]
    D --> F[Llama-3-70B Teachern(异步复核, 采样 5%)]
    E --> G[结果融合器]
    F --> G
    G --> H{置信度 > 0.85?}
    H -- 是 --> I[输出告警 + 根因 + 动作]
    H -- 否 --> J[降级: 仅标记可疑n推送人工复核工单]
    I --> K[写入时序库 + 告警中心]
    J --> K

{
  "alert_id": "alt_rtc_20240615_001234",
  "timestamp": "2024-06-15T14:32:11.234Z",
  "conference_id": "conf_7x9k2m",
  "participant_id": "user_abc123",
  "anomaly_type": "ICE_CONNECTION_FAILURE",
  "severity": "P1",
  "confidence": 0.93,
  "evidence_span": {
    "start_seq": 7,
    "end_seq": 9,
    "key_logs": ["DTLS HANDSHAKE TIMEOUT", "ICE STATE: FAILED"]
  },
  "root_cause_natural": "对端企业防火墙拦截 UDP 49152-65535 端口段，导致 DTLS 握手包丢失，ICE 连通性检查失败。",
  "suggested_actions": [
    {"type": "CONFIG_CHANGE", "target": "media_server", "action": "enable_turn_tls_fallback", "priority": 1},
    {"type": "NETWORK_POLICY", "target": "customer_firewall", "action": "allow_udp_range_49152_65535", "priority": 2},
    {"type": "CLIENT_SDK", "target": "web_sdk_v2.3.1", "action": "upgrade_to_v2.4.0_ice_tcp_support", "priority": 3}
  ],
  "knowledge_graph_links": [
    "https://wiki.internal/rtc/troubleshooting/ice-failure-firewall",
    "https://rfc-editor.org/rfc/rfc8445#section-9"
  ]
}

流量分发：
  95% → 规则引擎 (Baseline)
   5% → LLM Pipeline (Shadow, 仅记录不告警)
  
对比指标（持续 14 天）：
  - 召回率提升 ΔRecall
  - 误报率变化 ΔPrecision
  - 人工处理工单量变化 ΔManualTickets
  - 平均处理时长 ΔMTTR

# 低置信度样本自动入池，运维同学每日标注 20 条
def active_learning_loop():
    low_conf_samples = query("confidence < 0.7 AND not_labeled")
    for sample in low_conf_samples[:20]:
        push_to_label_platform(sample)
    
    # 每周重训练 LoRA Adapter (增量 500 步)
    if new_labeled_count > 200:
        launch_lora_finetune_job(
            base_model="qwen2-7b-lora-v3",
            new_data=fetch_new_labels(),
            output="qwen2-7b-lora-v4"
        )

# rfc_ingest.py
import re
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer

# 1. 专用 RFC 分割器：保留 Section 编号、伪代码块、ABNF 语法块
RFC_SPLITTER = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=[
        "nn### ", "nn## ", "nn# ",      # Markdown 标题
        "nnSection ", "nnAppendix ",     # RFC 章节
        "n```", "n```n",                 # 代码块边界
        "nn", "n", ". ", " ", ""         # 兜底
    ],
    keep_separator=True
)

# 2. 元数据提取：正则捕获 RFC 编号、Section 编号、协议关键字
def extract_meta(text: str, source_url: str) -> dict:
    rfc_id = re.search(r'RFCs+(d+)', text, re.IGNORECASE)
    section = re.search(r'Sections+(d+(.d+)*)', text)
    proto_kws = set(re.findall(r'b(SIP|SDP|ICE|DTLS|SRTP|RTP|RTCP|TURN|STUN|WebRTC)b', text, re.IGNORECASE))
    return {
        "rfc": f"RFC{rfc_id.group(1)}" if rfc_id else "UNKNOWN",
        "section": section.group(1) if section else "0",
        "protocols": list(proto_kws),
        "source_url": source_url,
        "token_count": len(text.split())
    }

# 3. 批量入库（Milvus + Neo4j 双写）
def ingest_rfc_corpus(rfc_dir: str):
    embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')  # 中英双语强
    for md_file in Path(rfc_dir).rglob("*.md"):
        raw = md_file.read_text(encoding='utf-8')
        chunks = RFC_SPLITTER.split_text(raw)
        vectors = embedder.encode(chunks, batch_size=32, show_progress_bar=True)
        metas = [extract_meta(c, f"file://{md_file}") for c in chunks]
        # Milvus 插入
        milvus_client.insert("rfc_chunks", [
            {"vector": v.tolist(), "text": c, **m} for v, c, m in zip(vectors, chunks, metas)
        ])
        # Neo4j 建立 RFC->Section->Keyword 索引边
        for m in metas:
            neo4j_session.run("""
                MERGE (r:RFC {id: $rfc})
                MERGE (s:Section {id: $rfc + '_' + $section})
                MERGE (r)-[:HAS_SECTION]->(s)
                WITH s
                UNWIND $protocols AS p
                MERGE (k:Keyword {name: p})
                MERGE (s)-[:MENTIONS]->(k)
            """, **m)

flowchart LR
    A[用户 Query / 异常片段] --> B[Query Rewritern(Llama-3-8B 扩展同义词/缩写)]
    B --> C1[稀疏检索 BM25n(Error Code/Keyword 精准匹配)]
    B --> C2[稠密检索 BGEn(语义相似 Top-50)]
    B --> C3[图谱检索 Cyphern(错误码->根因->动作 2跳)]
    C1 --> D[RRF 融合n(Rank Fusion k=60)]
    C2 --> D
    C3 --> D
    D --> E[Cross-Encoder 重排n(bge-reranker-large Top-10 -> Top-3)]
    E --> F[上下文压缩n(LLMLingua-2 压缩至 1k tokens)]
    F --> G[注入 Prompt]

# guardrail.py
def verify_citations(response: str, retrieved_docs: List[Doc]) -> Tuple[bool, str]:
    """
    1. 正则提取响应中所有 [RFCxxxx Section x.x] 或 [DocID: xxx] 引用
    2. 逐一在 retrieved_docs 中匹配原文片段（模糊匹配阈值 0.85）
    3. 若有引用无法匹配 -> 标记幻觉，触发拒答/重试
    """
    citations = re.findall(r'[(RFCd+|DocID:[^]]+)]', response)
    for cit in citations:
        matched = False
        for doc in retrieved_docs:
            if cit.startswith("RFC") and cit in doc.metadata.get("rfc", ""):
                matched = True; break
            if cit.startswith("DocID:") and cit.split(":")[1] == doc.metadata.get("doc_id"):
                matched = True; break
        if not matched:
            return False, f"Unverifiable citation: {cit}"
    return True, "OK"

# 推理循环中集成
for attempt in range(3):
    resp = llm.generate(prompt_with_rag)
    ok, msg = verify_citations(resp, retrieved)
    if ok: break
    prompt_with_rag += f"n[SYSTEM] 上一轮输出包含不可验证引用: {msg}。请仅基于提供的上下文回答，若无依据请明确声明“知识库无覆盖”。"
else:
    resp = "【低置信度】知识库无法支撑该结论，建议人工复核。"

# chaos-experiment.yaml (Argo Workflow Template)
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  name: rtc-chaos-validation-{{workflow.uid}}
spec:
  entrypoint: main
  templates:
  - name: main
    steps:
    - - name: inject-network-chaos
        template: network-chaos
        arguments:
          parameters:
          - name: target_ns
            value: rtc-prod
          - name: loss_percent
            value: "10"
    - - name: run-inference-shadow
        template: shadow-inference
        arguments:
          parameters:
          - name: duration_min
            value: "30"
    - - name: evaluate-results
        template: evaluation
        arguments:
          parameters:
          - name: baseline_metrics
            value: "{{steps.inject-network-chaos.outputs.parameters.baseline}}"
          - name: shadow_metrics
            value: "{{steps.run-inference-shadow.outputs.parameters.metrics}}"

  - name: network-chaos
    inputs:
      parameters:
      - name: target_ns
      - name: loss_percent
    container:
      image: chaos-mesh/chaos-daemon:latest
      command: ["/bin/sh", "-c"]
      args:
      - |
        kubectl apply -f - <<EOF
        apiVersion: chaos-mesh.org/v1alpha1
        kind: NetworkChaos
        metadata:
          name: rtc-signal-loss-{{workflow.uid}}
          namespace: {{inputs.parameters.target_ns}}
        spec:
          action: loss
          mode: one
          selector:
            namespaces: ["{{inputs.parameters.target_ns}}"]
            labelSelectors:
              app: signaling-gateway
          loss:
            loss: "{{inputs.parameters.loss_percent}}"
            correlation: "25"
          duration: "300s"
        EOF
        # 采集注入前 5min 基线指标
        python collect_baseline.py --output /tmp/baseline.json
        echo "baseline=$(cat /tmp/baseline.json | jq -c .)" >> $ARGO_OUTPUTS/parameters

  - name: shadow-inference
    inputs:
      parameters:
      - name: duration_min
    container:
      image: rtc-llm-inference:shadow-v4
      command: ["python", "shadow_runner.py"]
      args: ["--duration", "{{inputs.parameters.duration_min}}", "--output", "/tmp/shadow_metrics.json"]
      resources:
        limits:
          nvidia.com/gpu: 1

  - name: evaluation
    inputs:
      parameters:
      - name: baseline_metrics
      - name: shadow_metrics
    container:
      image: python:3.11-slim
      command: ["python", "evaluate.py"]
      args: ["--baseline", "{{inputs.parameters.baseline_metrics}}", "--shadow", "{{inputs.parameters.shadow_metrics}}"]
      env:
      - name: ALERT_WEBHOOK
        valueFrom: secretKeyRef: {name: alert-webhook, key: url}

def evaluate(baseline: dict, shadow: dict) -> dict:
    # 1. 召回率对比：影子模式发现的真实异常 / 基线规则引擎发现的异常
    recall_gain = (shadow['tp'] - baseline['tp']) / max(baseline['tp'] + baseline['fn'], 1)
    
    # 2. 误报率对比：影子模式误报 / 影子模式总告警
    precision_delta = shadow['precision'] - baseline['precision']
    
    # 3. 根因准确率：仅在双方都告警的样本上对比
    common_cases = set(shadow['alert_ids']) & set(baseline['alert_ids'])
    root_cause_acc = sum(
        1 for aid in common_cases 
        if shadow['root_cause'][aid] == ground_truth[aid]
    ) / max(len(common_cases), 1)
    
    # 4. 延迟 SLA：P99 < 200ms
    latency_ok = shadow['latency_p99_ms'] < 200
    
    # 5. 熔断触发检查：影子模式不应触发熔断
    circuit_breaker_triggered = shadow.get('circuit_breaker_events', 0) == 0
    
    verdict = "PASS" if all([
        recall_gain > 0.15,
        precision_delta > -0.02,
        root_cause_acc > 0.90,
        latency_ok,
        circuit_breaker_triggered
    ]) else "FAIL"
    
    return {
        "verdict": verdict,
        "metrics": {
            "recall_gain": recall_gain,
            "precision_delta": precision_delta,
            "root_cause_acc": root_cause_acc,
            "latency_p99_ms": shadow['latency_p99_ms']
        }
    }

# schemas/alert_card.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Annotated
from enum import Enum

class Severity(str, Enum):
    P0 = "P0"
    P1 = "P1"
    P2 = "P2"
    P3 = "P3"

class ActionType(str, Enum):
    CONFIG_CHANGE = "CONFIG_CHANGE"
    NETWORK_POLICY = "NETWORK_POLICY"
    CLIENT_SDK = "CLIENT_SDK"
    CAPACITY_EXPANSION = "CAPACITY_EXPANSION"

class SuggestedAction(BaseModel):
    type: ActionType
    target: Annotated[str, Field(min_length=1, max_length=64, pattern=r'^[a-z_][a-z0-9_.-]*$')]
    action: Annotated[str, Field(min_length=1, max_length=128)]
    priority: Annotated[int, Field(ge=1, le=5)]
    params: dict = Field(default_factory=dict)  # 结构化参数，供自动化执行器直接调用

class EvidenceSpan(BaseModel):
    start_seq: int
    end_seq: int
    key_logs: Annotated[list[str], Field(min_length=1, max_length=5)]

class AlertCard(BaseModel):
    alert_id: Annotated[str, Field(pattern=r'^alt_rtc_d{8}_d{6}$')]
    timestamp: Annotated[str, Field(pattern=r'^d{4}-d{2}-d{2}Td{2}:d{2}:d{2}.d{3}Z$')]
    conference_id: str
    participant_id: str
    anomaly_type: Annotated[str, Field(pattern=r'^[A-Z_][A-Z0-9_]*$')]  # 规范化枚举
    severity: Severity
    confidence: Annotated[float, Field(ge=0.0, le=1.0)]
    evidence_span: EvidenceSpan
    root_cause_natural: Annotated[str, Field(min_length=10, max_length=500)]
    suggested_actions: Annotated[list[SuggestedAction], Field(min_length=1, max_length=5)]
    knowledge_graph_links: Annotated[list[str], Field(max_length=10)]

    @field_validator('knowledge_graph_links')
    @classmethod
    def validate_urls(cls, v):
        for url in v:
            if not url.startswith(('https://wiki.internal/', 'https://rfc-editor.org/')):
                raise ValueError(f'Invalid KG link domain: {url}')
        return v

# constrained_gen.py
import outlines
from outlines.models import Transformers
from schemas.alert_card import AlertCard

model = Transformers.from_pretrained("Qwen2-7B-Int4-AWQ", device_map="auto")
# 核心：将 Pydantic 模型编译为正则文法，约束解码时仅允许合法 token
generator = outlines.generate.json(model, AlertCard)

# 流式增量解析器：边生成边校验，前端可实时渲染部分字段
def stream_generate(prompt: str):
    streamer = outlines.stream.json(model, AlertCard)
    partial = {}
    for chunk in streamer(prompt):
        # chunk 是已生成的部分 JSON 字符串
        try:
            partial = json.loads(chunk)
            yield {"status": "streaming", "data": partial}
        except json.JSONDecodeError:
            yield {"status": "streaming", "data": chunk}  # 未闭合 JSON，前端可展示原文
    # 最终校验
    try:
        validated = AlertCard.model_validate_json(chunk)
        yield {"status": "completed", "data": validated.model_dump()}
    except ValidationError as e:
        yield {"status": "failed", "error": str(e)}

// Go 消费者示例：使用 json-iterator 增量解析，配合 Schema 校验
func ConsumeAlertStream(reader io.Reader) (<-chan *AlertCard, <-chan error) {
    cards := make(chan *AlertCard, 10)
    errs := make(chan error, 1)
    go func() {
        defer close(cards); defer close(errs)
        dec := jsoniter.NewDecoder(reader)
        // 使用 jsoniter 的延迟解析，仅在需要时解析字段
        for dec.Read() {
            var raw jsoniter.Any = dec.Get()
            // 快速校验必填字段存在性
            if !raw.Get("alert_id").Exists() || !raw.Get("severity").Exists() {
                errs <- fmt.Errorf("missing required field")
                continue
            }
            // 完整反序列化并校验
            var card AlertCard
            if err := raw.ToVal(&card); err != nil {
                errs <- fmt.Errorf("schema validation failed: %w", err)
                continue
            }
            cards <- &card
        }
    }()
    return cards, errs
}

flowchart TB
    subgraph Signaling_Plane [信令平面]
        S1[SIP/WebRTC 信令] --> S2[会话级时序特征n(状态机、重传、SDP 语义)]
    end
    
    subgraph Media_Plane [媒体平面]
        M1[RTP/RTCP 流] --> M2[媒体质量特征n(MOS, Jitter, PLI/FIR/NACK 率,n关键帧间隔, 码率自适应曲线)]
        M3[WebRTC Internals / getStats] --> M2
    end
    
    subgraph Alignment [时序对齐层]
        S2 --> A1[统一时间基: NTP + 逻辑时钟]
        M2 --> A1
        A1 --> A2[滑动窗口对齐: 500ms 步长n同步键: Call-ID + SSRC 映射]
    end
    
    subgraph Fusion_Model [融合模型]
        A2 --> F1[Cross-Attention Fusionn(Q: Signaling, K/V: Media)]
        F1 --> F2[LLM Backbone (Qwen2-7B)n输入: 序列化的融合 Token]
        F2 --> F3[双头输出:n1. 根因分类 (信令/网络/编码/终端)n2. 因果解释生成]
    end

# ssrc_mapper.py
def build_ssrc_mapping(sdp_offer: str, sdp_answer: str, getstats_samples: List[dict]) -> Dict[str, dict]:
    """
    返回: {local_ssrc: {mid, kind, codec, remote_ssrc, cname, participant_id}}
    """
    # 1. 解析 SDP a=ssrc 行 & a=msid 行
    local_ssrcs = parse_ssrc_attributes(sdp_offer)  # 本端发送 SSRC
    remote_ssrcs = parse_ssrc_attributes(sdp_answer) # 对端发送 SSRC (即本端接收 SSRC)
    
    # 2. 关联 getStats 中的 outbound-rtp / inbound-rtp
    mapping = {}
    for stat in getstats_samples:
        if stat['type'] in ('outbound-rtp', 'inbound-rtp'):
            ssrc = str(stat['ssrc'])
            mid = stat.get('mid')
            kind = stat.get('kind')  # audio/video
            codec = stat.get('codecId', '')
            cname = stat.get('cname', '')
            
            # 方向判断
            direction = 'send' if stat['type'] == 'outbound-rtp' else 'recv'
            
            mapping[ssrc] = {
                'mid': mid,
                'kind': kind,
                'codec': codec,
                'direction': direction,
                'cname': cname,
                'remote_ssrc': remote_ssrcs.get(ssrc) if direction == 'send' else local_ssrcs.get(ssrc)
            }
    return mapping

def align_timestamps(signaling_events: List[Event], media_stats: List[Stat], ntp_offset_ms: int) -> List[AlignedFrame]:
    """
    1. 信令事件: timestamp_ms (NTP 校准后)
    2. 媒体统计: local_timestamp (本地单调时钟) + ntp_time (RTCP SR 中的 NTP 时间戳)
    3. 计算媒体侧时钟漂移: drift = (ntp_time - local_timestamp) - ntp_offset_ms
    4. 将媒体统计时间映射到统一 NTP 时间轴
    """
    # 线性回归拟合 drift(t) = a*t + b (每分钟重算一次)
    drift_model = fit_drift_model(media_stats)
    
    aligned = []
    for stat in media_stats:
        unified_ts = stat['local_timestamp'] + drift_model.predict(stat['local_timestamp'])
        # 最近邻匹配信令窗口
        sig_window = find_signaling_window(signaling_events, unified_ts, window_ms=500)
        aligned.append(AlignedFrame(
            timestamp=unified_ts,
            signaling_features=extract_signaling_features(sig_window),
            media_features=extract_media_features(stat),
            ssrc=stat['ssrc']
        ))
    return aligned

<|signaling_start|>
[STATE] JOIN -> INVITE -> 180 -> 200 -> ACK -> CANDIDATE(x12) -> CONNECTED
[SDP_OFFER] v=0 o=- 123 456 IN IP4 192.168.1.5 s= t=0 0 m=audio 50000 RTP/SAVPF 111 a=rtpmap:111 opus/48000/2 a=fmtp:111 minptime=10;useinbandfec=1 a=ssrc:112233 cname:user_a
[SDP_ANSWER] v=0 o=- 789 012 IN IP4 10.0.0.10 s= t=0 0 m=audio 60000 RTP/SAVPF 111 a=rtpmap:111 opus/48000/2 a=fmtp:111 minptime=10 a=ssrc:445566 cname:media_server
<|signaling_end|>

<|media_start|>
[WINDOW_001] t=1000ms SSRC=112233(kind=audio,send) jitter=12ms pl=0.0% nack=0 pli=0 fir=0 bitrate=64kbps mos=4.3
[WINDOW_002] t=1500ms SSRC=112233(kind=audio,send) jitter=45ms pl=2.1% nack=12 pli=0 fir=0 bitrate=32kbps mos=3.1  <-- 码率下降、抖动增大
[WINDOW_003] t=2000ms SSRC=445566(kind=audio,recv) jitter=5ms pl=0.0% nack=0 pli=0 fir=0 bitrate=64kbps mos=4.4
<|media_end|>

<|task|>
基于上述信令与媒体质量时序，判断是否存在异常，给出根因与修复建议。

# 利用网络模拟器 (Mahimahi / netem) + 真实客户端 采集
def collect_hard_negative_samples():
    scenarios = [
        # 场景 1: 单向音频 (防火墙非对称)
        {"name": "one_way_audio_fw", "netem": "loss 0% 10% (downlink only)", "expected": "NETWORK_ASYMMETRIC_FIREWALL"},
        # 场景 2: 视频冻结 (关键帧丢失)
        {"name": "video_freeze_keyframe_loss", "netem": "loss 5% correlation 80% (burst loss on I-frame)", "expected": "VIDEO_KEYFRAME_LOSS"},
        # 场景 3: 编码器降级 (CPU 限频)
        {"name": "encoder_downgrade_cpu_throttle", "cpu_quota": "20%", "expected": "ENCODER_CPU_THROTTLE"},
        # 场景 4: 回声/啸叫 (AEC 失效)
        {"name": "aec_failure", "audio_config": "disable_aec", "expected": "AUDIO_ECHO"},
    ]
    for sc in scenarios:
        run_webrtc_call(duration=60, **sc)
        save_aligned_dataset(sc['name'], sc['expected'])

{
  "anomaly_detected": true,
  "anomaly_type": "VIDEO_FREEZE_INTERMITTENT",
  "root_cause_natural": "媒体质量时序显示：发送端 SSRC=112233 (video) 在 t=1500ms 窗口出现突发丢包 (pl=8.2%, burst=3 packets)，导致关键帧 (I-frame) 丢失，解码端无法恢复参考帧，引发 2.3s 画面冻结。信令层 ICE 状态持续 CONNECTED，无重协商，排除信令层故障。RTCP NACK 重传请求激增但 RTT 稳定，表明网络链路存在间歇性拥塞而非链路中断。",
  "causal_chain": [
    {"step": 1, "event": "网络链路间歇性拥塞 (疑似 WiFi 信道竞争)", "evidence": "media: burst loss @ t=1500ms"},
    {"step": 2, "event": "关键帧分片丢失 (RTP 包序列号不连续)", "evidence": "media: nack list contains seq 1024,1025,1026 (I-frame fragments)"},
    {"step": 3, "event": "解码器参考帧缺失，触发 PLI 请求", "evidence": "media: pli_count spike from 0 to 3 in 200ms"},
    {"step": 4, "event": "远端编码器生成新 I-frame 需要 1 RTT + 编码延迟", "evidence": "media: keyframe_interval jump from 3000ms to 5300ms"},
    {"step": 5, "event": "本端渲染冻结 2.3s", "evidence": "media: freeze_duration=2300ms"}
  ],
  "suggested_actions": [
    {"type": "CLIENT_SDK", "action": "enable_ulpfec_flexfec", "priority": 1, "params": {"fec_overhead": "15%"}},
    {"type": "CLIENT_SDK", "action": "reduce_keyframe_interval", "priority": 2, "params": {"target_interval_ms": 1000}},
    {"type": "NETWORK_POLICY", "action": "qos_marking_dscp_ef", "priority": 3, "params": {"dscp": 46}}
  ]
}

# alert_aggregation.py
def aggregate_alerts(raw_alerts: List[AlertCard], window_min: int = 5) -> List[AggregatedAlert]:
    """
    1. 按 (root_cause_label, anomaly_type, severity) 分组
    2. 时间窗口内合并：受影响会议数、受影响用户数、Top 3 典型案例
    3. 生成「聚合告警卡片」推送给值班，而非单条告警轰炸
    """
    groups = defaultdict(list)
    for alert in raw_alerts:
        key = (alert.root_cause_label, alert.anomaly_type, alert.severity)
        groups[key].append(alert)
    
    aggregated = []
    for (cause, atype, sev), alerts in groups.items():
        if len(alerts) < 3:  # 少量离散告警保持原样
            aggregated.extend(alerts)
            continue
        
        # 构造聚合卡片
        agg = AggregatedAlert(
            aggregate_id=f"agg_{cause}_{int(time.time())}",
            root_cause=cause,
            anomaly_type=atype,
            severity=sev,
            affected_conferences=len(set(a.conference_id for a in alerts)),
            affected_users=len(set(a.participant_id for a in alerts)),
            first_occurrence=min(a.timestamp for a in alerts),
            last_occurrence=max(a.timestamp for a in alerts),
            typical_cases=sorted(alerts, key=lambda x: -x.confidence)[:3],
            suggested_actions=alerts[0].suggested_actions,  # 同根因动作一致
            status="FIRING"
        )
        aggregated.append(agg)
    return aggregated

# knowledge_distillation.py
def distill_knowledge(verified_alerts: List[AlertCard]) -> List[KnowledgeEntry]:
    """
    输入：人工确认闭环的告警 (status=RESOLVED, human_feedback=correct)
    输出：结构化知识条目，入库 Neo4j + 向量库，供 RAG 检索、新人培训、模型微调
    """
    entries = []
    for alert in verified_alerts:
        # 1. 提取通用化模板（脱敏化）
        template = generalize_alert(alert)  # 替换具体 IP/ID 为占位符
        
        # 2. 生成 Runbook (Markdown)
        runbook_md = f"""# 故障处理手册: {template.anomaly_type}
        
## 现象描述
{template.root_cause_natural}

## 典型信令特征

## 典型媒体质量特征
{template.media_signature}

## 根因分析
{template.causal_chain}

## 标准处置步骤 (Runbook)
{chr(10).join(f'{i}. {a.action} (目标: {a.target})' for i, a in enumerate(template.suggested_actions, 1))}

## 验证恢复标准
- 信令: ICE State = CONNECTED, 无重协商
- 媒体: MOS > 4.0, Jitter < 30ms, PL < 0.1%, Freeze = 0

## 关联知识
- RFC: {template.rfc_refs}
- 内部 Wiki: {template.wiki_links}
"""
        entries.append(KnowledgeEntry(
            id=f"kb_{template.anomaly_type}_{hash(template.root_cause_natural)[:8]}",
            anomaly_type=template.anomaly_type,
            root_cause=template.root_cause_label,
            runbook_markdown=runbook_md,
            embedding=embedder.encode(runbook_md),
            tags=["verified", "auto-generated", template.severity.lower()],
            created_at=datetime.utcnow()
        ))
    return entries