Serving Open Models with vLLM (elective)

# script_id: day_093_vllm_serving/naive_parallel
"""One user vs eight users: the same 8 Orbit questions, sequential then concurrent.
Install: uv add openai
Env: ATLAS_LOCAL_BASE_URL (default Ollama), ATLAS_LOCAL_MODEL, ORBIT_DATA
Needs: Ollama running with the model pulled (Day 91). Calls no paid API."""
import asyncio, json, os, time
from pathlib import Path
from openai import AsyncOpenAI

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
BASE = os.environ.get("ATLAS_LOCAL_BASE_URL", "http://localhost:11434/v1")
LOCAL_MODEL = os.environ.get("ATLAS_LOCAL_MODEL", "qwen3:4b-instruct-2507-q4_K_M")
rows = (DATA_DIR / "eval" / "qa_gold.jsonl").read_text().splitlines()[:8]
QUESTIONS = [json.loads(r)["question"] + " Answer in one sentence." for r in rows]

async def timed(client: AsyncOpenAI, q: str) -> float:
    t0 = time.perf_counter()
    await client.responses.create(model=LOCAL_MODEL, input=q)
    return time.perf_counter() - t0

async def main() -> None:
    client = AsyncOpenAI(base_url=BASE, api_key="ollama")          # key required but ignored
    seq = [await timed(client, q) for q in QUESTIONS]              # one user at a time
    t0 = time.perf_counter()
    par = await asyncio.gather(*(timed(client, q) for q in QUESTIONS))  # eight users at once
    wall = time.perf_counter() - t0
    print(f"sequential: mean wait {sum(seq) / len(seq):5.1f}s, all 8 done in {sum(seq):5.1f}s")
    print(f"concurrent: mean wait {sum(par) / len(par):5.1f}s, all 8 done in {wall:5.1f}s")

asyncio.run(main())

sequential: mean wait   2.3s, all 8 done in  18.4s
concurrent: mean wait   9.8s, all 8 done in  16.9s
(illustrative; your numbers will differ)

# script_id: day_093_vllm_serving/batching_by_hand
"""Static vs continuous batching on toy numbers. No GPU, no server.
Install: nothing (stdlib). Env: none."""
import heapq

def static_batching(lengths: list[int], slots: int) -> tuple[int, int]:
    """Each round runs until its longest request ends; freed slots stay idle."""
    steps = busy = 0
    for i in range(0, len(lengths), slots):
        batch = lengths[i:i + slots]
        steps += max(batch)
        busy += sum(batch)
    return steps, busy

def continuous_batching(lengths: list[int], slots: int) -> tuple[int, int]:
    """A waiting request takes whichever slot frees up first."""
    free_at = [0] * slots                 # min-heap: step at which each slot is free
    for n in lengths:
        start = heapq.heappop(free_at)
        heapq.heappush(free_at, start + n)
    return max(free_at), sum(lengths)

for name, lengths in {"3 requests": [50, 200, 100],
                      "6 requests": [50, 200, 100, 150, 50, 100]}.items():
    for fn in (static_batching, continuous_batching):
        steps, busy = fn(lengths, slots=3)
        print(f"{name}  {fn.__name__:<20} steps={steps:<4} GPU busy={busy / (steps * 3):.0%}")

3 requests  static_batching      steps=200  GPU busy=58%
3 requests  continuous_batching  steps=200  GPU busy=58%
6 requests  static_batching      steps=350  GPU busy=62%
6 requests  continuous_batching  steps=250  GPU busy=87%

# Colab: Runtime → Change runtime type → T4 GPU. In a cell, prefix each line with "!".
# Install line per the vLLM quickstart (vLLM 0.33, as of 2026-11; ⚠ verify against the quickstart).
pip install -q uv openai
uv venv --python 3.12 --seed /content/vllm-env      # keeps vLLM's pinned torch out of Colab's own env
uv pip install --python /content/vllm-env/bin/python vllm --torch-backend=auto
# Start in the background and log to a file: a foreground server would block the notebook.
nohup /content/vllm-env/bin/vllm serve Qwen/Qwen3-0.6B --dtype half --max-model-len 4096 \
  > vllm.log 2>&1 &
tail -n 5 vllm.log    # re-run until the log shows "Application startup complete" (first start downloads weights)

# Mac / CPU: llama-server with the Q4_K_M GGUF you built on Day 92.
brew install llama.cpp
# 4 slots = up to 4 requests decoded together. The -c context is shared across slots (⚠ verify with llama-server --help).
llama-server -m "$ATLAS_GGUF" --port 9931 --parallel 4 -c 16384
export ATLAS_SERVE_BASE_URL=http://localhost:9931/v1   # llama-server's default port is 9931, not 8080

# script_id: day_093_vllm_serving/first_call
"""First call to a self-hosted OpenAI-compatible server (vLLM or llama-server).
Install: uv add openai
Env: ATLAS_SERVE_BASE_URL (default http://localhost:8000/v1, vLLM's default),
     VLLM_API_KEY (only if the server was started with --api-key), ATLAS_SERVE_MODEL (optional)
Needs: a running `vllm serve` or `llama-server`. Calls no paid API."""
import os
import time
from typing import Literal
from openai import OpenAI

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
LOCAL_BASE_URL = os.environ.get("ATLAS_LOCAL_BASE_URL", "http://localhost:11434/v1")
LOCAL_MODEL = os.environ.get("ATLAS_LOCAL_MODEL", "qwen3:4b-instruct-2507-q4_K_M")
SERVE_BASE_URL = os.environ.get("ATLAS_SERVE_BASE_URL", "http://localhost:8000/v1")

def make_client(backend: Literal["openai", "local", "serve"] = "openai") -> tuple[OpenAI, str]:
    """Day 91's make_client (compact) plus a "serve" backend. Returns (client, model id)."""
    if backend == "openai":
        return OpenAI(), MODEL
    if backend == "local":
        return OpenAI(base_url=LOCAL_BASE_URL, api_key="ollama"), LOCAL_MODEL
    client = OpenAI(base_url=SERVE_BASE_URL, api_key=os.environ.get("VLLM_API_KEY", "EMPTY"))
    # Ask the server what it serves instead of guessing the name.
    model = os.environ.get("ATLAS_SERVE_MODEL") or client.models.list().data[0].id
    return client, model

client, model = make_client("serve")
t0 = time.perf_counter()
resp = client.responses.create(
    model=model, max_output_tokens=512,
    input="What are the default values of lease_seconds and heartbeat_seconds in orbit.toml?")
out = resp.usage.output_tokens if resp.usage else "?"
print(f"served model: {model}  ({time.perf_counter() - t0:.2f}s, {out} output tokens)")
print(resp.output_text[:300])

served model: Qwen/Qwen3-0.6B  (1.84s, 287 output tokens)
<think>
Okay, the user is asking about orbit.toml ...
(illustrative; your numbers and text will differ)

# script_id: day_093_vllm_serving/model_not_found
"""Break it: request a model name the server does not serve, then list what it does serve.
Install: uv add openai · Env: ATLAS_SERVE_BASE_URL, VLLM_API_KEY · Needs: a running vLLM server."""
import os
import openai
from openai import OpenAI

client = OpenAI(base_url=os.environ.get("ATLAS_SERVE_BASE_URL", "http://localhost:8000/v1"),
                api_key=os.environ.get("VLLM_API_KEY", "EMPTY"), max_retries=0)
try:
    client.responses.create(model="qwen3-0.6b", input="hi")   # Ollama-style tag, not the HF repo id
except openai.NotFoundError as e:
    served = [m.id for m in client.models.list().data]
    print(f"{e.message}\nserver offers: {served}")

Error code: 404 - The model `qwen3-0.6b` does not exist.
server offers: ['Qwen/Qwen3-0.6B']
(message text paraphrased from vLLM; ⚠ verify the exact wording on your version)

# script_id: day_093_vllm_serving/load_test
"""Concurrency sweep against an OpenAI-compatible server: req/s, output tok/s, TTFT p50/p95, errors.
Install: uv add openai
Env: ATLAS_SERVE_BASE_URL (default http://localhost:8000/v1), VLLM_API_KEY (if the server uses one),
     ATLAS_SERVE_MODEL (optional; else the first id from /v1/models), ORBIT_DATA
Needs: a running vLLM (Colab GPU) or llama-server (Mac/CPU). Calls no paid API."""
import asyncio
import json
import math
import os
import time
from dataclasses import dataclass
from pathlib import Path
import openai
from openai import AsyncOpenAI

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
SERVE_BASE_URL = os.environ.get("ATLAS_SERVE_BASE_URL", "http://localhost:8000/v1")
API_KEY = os.environ.get("VLLM_API_KEY", "EMPTY")

@dataclass
class RequestStats:
    ok: bool
    ttft_s: float | None      # time to the first streamed delta (text or reasoning)
    total_s: float
    out_tokens: int
    error: str = ""

def load_prompts() -> list[str]:
    rows = (DATA_DIR / "eval" / "qa_gold.jsonl").read_text().splitlines()
    return [json.loads(r)["question"] + " Answer in two sentences." for r in rows if r.strip()]

def percentile(xs: list[float], p: float) -> float:
    """Nearest-rank percentile (Day 42's helper, compact). NaN when there is no data."""
    if not xs:
        return float("nan")
    s = sorted(xs)
    return s[max(0, math.ceil(p / 100 * len(s)) - 1)]

# script_id: day_093_vllm_serving/load_test
async def one_request(client: AsyncOpenAI, model: str, prompt: str,
                      max_tokens: int = 256, timeout_s: float = 60.0) -> RequestStats:
    t0 = time.perf_counter()
    ttft, deltas, usage_out = None, 0, None
    try:
        async with asyncio.timeout(timeout_s):         # caps the whole request, not one socket read
            stream = await client.responses.create(
                model=model, input=prompt, max_output_tokens=max_tokens, stream=True)
            async for ev in stream:
                if ev.type.endswith(".delta"):         # output_text.delta, or a reasoning delta
                    if ttft is None:
                        ttft = time.perf_counter() - t0
                    deltas += 1
                elif ev.type == "response.completed" and ev.response.usage:
                    usage_out = ev.response.usage.output_tokens
    except TimeoutError:
        return RequestStats(False, ttft, time.perf_counter() - t0, deltas, "timeout")
    except openai.APIError as e:                       # HTTP 4xx/5xx and connection errors
        return RequestStats(False, ttft, time.perf_counter() - t0, deltas, type(e).__name__)
    return RequestStats(True, ttft, time.perf_counter() - t0, usage_out or deltas)

# script_id: day_093_vllm_serving/load_test
@dataclass
class LoadReport:
    concurrency: int
    rps: float            # successful requests per second
    out_tok_s: float      # output tokens per second, summed over all requests
    ttft_p50: float
    ttft_p95: float
    errors: int
    n: int = 0

async def run_level(client: AsyncOpenAI, model: str, prompts: list[str], concurrency: int,
                    n_requests: int, timeout_s: float = 60.0) -> LoadReport:
    sem = asyncio.Semaphore(concurrency)              # at most `concurrency` requests in flight

    async def worker(i: int) -> RequestStats:
        async with sem:
            return await one_request(client, model, prompts[i % len(prompts)], timeout_s=timeout_s)

    t0 = time.perf_counter()
    stats = await asyncio.gather(*(worker(i) for i in range(n_requests)))
    wall = time.perf_counter() - t0
    ok = [s for s in stats if s.ok]
    ttfts = [s.ttft_s for s in ok if s.ttft_s is not None]
    return LoadReport(concurrency, len(ok) / wall, sum(s.out_tokens for s in ok) / wall,
                      percentile(ttfts, 50), percentile(ttfts, 95), len(stats) - len(ok), len(stats))

def load_test(base_url: str, model: str, concurrency: int, n_requests: int,
              timeout_s: float = 60.0) -> LoadReport:
    async def _go() -> LoadReport:
        async with AsyncOpenAI(base_url=base_url, api_key=API_KEY, max_retries=0) as client:
            return await run_level(client, model, load_prompts(), concurrency, n_requests, timeout_s)
    return asyncio.run(_go())

# script_id: day_093_vllm_serving/load_test
def served_model(base_url: str) -> str:
    if name := os.environ.get("ATLAS_SERVE_MODEL"):
        return name
    return openai.OpenAI(base_url=base_url, api_key=API_KEY).models.list().data[0].id

def sweep(model: str, levels: tuple[int, ...] = (1, 4, 16, 32), per_level: int = 4) -> list[LoadReport]:
    load_test(SERVE_BASE_URL, model, concurrency=1, n_requests=2)        # warm-up, discarded
    reports = []
    for c in levels:
        r = load_test(SERVE_BASE_URL, model, concurrency=c, n_requests=max(16, per_level * c))
        print(f"c={c:>3}  {r.rps:6.2f} req/s  {r.out_tok_s:7.0f} out tok/s  "
              f"TTFT p50 {r.ttft_p50:5.2f}s p95 {r.ttft_p95:5.2f}s  errors {r.errors}/{r.n}")
        reports.append(r)
    return reports

def write_report(reports: list[LoadReport], model: str, path: Path = Path("reports/load_test.md")) -> Path:
    lines = [f"# Load test: `{model}` at {SERVE_BASE_URL}", "",
             "| concurrency | req/s | out tok/s | TTFT p50 (s) | TTFT p95 (s) | errors |",
             "|---|---|---|---|---|---|"]
    lines += [f"| {r.concurrency} | {r.rps:.2f} | {r.out_tok_s:.0f} | {r.ttft_p50:.2f} "
              f"| {r.ttft_p95:.2f} | {r.errors}/{r.n} |" for r in reports]
    best = max(reports, key=lambda r: r.out_tok_s)
    lines += ["", f"Peak: {best.out_tok_s:.0f} output tok/s at concurrency {best.concurrency}."]
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n")
    return path

if __name__ == "__main__":
    model = served_model(SERVE_BASE_URL)
    print("wrote", write_report(sweep(model), model))

c=  1    0.62 req/s       98 out tok/s  TTFT p50  0.03s p95  0.04s  errors 0/16
c=  4    2.31 req/s      371 out tok/s  TTFT p50  0.04s p95  0.06s  errors 0/16
c= 16    7.90 req/s     1262 out tok/s  TTFT p50  0.07s p95  0.12s  errors 0/64
c= 32   10.40 req/s     1655 out tok/s  TTFT p50  0.15s p95  0.31s  errors 0/128
wrote reports/load_test.md
(illustrative shape for a 0.6B model on a T4; your numbers will differ)

# script_id: day_093_vllm_serving/capacity_plan
"""From one load-test row to a fleet size and an average utilization. Stdlib only.
Install: nothing. Env: none. Replace the sample values with rows from reports/load_test.md."""
import math

LEVEL_RPS = 7.90   # req/s at the highest concurrency that met the SLO (sample row: c=16)
PEAK_RPS = 2.0     # Orbit's rota at its busiest (an assumption; measure it from your logs)
AVG_RPS = 0.3      # average over 24 hours, nights and weekends included
HEADROOM = 0.7     # plan to run each replica at <=70% of measured capacity

def replicas_needed(peak_rps: float, rps_per_replica: float, headroom: float = HEADROOM) -> int:
    return max(1, math.ceil(peak_rps / (rps_per_replica * headroom)))

n = replicas_needed(PEAK_RPS, LEVEL_RPS)
utilization = AVG_RPS / (n * LEVEL_RPS)
print(f"replicas for peak: {n}")
print(f"average utilization: {utilization:.0%} of the capacity you pay for")

replicas for peak: 1
average utilization: 4% of the capacity you pay for

# script_id: day_093_vllm_serving/cost_compare
"""Self-hosted $/1M output tokens vs an API price, and the break-even daily volume. Stdlib only.
Install: nothing. Env: ATLAS_GPU_USD_PER_HR (your provider's on-demand price),
     ATLAS_OUT_TOK_S (output tok/s from reports/load_test.md at your SLO level)."""
import os

def cost_per_million(gpu_usd_per_hr: float, out_tok_s: float, utilization: float = 1.0) -> float:
    """$ per 1M output tokens; utilization = share of paid GPU hours doing useful work."""
    return gpu_usd_per_hr / (out_tok_s * 3600 * utilization) * 1_000_000

def breakeven_tokens_per_day(gpu_usd_per_hr: float, api_usd_per_m: float, hours: float = 24) -> float:
    """Daily output tokens above which an always-on GPU is cheaper than the API."""
    return gpu_usd_per_hr * hours / api_usd_per_m * 1_000_000

# PLACEHOLDER inputs: set both env vars to your own numbers before deciding anything.
GPU_USD_PER_HR = float(os.environ.get("ATLAS_GPU_USD_PER_HR", "1.00"))
OUT_TOK_S = float(os.environ.get("ATLAS_OUT_TOK_S", "1262"))     # sample report, c=16
API_OUT_USD_PER_M = 0.50   # gpt-6-luna output, as of 2026-11; verify at the pricing page

if __name__ == "__main__":
    for u in (0.04, 0.50, 1.00):
        usd = cost_per_million(GPU_USD_PER_HR, OUT_TOK_S, u)
        print(f"self-host at {u:>4.0%} utilization: ${usd:.3f} per 1M output tokens")
    print(f"API (gpt-6-luna):              ${API_OUT_USD_PER_M:.3f} per 1M output tokens")
    be = breakeven_tokens_per_day(GPU_USD_PER_HR, API_OUT_USD_PER_M)
    cap = OUT_TOK_S * 86_400
    print(f"break-even: {be / 1e6:.0f}M output tokens/day = {be / cap:.0%} of this GPU's "
          f"daily capacity ({cap / 1e6:.1f}M)")

self-host at   4% utilization: $5.503 per 1M output tokens
self-host at  50% utilization: $0.440 per 1M output tokens
self-host at 100% utilization: $0.220 per 1M output tokens
API (gpt-6-luna):              $0.500 per 1M output tokens
break-even: 48M output tokens/day = 44% of this GPU's daily capacity (109.0M)