vLLM: Production LLM Serving from Zero to Scale¶

# hello_vllm.py
from vllm import LLM, SamplingParams

# Load model — this downloads from HuggingFace on first run
# (set HF_TOKEN env var for gated models like LLaMA)
llm = LLM(model="Qwen/Qwen2.5-7B-Instruct")

# Sampling parameters
params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=256
)

# Generate — notice we pass a LIST of prompts (batching!)
prompts = [
    "What is PagedAttention in vLLM?",
    "Write a Python function to check if a number is prime.",
    "Explain the concept of entropy in information theory.",
]

outputs = llm.generate(prompts, params)

for output in outputs:
    prompt = output.prompt
    response = output.outputs[0].text
    print(f"Prompt: {prompt!r}")
    print(f"Response: {response}")
    print("─" * 60)

from vllm import SamplingParams

params = SamplingParams(
    # Core sampling
    temperature=0.7,      # 0 = deterministic, >1 = more random
    top_p=0.9,           # nucleus sampling: top tokens summing to 90% probability
    top_k=50,            # only consider top 50 tokens

    # Length control
    max_tokens=512,       # maximum tokens to generate
    min_tokens=10,        # minimum before EOS is allowed

    # Repetition control
    presence_penalty=0.1, # penalize tokens that already appeared
    frequency_penalty=0.1,# penalize tokens proportional to how often they appeared
    repetition_penalty=1.1,# multiplicative penalty (>1 reduces repetition)

    # Stopping conditions
    stop=["<|eot_id|>", "Human:", "User:"],  # stop on these strings
    stop_token_ids=[128009],                  # stop on these token IDs

    # Multiple outputs
    n=3,                  # generate 3 different completions
    best_of=5,           # generate 5 internally, return the 3 highest-scoring

    # Determinism
    seed=42,             # for reproducible outputs
)

# auth_proxy.py — simple FastAPI proxy with API key auth
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx

app = FastAPI()
security = HTTPBearer()

VALID_KEYS = {"sk-user1-key-here", "sk-user2-key-here"}
VLLM_BASE = "http://localhost:8000"

def verify_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if credentials.credentials not in VALID_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")

@app.api_route("/{path:path}", methods=["GET", "POST", "DELETE"])
async def proxy(request: Request, path: str, _=Depends(verify_key)):
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method=request.method,
            url=f"{VLLM_BASE}/{path}",
            content=await request.body(),
            headers={k: v for k, v in request.headers.items()
                     if k.lower() not in ("host", "authorization")},
            timeout=300.0,
        )
    return response.json()

from vllm import LLM, SamplingParams
from vllm.sampling_params import GuidedDecodingParams

llm = LLM(model="Qwen/Qwen2.5-7B-Instruct")

# Define the schema you want
product_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "price": {"type": "number"},
        "in_stock": {"type": "boolean"},
        "tags": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["name", "price", "in_stock"]
}

params = SamplingParams(
    temperature=0.1,
    max_tokens=200,
    guided_decoding=GuidedDecodingParams(json=product_schema)
)

output = llm.generate(
    ["Extract product info: 'The Nike Air Max 90 costs $150 and is available in stores. Tags: shoes, running, nike'"],
    params
)

import json
result = json.loads(output[0].outputs[0].text)
print(result)
# {"name": "Nike Air Max 90", "price": 150.0, "in_stock": true, "tags": ["shoes", "running", "nike"]}

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

class Product(BaseModel):
    name: str
    price: float
    in_stock: bool
    tags: list[str]

response = client.beta.chat.completions.parse(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[
        {"role": "user",
         "content": "Extract: 'The Nike Air Max 90 costs $150 and is available.'"}
    ],
    response_format=Product,
)

product = response.choices[0].message.parsed
print(f"{product.name}: ${product.price}")

from openai import OpenAI
import json

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3-8B-Instruct",
    messages=[{"role": "user", "content": "What's the weather in Bangkok?"}],
    tools=tools,
    tool_choice="auto"
)

# Check if the model called a tool
message = response.choices[0].message
if message.tool_calls:
    tool_call = message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    print(f"Model called: {function_name}({function_args})")
    # → Model called: get_weather({"city": "Bangkok", "unit": "celsius"})

# Use the customer-service LoRA adapter
response = client.chat.completions.create(
    model="customer-service",  # ← adapter name, not base model
    messages=[{"role": "user", "content": "How do I return a product?"}]
)

# Quick cost calculator for self-hosted vLLM
def estimate_monthly_cost(
    requests_per_day: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    gpu_cost_per_hour: float,  # e.g., $2.50/hr for A10G on AWS
    throughput_tokens_per_sec: float  # from your benchmark
) -> dict:

    tokens_per_day = requests_per_day * (avg_input_tokens + avg_output_tokens)
    tokens_per_month = tokens_per_day * 30

    # GPU hours needed
    seconds_needed_per_day = (requests_per_day * avg_output_tokens) / throughput_tokens_per_sec
    gpu_hours_per_day = seconds_needed_per_day / 3600
    gpu_hours_per_month = gpu_hours_per_day * 30

    # Assume 24/7 instance (always-on for availability)
    monthly_cost_always_on = 24 * 30 * gpu_cost_per_hour

    # Equivalent OpenAI cost (GPT-4o: $5/1M input, $15/1M output)
    openai_cost = (
        (tokens_per_month / 1_000_000 * 5)    # input
        + (tokens_per_month * avg_output_tokens / (avg_input_tokens + avg_output_tokens) / 1_000_000 * 15)  # output
    )

    return {
        "tokens_per_month": f"{tokens_per_month:,}",
        "gpu_hours_if_scaled": f"{gpu_hours_per_month:.0f} hours",
        "monthly_cost_always_on": f"${monthly_cost_always_on:.0f}",
        "openai_equivalent_cost": f"${openai_cost:.0f}",
        "savings_vs_openai": f"${openai_cost - monthly_cost_always_on:.0f}"
    }

# Example: 10,000 requests/day, 500 input tokens, 300 output tokens
# A10G at $2.50/hr, throughput 2,000 tokens/sec
result = estimate_monthly_cost(
    requests_per_day=10_000,
    avg_input_tokens=500,
    avg_output_tokens=300,
    gpu_cost_per_hour=2.50,
    throughput_tokens_per_sec=2000
)
for k, v in result.items():
    print(f"{k}: {v}")