推論最佳化攻擊

┌──────────────┐     候選 token          ┌──────────────┐
│  Draft 模型  │ ──────────────────────── │ 目標模型      │
│  (1-3B)      │  "The quick brown fox"   │ (70B+)        │
│  快速、便宜  │ ────────────────────────│ 慢、準確      │
└──────────────┘     接受／拒絕          └──────────────┘

# 推測解碼驗證
def speculative_verify(draft_tokens, target_model, prompt):
    """目標模型並行檢查 draft 預測。"""
    target_logits = target_model(prompt + draft_tokens)
 
    accepted = []
    for i, token in enumerate(draft_tokens):
        draft_prob = draft_model_prob(token, context=prompt + accepted)
        target_prob = target_logits[i].softmax(-1)[token].item()
 
        # 若目標模型同意則接受（修改之拒絕取樣）
        if random.random() < min(1, target_prob / draft_prob):
            accepted.append(token)
        else:
            # 自調整後之分布重新取樣
            corrected = resample(target_logits[i], draft_logits[i])
            accepted.append(corrected)
            break  # 拒絕剩餘 draft token
 
    return accepted

import time
import asyncio
 
async def probe_batch_occupancy(api_url: str) -> dict:
    """量測延遲變異以推論批次佔用率。
    高變異 = 具可變負載之動態批次。
    一致快速 = 低佔用。
    一致緩慢 = 高佔用。"""
 
    latencies = []
    for _ in range(50):
        start = time.perf_counter()
        await send_request(api_url, "Hello", max_tokens=1)
        latencies.append(time.perf_counter() - start)
        await asyncio.sleep(0.1)
 
    return {
        "mean": sum(latencies) / len(latencies),
        "variance": variance(latencies),
        "min": min(latencies),
        "max": max(latencies),
    }

# 於共享 GPU 系統上，監控對等 GPU 記憶體使用
# 以推論其他租戶之推論模式
import pynvml
pynvml.nvmlInit()
 
def monitor_peer_gpu_memory(gpu_indices: list, duration_s: float):
    """取樣 GPU 記憶體使用以偵測推論活動。"""
    samples = []
    start = time.time()
    while time.time() - start < duration_s:
        snapshot = {}
        for idx in gpu_indices:
            handle = pynvml.nvmlDeviceGetHandleByIndex(idx)
            info = pynvml.nvmlDeviceGetMemoryInfo(handle)
            snapshot[idx] = info.used
        samples.append(snapshot)
        time.sleep(0.01)
    return samples