✍️ 필사 모드: LLM 서빙: Speculative Decoding 프로덕션 벤치마크 2026

GPU: 4x NVIDIA A100 80GB SXM (NVLink 연결)
CPU: AMD EPYC 7763 64-Core
RAM: 512GB DDR4
OS: Ubuntu 22.04 LTS
CUDA: 12.4
Driver: 550.90.07

vLLM: 0.7.3
PyTorch: 2.5.1
Transformers: 4.47.0
Python: 3.11.10

# 프롬프트셋 구성 (500개)
prompt_distribution = {
    "short_qa": 150,          # 1-2문장 질문, 예상 출력 50-100 tokens
    "summarization": 100,      # 500-1000 단어 문서 요약, 예상 출력 150-300 tokens
    "code_generation": 80,     # 함수/클래스 생성, 예상 출력 100-500 tokens
    "creative_writing": 50,    # 스토리/에세이, 예상 출력 300-800 tokens
    "translation": 70,         # 한영/영한 번역, 예상 출력 100-300 tokens
    "structured_output": 50,   # JSON/YAML 생성, 예상 출력 50-200 tokens
}

import json
import time
import statistics
from dataclasses import dataclass, asdict
from openai import OpenAI

@dataclass
class BenchmarkResult:
    method: str
    model: str
    draft_model: str
    num_spec_tokens: int
    prompt_category: str
    ttft_ms: float
    tpot_ms: float
    e2e_ms: float
    output_tokens: int
    accept_ratio: float
    gpu_memory_gb: float

def run_single_benchmark(
    client: OpenAI,
    model: str,
    prompt: str,
    max_tokens: int,
) -> dict:
    """단일 프롬프트에 대한 벤치마크 실행"""
    start = time.perf_counter()
    first_token_time = None
    token_count = 0

    stream = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
        temperature=0.0,  # greedy decoding으로 통일
        stream=True,
    )

    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            if first_token_time is None:
                first_token_time = time.perf_counter()
            token_count += 1

    end = time.perf_counter()

    ttft = (first_token_time - start) if first_token_time else (end - start)
    e2e = end - start
    tpot = (end - first_token_time) / max(token_count - 1, 1) if first_token_time else e2e

    return {
        "ttft_ms": round(ttft * 1000, 2),
        "tpot_ms": round(tpot * 1000, 2),
        "e2e_ms": round(e2e * 1000, 2),
        "output_tokens": token_count,
    }

def run_full_benchmark(
    base_url: str,
    model: str,
    prompts: list[dict],
    warmup_runs: int = 5,
    measure_runs: int = 3,
) -> list[dict]:
    """전체 프롬프트셋에 대한 벤치마크 실행"""
    client = OpenAI(base_url=f"{base_url}/v1", api_key="benchmark")

    # Warmup
    for i in range(warmup_runs):
        run_single_benchmark(client, model, prompts[0]["text"], 64)

    results = []
    for run_idx in range(measure_runs):
        for prompt in prompts:
            result = run_single_benchmark(
                client, model, prompt["text"], prompt["max_tokens"]
            )
            result["category"] = prompt["category"]
            result["run"] = run_idx
            results.append(result)

    return results

# Temperature별 accept ratio 측정 결과 (EAGLE-3, Llama 3.1 70B)
temperature_results = {
    0.0: {"accept_ratio": 0.81, "speedup": 2.89},
    0.3: {"accept_ratio": 0.76, "speedup": 2.61},
    0.5: {"accept_ratio": 0.71, "speedup": 2.38},
    0.7: {"accept_ratio": 0.64, "speedup": 2.08},
    1.0: {"accept_ratio": 0.55, "speedup": 1.72},
    1.2: {"accept_ratio": 0.47, "speedup": 1.41},
    1.5: {"accept_ratio": 0.38, "speedup": 1.15},  # 거의 효과 없음
}

# 결론: temperature > 1.0이면 speculative decoding 비활성화 권장
TEMP_THRESHOLD = 1.0

# 동시 요청별 벤치마크 실행 코드
import asyncio
import aiohttp
import time

async def concurrent_benchmark(
    base_url: str,
    model: str,
    prompts: list[dict],
    concurrency: int,
) -> dict:
    """동시 요청 수별 벤치마크"""
    semaphore = asyncio.Semaphore(concurrency)
    results = []

    async def single_request(session, prompt):
        async with semaphore:
            start = time.perf_counter()
            payload = {
                "model": model,
                "messages": [{"role": "user", "content": prompt["text"]}],
                "max_tokens": prompt["max_tokens"],
                "temperature": 0.0,
                "stream": False,
            }
            async with session.post(
                f"{base_url}/v1/chat/completions",
                json=payload,
            ) as resp:
                data = await resp.json()
                end = time.perf_counter()
                tokens = data["usage"]["completion_tokens"]
                return {
                    "e2e_ms": (end - start) * 1000,
                    "output_tokens": tokens,
                    "tokens_per_sec": tokens / (end - start),
                }

    async with aiohttp.ClientSession() as session:
        tasks = [single_request(session, p) for p in prompts]
        total_start = time.perf_counter()
        results = await asyncio.gather(*tasks)
        total_elapsed = time.perf_counter() - total_start

    total_tokens = sum(r["output_tokens"] for r in results)
    return {
        "concurrency": concurrency,
        "total_tokens": total_tokens,
        "total_elapsed_sec": round(total_elapsed, 2),
        "throughput_tokens_per_sec": round(total_tokens / total_elapsed, 1),
        "avg_e2e_ms": round(statistics.mean(r["e2e_ms"] for r in results), 1),
        "p95_e2e_ms": round(
            sorted(r["e2e_ms"] for r in results)[int(len(results) * 0.95)], 1
        ),
    }

#!/bin/bash
# run_benchmark.sh - 전체 벤치마크 실행 스크립트
set -euo pipefail

MODEL="meta-llama/Llama-3.1-70B-Instruct"
DRAFT_MODELS=(
    "none"                                    # baseline
    "meta-llama/Llama-3.1-8B-Instruct"       # vanilla SD
    "eagle3-llama3.1-70b-instruct"            # EAGLE-3
)
SPEC_TOKENS=(0 5 5)
METHODS=("baseline" "vanilla" "eagle3")

RESULTS_DIR="benchmark_results/$(date +%Y%m%d_%H%M%S)"
mkdir -p "$RESULTS_DIR"

for i in "${!METHODS[@]}"; do
    method="${METHODS[$i]}"
    draft="${DRAFT_MODELS[$i]}"
    n_tokens="${SPEC_TOKENS[$i]}"

    echo "=== Running benchmark: $method ==="

    # 서버 시작
    if [ "$method" = "baseline" ]; then
        python -m vllm.entrypoints.openai.api_server \
            --model "$MODEL" \
            --tensor-parallel-size 4 \
            --gpu-memory-utilization 0.92 \
            --port 8000 &
    elif [ "$method" = "eagle3" ]; then
        python -m vllm.entrypoints.openai.api_server \
            --model "$MODEL" \
            --speculative-model "$draft" \
            --speculative-method eagle \
            --num-speculative-tokens "$n_tokens" \
            --tensor-parallel-size 4 \
            --gpu-memory-utilization 0.92 \
            --port 8000 &
    else
        python -m vllm.entrypoints.openai.api_server \
            --model "$MODEL" \
            --speculative-model "$draft" \
            --num-speculative-tokens "$n_tokens" \
            --tensor-parallel-size 4 \
            --gpu-memory-utilization 0.92 \
            --port 8000 &
    fi

    SERVER_PID=$!
    sleep 60  # 모델 로딩 대기

    # 벤치마크 실행
    python benchmark.py \
        --prompts eval_prompts.json \
        --model "$MODEL" \
        --warmup 10 \
        --runs 3 \
        --output "$RESULTS_DIR/${method}.json"

    # 서버 종료
    kill $SERVER_PID
    wait $SERVER_PID 2>/dev/null || true
    sleep 10
done

# 결과 집계
python aggregate_results.py --input-dir "$RESULTS_DIR" --output "$RESULTS_DIR/summary.json"
echo "Results saved to $RESULTS_DIR/summary.json"

# aggregate_results.py
import json
import sys
from pathlib import Path

def aggregate_results(results_dir: str) -> dict:
    """벤치마크 결과 파일들을 읽어 비교 테이블 생성"""
    results_path = Path(results_dir)
    summary = {}

    for result_file in sorted(results_path.glob("*.json")):
        if result_file.name == "summary.json":
            continue
        method = result_file.stem
        data = json.load(open(result_file))

        # P50, P95 계산
        tpot_values = [r["tpot_ms"] for r in data]
        e2e_values = [r["e2e_ms"] for r in data]

        summary[method] = {
            "tpot_p50": round(sorted(tpot_values)[len(tpot_values) // 2], 2),
            "tpot_p95": round(sorted(tpot_values)[int(len(tpot_values) * 0.95)], 2),
            "e2e_p50": round(sorted(e2e_values)[len(e2e_values) // 2], 2),
            "e2e_p95": round(sorted(e2e_values)[int(len(e2e_values) * 0.95)], 2),
            "total_requests": len(data),
        }

    # Speedup 계산 (baseline 대비)
    if "baseline" in summary:
        baseline_e2e = summary["baseline"]["e2e_p50"]
        for method in summary:
            summary[method]["speedup"] = round(
                baseline_e2e / summary[method]["e2e_p50"], 2
            )

    return summary

if __name__ == "__main__":
    results_dir = sys.argv[1] if len(sys.argv) > 1 else "benchmark_results/latest"
    summary = aggregate_results(results_dir)
    print(json.dumps(summary, indent=2))

Q: Target 모델을 자주 교체하는가?
├─ Yes: Vanilla SD (재학습 불필요)
│       또는 Prompt Lookup (학습 불필요)
└─ No: Q: GPU 메모리 여유가 있는가?
       ├─ Yes (>10GB 여유): EAGLE-3 (최고 성능)
       └─ No:  Q: 경량 학습 인프라가 있는가?
              ├─ Yes: Medusa-2 (메모리 효율적)
              └─ No:  Vanilla SD (Llama 3.1 8B as draft)