1import hashlib
2import numpy as np
3from openai import OpenAI
4from redis import Redis
5import json
6 
7client = OpenAI()
8redis = Redis(host="localhost", port=6379)
9 
10# Level 1: Semantic caching
11# Cache LLM responses by semantic similarity of queries
12# 20-40% of production queries are near-duplicates — serve them from cache
13 
SIMILARITY_THRESHOLD=0.92: Tune higher to avoid wrong cache hits, lower to increase hit rate
14SIMILARITY_THRESHOLD = 0.92  # Cosine similarity threshold for cache hit
15 
16def get_embedding(text: str) -> list[float]:
17    return client.embeddings.create(
18        input=text, model="text-embedding-3-small"
19    ).data[0].embedding
20 
21def semantic_cache_lookup(query: str) -> str | None:
22    query_embedding = get_embedding(query)
23 
Cache lookup scales O(n) with cache size — use vector DB (Qdrant, Pinecone) for >10K cached queries
24    # Search for similar cached queries
25    cached_keys = redis.keys("embed:*")
26    for key in cached_keys:
27        cached_data = json.loads(redis.get(key))
28        cached_embedding = cached_data["embedding"]
29 
Cosine similarity: values >0.92 indicate semantically equivalent queries safe to cache-serve
30        # Cosine similarity
31        similarity = np.dot(query_embedding, cached_embedding) / (
32            np.linalg.norm(query_embedding) * np.linalg.norm(cached_embedding)
33        )
34 
35        if similarity >= SIMILARITY_THRESHOLD:
36            return cached_data["response"]  # Cache hit
37 
38    return None  # Cache miss
39 
40def cached_completion(query: str, system_prompt: str) -> str:
41    # Try cache first
42    cached = semantic_cache_lookup(query)
43    if cached:
44        return cached
45 
46    # Generate response
47    response = client.chat.completions.create(
48        model="gpt-4o",
TTL=3600: Expire cache entries after 1 hour — balance freshness vs. hit rate for your use case
49        messages=[
50            {"role": "system", "content": system_prompt},
51            {"role": "user", "content": query}
52        ],
53        stream=False,
54    ).choices[0].message.content
55 
56    # Store in cache with embedding
57    embed_key = f"embed:{hashlib.sha256(query.encode()).hexdigest()[:16]}"
58    redis.setex(embed_key, 3600, json.dumps({  # 1-hour TTL
59        "embedding": get_embedding(query),
60        "response": response,
61        "query": query,
62    }))
63 
64    return response

1from vllm import LLM, SamplingParams
2from vllm.engine.async_llm_engine import AsyncLLMEngine
3from vllm.engine.arg_utils import AsyncEngineArgs
4import asyncio
5 
6# Level 2: vLLM — production inference server with PagedAttention
7# PagedAttention: manages KV cache like virtual memory, enables continuous batching
8# Result: 2-4x more throughput on same hardware vs. naive inference
vLLM's PagedAttention: KV cache management like OS virtual memory — eliminates memory fragmentation
9 
10engine_args = AsyncEngineArgs(
11    model="meta-llama/Llama-3-8B-Instruct",
tensor_parallel_size=2: Model sharded across 2 GPUs. Use for models too large for 1 GPU.
12    tensor_parallel_size=2,          # Split model across 2 GPUs
13    dtype="bfloat16",                # BF16: stable training, good inference quality
14    gpu_memory_utilization=0.90,     # Use 90% of GPU VRAM for KV cache
enable_prefix_caching: Cache KV states for identical system prompt prefixes — big win when system prompt is long
15    max_model_len=8192,              # Max context length
quantization=awq: AWQ 4-bit cuts VRAM 65% with <2% quality loss on most tasks
16    enable_prefix_caching=True,      # Cache KV states for repeated system prompts
17    quantization="awq",              # AWQ 4-bit: 50-75% memory reduction
18)
19 
20engine = AsyncLLMEngine.from_engine_args(engine_args)
21 
22async def generate(prompt: str, max_tokens: int = 512) -> str:
23    sampling_params = SamplingParams(
24        temperature=0.7,
25        top_p=0.9,
26        max_tokens=max_tokens,
27    )
28 
29    # vLLM handles batching automatically via PagedAttention
30    # New requests join existing batches without waiting for current batch to finish
Continuous batching: unlike naive batching, new requests join mid-flight. No head-of-line blocking.
31    results_generator = engine.generate(prompt, sampling_params, request_id=str(id(prompt)))
32 
33    final_output = None
34    async for request_output in results_generator:
35        final_output = request_output
36 
37    return final_output.outputs[0].text
38 
39async def benchmark_throughput():
40    # With vLLM: 1,200-1,800 tokens/second on 2x A100 80GB (Llama-3-8B)
41    # Without vLLM (naive HF): 300-400 tokens/second on same hardware
vLLM throughput: 4-5x vs. naive HuggingFace inference on same hardware
42    # 4-5x throughput improvement from PagedAttention alone
43    prompts = [f"Prompt {i}: Write a paragraph about AI." for i in range(100)]
44    tasks = [generate(p) for p in prompts]
45    results = await asyncio.gather(*tasks)
46    return results

1from openai import OpenAI
2from anthropic import Anthropic
3from pydantic import BaseModel
4 
5client_openai = OpenAI()
6client_anthropic = Anthropic()
7 
8# Level 3: Intelligent model routing
9# Route queries to the cheapest model that can handle them
10# Save 60-80% inference cost vs. always using the most expensive model
Model routing: match query complexity to model capability/cost — biggest lever for cost reduction
11 
12class RouterDecision(BaseModel):
13    model: str                    # Which model to use
14    reasoning: str                # Why this model was chosen
15    complexity_score: int         # 1-5
16 
17ROUTING_SYSTEM_PROMPT = """Classify this query's complexity to determine the optimal model.
18Score 1-5: 
191-2 = Simple (factual lookup, formatting, short completion) -> use gpt-4o-mini
203 = Medium (multi-step reasoning, moderate length) -> use gpt-4o  
214-5 = Complex (deep analysis, creative, long-form, nuanced) -> use claude-3-5-sonnet
Use the cheapest model for routing itself — a 100-token routing call costs 10x less than the main call
22 
23Respond with JSON: model, reasoning, complexity_score"""
24 
25def route_query(user_query: str) -> tuple[str, str]:
The router classifies query complexity in <200ms — negligible latency for 60-80% cost savings
26    # Use a cheap model to route — the routing call costs almost nothing
27    routing_response = client_openai.beta.chat.completions.parse(
28        model="gpt-4o-mini",         # Cheap model for routing decisions
29        messages=[
30            {"role": "system", "content": ROUTING_SYSTEM_PROMPT},
31            {"role": "user", "content": user_query}
32        ],
33        response_format=RouterDecision,
34    )
35 
36    decision = routing_response.choices[0].message.parsed
37    return decision.model, decision.reasoning
38 
39def intelligent_completion(user_query: str, system_prompt: str) -> str:
40    model, reason = route_query(user_query)
41 
42    if model == "gpt-4o-mini":
43        # $0.15/1M input tokens
44        response = client_openai.chat.completions.create(
45            model="gpt-4o-mini",
claude-3-5-sonnet at $3/1M tokens: 20x more expensive than gpt-4o-mini, justified only for complex tasks
46            messages=[{"role": "system", "content": system_prompt},
47                      {"role": "user", "content": user_query}]
48        )
49        return response.choices[0].message.content
50 
51    elif model == "claude-3-5-sonnet":
52        # $3/1M input tokens — 20x more expensive, reserved for complex tasks
53        response = client_anthropic.messages.create(
54            model="claude-3-5-sonnet-20241022",
55            max_tokens=2048,
56            system=system_prompt,
57            messages=[{"role": "user", "content": user_query}]
58        )
59        return response.content[0].text