class TerminologyLogitsProcessor(LogitsProcessor):
    def __init__(self, tokenizer, term_trie: TermTrie, boost_score: float = 20.0):
        self.tokenizer = tokenizer
        self.trie = term_trie
        self.boost = boost_score

    def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor) -> torch.FloatTensor:
        # 1. 获取当前解码前缀对应的 Trie 节点
        prefix_tokens = input_ids[0].tolist()
        node = self.trie.match_longest_prefix(prefix_tokens)
        
        if node and node.children:
            # 2. 仅允许术语后续 Token，其余置 -inf
            allowed_tokens = [child.token_id for child in node.children.values()]
            mask = torch.full_like(scores, -float('inf'))
            mask[:, allowed_tokens] = 0
            scores = scores + mask
            
            # 3. 对术语 Token 施加正向加分，加速收敛
            scores[:, allowed_tokens] += self.boost
        return scores

graph LR
    A[音频网关] --> B(流式 ASR 服务)
    B --> C{流式翻译编排器}
    C --> D[流式 LLM 推理集群]
    C --> E[术语干预中间件]
    D --> F[流式后处理/格式化]
    F --> G[WebSocket 推送网关]
    G --> H[前端字幕渲染 SDK]

# 核心依赖
pip install fastapi uvicorn torch transformers accelerate websockets
# 量化加速 (可选)
pip install auto-gptq bitsandbytes

stream_translate/
├── main.py              # FastAPI 入口 + WebSocket 路由
├── core/
│   ├── stream_asr.py    # 流式 ASR 封装
│   ├── stream_llm.py    # 流式 LLM 推理 + Logits Processor
│   └── terminology.py   # 术语 Trie 树构建与匹配
├── schemas/
│   └── events.py        # 事件定义: Delta, Correction, TermHit
└── config.yaml          # 模型路径、Wait-k、术语库路径

# main.py 片段
@app.websocket("/ws/translate/{meeting_id}")
async def translate_ws(websocket: WebSocket, meeting_id: str):
    await websocket.accept()
    # 1. 初始化会话上下文 (加载术语库、预热 KV Cache)
    session = TranslationSession(meeting_id, terminology_manager)
    
    try:
        while True:
            # 2. 接收前端音频分片
            audio_chunk = await websocket.receive_bytes()
            
            # 3. 流式处理管线
            async for event in session.process_stream(audio_chunk):
                # 4. 实时推送事件
                await websocket.send_json(event.model_dump())
    except WebSocketDisconnect:
        session.cleanup()

const ws = new WebSocket('wss://api.yourdomain.com/ws/translate/meeting_123');
ws.onmessage = (e) => {
    const event = JSON.parse(e.data);
    if (event.type === 'delta') {
        appendSubtitle(event.text); // 追加模式
    } else if (event.type === 'correction') {
        replaceLastSubtitle(event.new_text); // 替换模式
    }
};
// 发送音频流 (需配合 MediaRecorder / WebAudio API)
mediaRecorder.ondataavailable = (e) => ws.send(e.data);

# core/policy_router.py
class LanguageAwarePolicyRouter:
    # 语种拓扑配置表 (建议外置配置中心动态下发)
    LANG_POLICY_MAP = {
        "zh-en": {"strategy": "FixedWaitK", "params": {"k": 4}},
        "zh-ja": {"strategy": "AdaptiveMILK", "params": {"max_k": 8, "entropy_thresh": 0.65}},
        "zh-de": {"strategy": "VerbAnchoredWait", "params": {"verb_pos_tags": ["VV", "VAF", "VMF"]}},
        "zh-ar": {"strategy": "VerbAnchoredWait", "params": {"verb_pos_tags": ["V"]}},
        # 低资源语言回退策略
        "default": {"strategy": "FixedWaitK", "params": {"k": 6}}
    }

    def __init__(self, tokenizer, pos_tagger):
        self.tokenizer = tokenizer
        self.pos_tagger = pos_tagger  # 轻量级 POS Tagger (如 fasttext + CRF)

    def get_stream_processor(self, src_lang: str, tgt_lang: str) -> StreamProcessor:
        key = f"{src_lang}-{tgt_lang}"
        policy_conf = self.LANG_POLICY_MAP.get(key, self.LANG_POLICY_MAP["default"])
        
        # 动态实例化策略处理器
        strategy_cls = STRATEGY_REGISTRY[policy_conf["strategy"]]
        return strategy_cls(tokenizer=self.tokenizer, **policy_conf["params"], pos_tagger=self.pos_tagger)

# experiment/config.yaml
experiment_id: "stream_v2.1_adaptive_waitk"
traffic_allocation: 0.10  # 10% 灰度
layers:
  - name: "strategy_layer"
    buckets: 100
    variants:
      - name: "control_fixed_k4"
        weight: 50
        config: {strategy: "FixedWaitK", k: 4}
      - name: "treatment_milk"
        weight: 50
        config: {strategy: "AdaptiveMILK", max_k: 8}
guardrail_metrics:
  - metric: "flicker_rate"
    threshold: 0.05  # 闪烁率不得劣化 5%
  - metric: "p99_latency"
    threshold: 2000  # ms
success_metrics:
  - metric: "stream_comet"
    min_lift: 0.01
  - metric: "term_acc"
    min_lift: 0.005

# scheduler/priority_scheduler.py
class MeetingAwareScheduler:
    def __init__(self, max_num_seqs: int, max_num_batched_tokens: int):
        self.waiting_queue = PriorityQueue()  # (priority, arrival_time, seq_group)
        self.running: List[SequenceGroup] = []
        self.max_tokens = max_num_batched_tokens

    def _calculate_priority(self, seq_group: SequenceGroup) -> int:
        # 业务优先级: VIP会议(0) > 普通会议(1) > 直播字幕(2) > 离线任务(3)
        biz_prio = seq_group.metadata.get("business_priority", 1)
        # 延迟敏感度: 实时同传(0) > 会议记录(1)
        lat_prio = 0 if seq_group.metadata.get("realtime") else 1
        # 公平性: 等待时间越久优先级越高 (防止饥饿)
        wait_bonus = min(int((time.time() - seq_group.arrival_time) / 10), 5)
        return (biz_prio * 100 + lat_prio * 10 - wait_bonus)

    def schedule(self) -> Tuple[List[SequenceGroup], List[SequenceGroup]]:
        # 1. 抢占检查: 高优先级新请求到来，驱逐低优先级 running seq (发送中断信号)
        # 2. 选取新 batch: 在 token budget 内，按优先级填充
        # 3. 处理 Preemption: 被驱逐的 seq 保留 KV Cache (Swap 到 CPU/磁盘)，标记为 PREEMPTED
        pass