LLM Speculative Decoding 서빙 최적화 플레이북

import time
import json
import statistics
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")

def measure_baseline(
    model: str,
    prompts_file: str,
    num_runs: int = 3,
) -> dict:
    """현재 서빙의 latency/throughput 베이스라인 측정"""
    prompts = json.load(open(prompts_file))

    all_ttft = []    # Time To First Token
    all_tpot = []    # Time Per Output Token
    all_e2e = []     # End-to-End latency
    total_tokens = 0

    for run in range(num_runs):
        for prompt in prompts:
            start = time.perf_counter()
            first_token_time = None
            token_count = 0

            stream = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt["text"]}],
                max_tokens=prompt.get("max_tokens", 256),
                temperature=0.0,
                stream=True,
            )
            for chunk in stream:
                if chunk.choices[0].delta.content:
                    if first_token_time is None:
                        first_token_time = time.perf_counter()
                    token_count += 1

            end = time.perf_counter()

            all_ttft.append(first_token_time - start)
            all_e2e.append(end - start)
            if token_count > 1:
                all_tpot.append((end - first_token_time) / (token_count - 1))
            total_tokens += token_count

    return {
        "ttft_p50_ms": round(statistics.median(all_ttft) * 1000, 1),
        "ttft_p95_ms": round(sorted(all_ttft)[int(len(all_ttft) * 0.95)] * 1000, 1),
        "tpot_p50_ms": round(statistics.median(all_tpot) * 1000, 1),
        "tpot_p95_ms": round(sorted(all_tpot)[int(len(all_tpot) * 0.95)] * 1000, 1),
        "e2e_p50_ms": round(statistics.median(all_e2e) * 1000, 1),
        "e2e_p95_ms": round(sorted(all_e2e)[int(len(all_e2e) * 0.95)] * 1000, 1),
        "total_tokens": total_tokens,
        "avg_tokens_per_sec": round(total_tokens / sum(all_e2e), 1),
    }

# 사용 예
baseline = measure_baseline("meta-llama/Llama-3.1-70B-Instruct", "eval_prompts.json")
json.dump(baseline, open("baseline_metrics.json", "w"), indent=2)
print(json.dumps(baseline, indent=2))

Target 모델 -> Draft 모델 매칭 가이드

Llama 3.1 70B  -> Llama 3.1 8B (같은 family, 어휘 동일)
                  또는 EAGLE-3 draft head (학습 필요, 최고 성능)

Mistral Large  -> Mistral 7B (같은 tokenizer)

Qwen 2.5 72B  -> Qwen 2.5 1.5B 또는 Qwen 2.5 7B

자체 학습 모델  -> 같은 tokenizer의 소형 모델
                  또는 Medusa head / EAGLE head 학습

from transformers import AutoTokenizer

def verify_draft_compatibility(target_model: str, draft_model: str) -> dict:
    """Draft/Target 모델 간 tokenizer 호환성 검증"""
    target_tok = AutoTokenizer.from_pretrained(target_model)
    draft_tok = AutoTokenizer.from_pretrained(draft_model)

    # 1. 어휘 크기 일치 확인
    vocab_match = target_tok.vocab_size == draft_tok.vocab_size

    # 2. 특수 토큰 일치 확인
    special_match = (
        target_tok.bos_token_id == draft_tok.bos_token_id and
        target_tok.eos_token_id == draft_tok.eos_token_id and
        target_tok.pad_token_id == draft_tok.pad_token_id
    )

    # 3. 샘플 텍스트 인코딩 결과 비교
    test_texts = [
        "Hello, how are you?",
        "서울의 날씨는 어떤가요?",
        "def fibonacci(n): return n if n < 2 else fibonacci(n-1) + fibonacci(n-2)",
    ]
    encoding_match = all(
        target_tok.encode(t) == draft_tok.encode(t) for t in test_texts
    )

    return {
        "vocab_size_match": vocab_match,
        "special_tokens_match": special_match,
        "encoding_match": encoding_match,
        "compatible": vocab_match and special_match and encoding_match,
        "target_vocab_size": target_tok.vocab_size,
        "draft_vocab_size": draft_tok.vocab_size,
    }

result = verify_draft_compatibility(
    "meta-llama/Llama-3.1-70B-Instruct",
    "meta-llama/Llama-3.1-8B-Instruct"
)
print(result)
# {'vocab_size_match': True, 'special_tokens_match': True,
#  'encoding_match': True, 'compatible': True, ...}

# vLLM에서 speculative decoding 활성화 (Llama 3.1 70B + 8B)
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-70B-Instruct \
    --speculative-model meta-llama/Llama-3.1-8B-Instruct \
    --num-speculative-tokens 5 \
    --speculative-disable-mqa-scorer \
    --tensor-parallel-size 4 \
    --gpu-memory-utilization 0.92 \
    --max-model-len 4096 \
    --port 8000

# EAGLE-3 draft head 사용 (더 높은 accept ratio)
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3.1-70B-Instruct \
    --speculative-model eagle3-llama3.1-70b-instruct \
    --speculative-method eagle \
    --num-speculative-tokens 5 \
    --tensor-parallel-size 4 \
    --gpu-memory-utilization 0.92 \
    --max-model-len 4096 \
    --use-v2-block-manager \
    --port 8000

# speculative_decoding_config.yaml
# 이 설정을 기반으로 실험하고, accept ratio와 latency를 보면서 조정

# Draft 토큰 수: 너무 많으면 reject 증가, 너무 적으면 이점 감소
num_speculative_tokens: 5 # 시작값. 3-7 범위에서 실험

# Speculative decoding이 효과 없는 경우 자동 비활성화
speculative_disable_by_batch_size: 8 # 배치 8 이상이면 비활성화

# temperature > 0일 때의 처리
# typical acceptance sampling 사용 (품질 유지)
speculative_draft_tensor_parallel_size: 1 # draft는 TP=1로 충분

import subprocess
import json

def find_optimal_spec_tokens(
    target_model: str,
    draft_model: str,
    eval_prompts: str,
    candidates: list[int] = [3, 4, 5, 6, 7, 8],
) -> dict:
    """다양한 num_speculative_tokens 값으로 벤치마크 실행"""
    results = {}
    for n in candidates:
        print(f"Testing num_speculative_tokens={n}")
        # 서버 시작 (실제로는 subprocess로 관리)
        # 여기서는 결과 수집 로직만 표시
        metrics = run_benchmark(target_model, draft_model, n, eval_prompts)
        results[n] = {
            "accept_ratio": metrics["accept_ratio"],
            "tpot_p50_ms": metrics["tpot_p50_ms"],
            "e2e_speedup": metrics["baseline_e2e"] / metrics["e2e_p50_ms"],
            "gpu_memory_gb": metrics["gpu_memory_gb"],
        }

    # 최적값 선택: speedup이 가장 높은 값
    best_n = max(results, key=lambda n: results[n]["e2e_speedup"])
    results["recommended"] = best_n
    return results

# prometheus.yml - vLLM 메트릭 스크래핑 설정
scrape_configs:
  - job_name: 'vllm-speculative'
    scrape_interval: 15s
    static_configs:
      - targets: ['vllm-server:8000']
    metrics_path: /metrics

# draft 토큰 수용률
vllm:spec_decode_draft_acceptance_rate

# 위치별 수용률 (position 0이 가장 높고 뒤로 갈수록 떨어짐)
vllm:spec_decode_per_position_acceptance_rate{position="0"}
vllm:spec_decode_per_position_acceptance_rate{position="1"}

# 평균 수용 길이
vllm:spec_decode_mean_accepted_length

# grafana_alerts.yaml
groups:
  - name: speculative_decoding_alerts
    rules:
      # Accept ratio가 0.5 이하로 떨어지면 경고
      - alert: LowAcceptRatio
        expr: vllm:spec_decode_draft_acceptance_rate < 0.5
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: 'Speculative decoding accept ratio 하락'
          description: |
            Accept ratio가 {{ $value | printf "%.2f" }}로 하락했습니다.
            0.5 미만이면 speculative decoding이 오히려 overhead가 됩니다.
            Draft 모델 교체 또는 speculative decoding 비활성화를 검토하세요.

      # Accept ratio가 0.3 이하면 즉시 비활성화 권고
      - alert: CriticalAcceptRatio
        expr: vllm:spec_decode_draft_acceptance_rate < 0.3
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: 'Speculative decoding 비활성화 필요'
          description: |
            Accept ratio {{ $value | printf "%.2f" }}. 즉시 fallback 디코딩으로 전환하세요.

# /etc/nginx/conf.d/llm-router.conf
upstream vllm_speculative {
    server 10.0.1.10:8000;  # speculative decoding 활성화 서버
}

upstream vllm_standard {
    server 10.0.1.20:8000;  # 표준 디코딩 서버
}

# Lua 기반 동적 라우팅
server {
    listen 80;

    location /v1/chat/completions {
        access_by_lua_block {
            local cjson = require "cjson"
            ngx.req.read_body()
            local body = cjson.decode(ngx.req.get_body_data())

            -- 라우팅 조건 판단
            local use_speculative = true

            -- temperature가 높으면 표준 디코딩
            if body.temperature and body.temperature > 1.0 then
                use_speculative = false
            end

            -- max_tokens가 짧으면 표준 디코딩
            if body.max_tokens and body.max_tokens < 32 then
                use_speculative = false
            end

            -- stream=false이고 짧은 응답이면 표준 디코딩
            if not body.stream and body.max_tokens and body.max_tokens < 64 then
                use_speculative = false
            end

            if use_speculative then
                ngx.var.upstream = "vllm_speculative"
            else
                ngx.var.upstream = "vllm_standard"
            end
        }

        proxy_pass http://$upstream;
        proxy_set_header Host $host;
    }
}

import requests
import time
from dataclasses import dataclass

@dataclass
class FallbackConfig:
    accept_ratio_threshold: float = 0.4
    latency_regression_pct: float = 20.0  # baseline 대비 20% 이상 느려지면
    check_interval_sec: int = 60
    consecutive_failures: int = 3

class SpeculativeDecodingGuard:
    """Speculative decoding 상태를 모니터링하고 자동 fallback 결정"""

    def __init__(self, config: FallbackConfig, prometheus_url: str):
        self.config = config
        self.prometheus_url = prometheus_url
        self.failure_count = 0

    def query_prometheus(self, query: str) -> float:
        resp = requests.get(
            f"{self.prometheus_url}/api/v1/query",
            params={"query": query},
        )
        result = resp.json()["data"]["result"]
        return float(result[0]["value"][1]) if result else 0.0

    def should_fallback(self) -> tuple[bool, str]:
        # 1. Accept ratio 확인
        accept_ratio = self.query_prometheus(
            'vllm:spec_decode_draft_acceptance_rate'
        )
        if accept_ratio < self.config.accept_ratio_threshold:
            self.failure_count += 1
            if self.failure_count >= self.config.consecutive_failures:
                return True, f"accept_ratio={accept_ratio:.2f} < {self.config.accept_ratio_threshold}"
        else:
            self.failure_count = 0

        # 2. Latency regression 확인
        current_p95 = self.query_prometheus(
            'histogram_quantile(0.95, rate(vllm:e2e_request_latency_seconds_bucket[5m]))'
        )
        baseline_p95 = self.query_prometheus(
            'vllm:baseline_e2e_p95_seconds'  # 베이스라인 메트릭 별도 기록 필요
        )
        if baseline_p95 > 0:
            regression_pct = ((current_p95 - baseline_p95) / baseline_p95) * 100
            if regression_pct > self.config.latency_regression_pct:
                return True, f"latency regression {regression_pct:.1f}% > {self.config.latency_regression_pct}%"

        return False, "healthy"

    def run(self):
        while True:
            should_fb, reason = self.should_fallback()
            if should_fb:
                print(f"[FALLBACK] Speculative decoding 비활성화: {reason}")
                self.trigger_fallback()
            time.sleep(self.config.check_interval_sec)

    def trigger_fallback(self):
        """표준 디코딩 서버로 트래픽 전환"""
        # 실제 구현: 로드밸런서 가중치 변경 또는 feature flag 토글
        requests.post(
            "http://config-server/api/v1/flags",
            json={"speculative_decoding_enabled": False},
        )

import json
import datetime
from typing import Any

def weekly_speculative_decoding_report(
    prometheus_url: str,
    baseline_file: str,
) -> dict[str, Any]:
    """주간 speculative decoding 운영 리포트 생성"""
    baseline = json.load(open(baseline_file))

    report = {
        "report_date": datetime.date.today().isoformat(),
        "period": "last_7d",
    }

    # 1. Accept ratio 추이
    report["accept_ratio"] = {
        "current_avg": query_prom(prometheus_url,
            'avg_over_time(vllm:spec_decode_draft_acceptance_rate[7d])'),
        "min": query_prom(prometheus_url,
            'min_over_time(vllm:spec_decode_draft_acceptance_rate[7d])'),
        "max": query_prom(prometheus_url,
            'max_over_time(vllm:spec_decode_draft_acceptance_rate[7d])'),
    }

    # 2. Latency 개선율
    current_e2e_p50 = query_prom(prometheus_url,
        'histogram_quantile(0.5, rate(vllm:e2e_request_latency_seconds_bucket[7d]))')
    report["speedup"] = {
        "e2e_p50_speedup": round(baseline["e2e_p50_ms"] / (current_e2e_p50 * 1000), 2),
        "baseline_e2e_p50_ms": baseline["e2e_p50_ms"],
        "current_e2e_p50_ms": round(current_e2e_p50 * 1000, 1),
    }

    # 3. Fallback 발생 횟수
    report["fallback_count"] = int(query_prom(prometheus_url,
        'count_over_time(ALERTS{alertname="LowAcceptRatio"}[7d])'))

    # 4. 리소스 사용량 (speculative 추가분)
    report["gpu_memory_overhead_gb"] = query_prom(prometheus_url,
        'avg_over_time(vllm:gpu_cache_usage_perc[7d])') * 80  # A100 80GB 기준

    # 5. 권고 사항
    recommendations = []
    if report["accept_ratio"]["current_avg"] < 0.55:
        recommendations.append("Draft 모델 교체 또는 EAGLE-3 head 학습 권고")
    if report["speedup"]["e2e_p50_speedup"] < 1.3:
        recommendations.append("개선폭이 1.3x 미만. 비용 대비 효과 재검토 필요")
    if report["fallback_count"] > 5:
        recommendations.append(f"주간 fallback {report['fallback_count']}회. Draft 모델 품질 점검")

    report["recommendations"] = recommendations
    return report

# 1. Accept ratio 확인
curl -s http://localhost:8000/metrics | grep spec_decode_draft_acceptance_rate
# 0.3 미만이면 draft 모델 문제

# 2. Draft 모델 추론 시간 확인
curl -s http://localhost:8000/metrics | grep spec_decode_draft_latency
# Draft 추론이 target 단일 토큰 추론의 50% 이상이면 비효율

# 3. GPU 메모리 부족으로 인한 swap 확인
nvidia-smi --query-gpu=memory.used,memory.total --format=csv
# 95% 이상이면 KV cache 부족으로 speculative 이점 상쇄

RuntimeError: Draft model vocab size 32000 != target model vocab size 128256