Inference Optimization Patterns: Maximizing LLM Throughput and Efficiency

Introduction: LLM inference is expensive—both in compute and latency. Every token generated requires a forward pass through billions of parameters, and users expect responses in seconds, not minutes. Inference optimization techniques reduce costs and improve responsiveness without sacrificing output quality. This guide covers practical optimization strategies: batching requests to maximize GPU utilization, managing KV caches to avoid redundant computation, quantizing models to reduce memory footprint, speculative decoding to accelerate generation, and continuous batching for production throughput. Whether you’re serving a chatbot to millions of users or running inference on edge devices, these techniques determine whether your LLM application is viable at scale.

Request Batching

from dataclasses import dataclass, field
from typing import Any, Optional, Callable
from datetime import datetime
import asyncio
from collections import deque

@dataclass
class InferenceRequest:
    """A single inference request."""
    
    id: str
    prompt: str
    max_tokens: int = 100
    temperature: float = 0.7
    created_at: datetime = field(default_factory=datetime.utcnow)
    future: asyncio.Future = None

@dataclass
class InferenceResult:
    """Result of inference."""
    
    request_id: str
    text: str
    tokens_generated: int
    latency_ms: float

class DynamicBatcher:
    """Dynamic request batching for inference."""
    
    def __init__(
        self,
        model: Any,
        max_batch_size: int = 32,
        max_wait_ms: float = 50.0,
        max_tokens_per_batch: int = 4096
    ):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.max_tokens_per_batch = max_tokens_per_batch
        
        self._queue: deque[InferenceRequest] = deque()
        self._lock = asyncio.Lock()
        self._batch_event = asyncio.Event()
        self._running = False
    
    async def start(self):
        """Start the batcher."""
        self._running = True
        asyncio.create_task(self._batch_loop())
    
    async def stop(self):
        """Stop the batcher."""
        self._running = False
        self._batch_event.set()
    
    async def infer(self, request: InferenceRequest) -> InferenceResult:
        """Submit request and wait for result."""
        
        request.future = asyncio.get_event_loop().create_future()
        
        async with self._lock:
            self._queue.append(request)
            self._batch_event.set()
        
        return await request.future
    
    async def _batch_loop(self):
        """Main batching loop."""
        
        while self._running:
            await self._batch_event.wait()
            self._batch_event.clear()
            
            # Wait for more requests or timeout
            await asyncio.sleep(self.max_wait_ms / 1000)
            
            # Collect batch
            batch = await self._collect_batch()
            
            if batch:
                # Process batch
                results = await self._process_batch(batch)
                
                # Deliver results
                for request, result in zip(batch, results):
                    if not request.future.done():
                        request.future.set_result(result)
    
    async def _collect_batch(self) -> list[InferenceRequest]:
        """Collect requests into a batch."""
        
        batch = []
        total_tokens = 0
        
        async with self._lock:
            while self._queue and len(batch) < self.max_batch_size:
                request = self._queue[0]
                
                # Estimate tokens
                prompt_tokens = len(request.prompt.split()) * 1.3
                request_tokens = prompt_tokens + request.max_tokens
                
                if total_tokens + request_tokens > self.max_tokens_per_batch:
                    break
                
                batch.append(self._queue.popleft())
                total_tokens += request_tokens
        
        return batch
    
    async def _process_batch(
        self,
        batch: list[InferenceRequest]
    ) -> list[InferenceResult]:
        """Process a batch of requests."""
        
        start_time = datetime.utcnow()
        
        # Prepare batch inputs
        prompts = [r.prompt for r in batch]
        max_tokens = max(r.max_tokens for r in batch)
        
        # Run inference
        outputs = await self.model.generate_batch(
            prompts,
            max_tokens=max_tokens
        )
        
        end_time = datetime.utcnow()
        latency = (end_time - start_time).total_seconds() * 1000
        
        # Create results
        results = []
        for request, output in zip(batch, outputs):
            results.append(InferenceResult(
                request_id=request.id,
                text=output,
                tokens_generated=len(output.split()),
                latency_ms=latency
            ))
        
        return results

class ContinuousBatcher:
    """Continuous batching for streaming inference."""
    
    def __init__(
        self,
        model: Any,
        max_batch_size: int = 64,
        max_sequence_length: int = 2048
    ):
        self.model = model
        self.max_batch_size = max_batch_size
        self.max_sequence_length = max_sequence_length
        
        self._active_requests: dict[str, InferenceRequest] = {}
        self._pending_queue: deque[InferenceRequest] = deque()
        self._lock = asyncio.Lock()
    
    async def add_request(self, request: InferenceRequest):
        """Add request to processing."""
        
        async with self._lock:
            if len(self._active_requests) < self.max_batch_size:
                self._active_requests[request.id] = request
            else:
                self._pending_queue.append(request)
    
    async def step(self) -> dict[str, str]:
        """Generate one token for all active requests."""
        
        if not self._active_requests:
            return {}
        
        # Get current state of all active requests
        requests = list(self._active_requests.values())
        
        # Generate next token for batch
        next_tokens = await self.model.generate_next_token_batch(
            [r.prompt for r in requests]
        )
        
        results = {}
        completed = []
        
        for request, token in zip(requests, next_tokens):
            # Update prompt with new token
            request.prompt += token
            results[request.id] = token
            
            # Check completion
            if self._is_complete(request, token):
                completed.append(request.id)
        
        # Remove completed requests
        async with self._lock:
            for req_id in completed:
                del self._active_requests[req_id]
                
                # Add pending request
                if self._pending_queue:
                    new_request = self._pending_queue.popleft()
                    self._active_requests[new_request.id] = new_request
        
        return results
    
    def _is_complete(self, request: InferenceRequest, token: str) -> bool:
        """Check if request is complete."""
        
        # Check for EOS token
        if token in ["<|endoftext|>", "", "<|end|>"]:
            return True
        
        # Check max tokens
        generated = len(request.prompt.split()) - len(request.prompt.split())
        if generated >= request.max_tokens:
            return True
        
        return False

KV Cache Management

from dataclasses import dataclass
from typing import Any, Optional
import torch

@dataclass
class KVCacheEntry:
    """A KV cache entry."""
    
    key: torch.Tensor
    value: torch.Tensor
    sequence_length: int
    last_accessed: float

class KVCacheManager:
    """Manage KV caches for inference."""
    
    def __init__(
        self,
        num_layers: int,
        num_heads: int,
        head_dim: int,
        max_cache_size_gb: float = 8.0,
        device: str = "cuda"
    ):
        self.num_layers = num_layers
        self.num_heads = num_heads
        self.head_dim = head_dim
        self.max_cache_size = int(max_cache_size_gb * 1e9)
        self.device = device
        
        self._caches: dict[str, list[KVCacheEntry]] = {}
        self._current_size = 0
    
    def get_cache(self, request_id: str) -> Optional[list[KVCacheEntry]]:
        """Get KV cache for request."""
        
        cache = self._caches.get(request_id)
        if cache:
            import time
            for entry in cache:
                entry.last_accessed = time.time()
        return cache
    
    def set_cache(
        self,
        request_id: str,
        keys: list[torch.Tensor],
        values: list[torch.Tensor],
        seq_len: int
    ):
        """Set KV cache for request."""
        
        import time
        
        # Calculate size
        cache_size = sum(
            k.numel() * k.element_size() + v.numel() * v.element_size()
            for k, v in zip(keys, values)
        )
        
        # Evict if necessary
        while self._current_size + cache_size > self.max_cache_size:
            if not self._evict_oldest():
                break
        
        # Store cache
        entries = []
        for k, v in zip(keys, values):
            entries.append(KVCacheEntry(
                key=k,
                value=v,
                sequence_length=seq_len,
                last_accessed=time.time()
            ))
        
        self._caches[request_id] = entries
        self._current_size += cache_size
    
    def extend_cache(
        self,
        request_id: str,
        new_keys: list[torch.Tensor],
        new_values: list[torch.Tensor]
    ):
        """Extend existing cache with new tokens."""
        
        cache = self._caches.get(request_id)
        if not cache:
            return
        
        for i, (entry, new_k, new_v) in enumerate(zip(cache, new_keys, new_values)):
            # Concatenate along sequence dimension
            entry.key = torch.cat([entry.key, new_k], dim=-2)
            entry.value = torch.cat([entry.value, new_v], dim=-2)
            entry.sequence_length += new_k.shape[-2]
    
    def delete_cache(self, request_id: str):
        """Delete cache for request."""
        
        if request_id in self._caches:
            cache = self._caches[request_id]
            cache_size = sum(
                e.key.numel() * e.key.element_size() +
                e.value.numel() * e.value.element_size()
                for e in cache
            )
            del self._caches[request_id]
            self._current_size -= cache_size
    
    def _evict_oldest(self) -> bool:
        """Evict oldest cache entry."""
        
        if not self._caches:
            return False
        
        # Find oldest
        oldest_id = min(
            self._caches.keys(),
            key=lambda k: min(e.last_accessed for e in self._caches[k])
        )
        
        self.delete_cache(oldest_id)
        return True

class PagedKVCache:
    """Paged attention KV cache (vLLM style)."""
    
    def __init__(
        self,
        num_layers: int,
        num_heads: int,
        head_dim: int,
        block_size: int = 16,
        num_blocks: int = 1000,
        device: str = "cuda"
    ):
        self.num_layers = num_layers
        self.num_heads = num_heads
        self.head_dim = head_dim
        self.block_size = block_size
        self.num_blocks = num_blocks
        self.device = device
        
        # Pre-allocate block pool
        self.key_blocks = torch.zeros(
            num_blocks, num_layers, block_size, num_heads, head_dim,
            device=device, dtype=torch.float16
        )
        self.value_blocks = torch.zeros(
            num_blocks, num_layers, block_size, num_heads, head_dim,
            device=device, dtype=torch.float16
        )
        
        # Block allocation tracking
        self.free_blocks = list(range(num_blocks))
        self.sequence_blocks: dict[str, list[int]] = {}
    
    def allocate_blocks(self, request_id: str, num_tokens: int) -> list[int]:
        """Allocate blocks for a sequence."""
        
        num_blocks_needed = (num_tokens + self.block_size - 1) // self.block_size
        
        if len(self.free_blocks) < num_blocks_needed:
            raise RuntimeError("Out of KV cache blocks")
        
        allocated = []
        for _ in range(num_blocks_needed):
            block_id = self.free_blocks.pop()
            allocated.append(block_id)
        
        self.sequence_blocks[request_id] = allocated
        return allocated
    
    def free_blocks_for_sequence(self, request_id: str):
        """Free blocks for a sequence."""
        
        if request_id in self.sequence_blocks:
            blocks = self.sequence_blocks.pop(request_id)
            self.free_blocks.extend(blocks)
    
    def write_kv(
        self,
        request_id: str,
        layer_idx: int,
        position: int,
        key: torch.Tensor,
        value: torch.Tensor
    ):
        """Write KV to cache."""
        
        blocks = self.sequence_blocks.get(request_id, [])
        block_idx = position // self.block_size
        offset = position % self.block_size
        
        if block_idx >= len(blocks):
            # Need more blocks
            new_block = self.free_blocks.pop()
            blocks.append(new_block)
            self.sequence_blocks[request_id] = blocks
        
        block_id = blocks[block_idx]
        self.key_blocks[block_id, layer_idx, offset] = key
        self.value_blocks[block_id, layer_idx, offset] = value
    
    def read_kv(
        self,
        request_id: str,
        layer_idx: int,
        positions: list[int]
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Read KV from cache."""
        
        blocks = self.sequence_blocks.get(request_id, [])
        
        keys = []
        values = []
        
        for pos in positions:
            block_idx = pos // self.block_size
            offset = pos % self.block_size
            
            if block_idx < len(blocks):
                block_id = blocks[block_idx]
                keys.append(self.key_blocks[block_id, layer_idx, offset])
                values.append(self.value_blocks[block_id, layer_idx, offset])
        
        return torch.stack(keys), torch.stack(values)

Model Quantization

from dataclasses import dataclass
from typing import Any, Optional
from enum import Enum
import torch
import torch.nn as nn

class QuantizationType(Enum):
    """Quantization types."""
    
    INT8 = "int8"
    INT4 = "int4"
    FP8 = "fp8"
    NF4 = "nf4"  # Normal float 4-bit

@dataclass
class QuantizationConfig:
    """Quantization configuration."""
    
    quant_type: QuantizationType
    group_size: int = 128
    use_double_quant: bool = False
    compute_dtype: torch.dtype = torch.float16

class Int8Quantizer:
    """INT8 quantization."""
    
    def __init__(self, symmetric: bool = True):
        self.symmetric = symmetric
    
    def quantize(self, tensor: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
        """Quantize tensor to INT8."""
        
        if self.symmetric:
            # Symmetric quantization
            scale = tensor.abs().max() / 127
            quantized = (tensor / scale).round().clamp(-128, 127).to(torch.int8)
            return quantized, scale
        else:
            # Asymmetric quantization
            min_val = tensor.min()
            max_val = tensor.max()
            scale = (max_val - min_val) / 255
            zero_point = (-min_val / scale).round()
            quantized = ((tensor / scale) + zero_point).round().clamp(0, 255).to(torch.uint8)
            return quantized, torch.tensor([scale, zero_point])
    
    def dequantize(
        self,
        quantized: torch.Tensor,
        scale: torch.Tensor
    ) -> torch.Tensor:
        """Dequantize INT8 to float."""
        
        if self.symmetric:
            return quantized.float() * scale
        else:
            scale_val, zero_point = scale[0], scale[1]
            return (quantized.float() - zero_point) * scale_val

class Int4Quantizer:
    """INT4 quantization with grouping."""
    
    def __init__(self, group_size: int = 128):
        self.group_size = group_size
    
    def quantize(self, tensor: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
        """Quantize tensor to INT4."""
        
        # Reshape for grouping
        original_shape = tensor.shape
        tensor = tensor.reshape(-1, self.group_size)
        
        # Compute scales per group
        scales = tensor.abs().max(dim=1, keepdim=True).values / 7
        scales = scales.clamp(min=1e-8)
        
        # Quantize
        quantized = (tensor / scales).round().clamp(-8, 7).to(torch.int8)
        
        # Pack two INT4 values into one INT8
        packed = self._pack_int4(quantized)
        
        return packed, scales.reshape(-1)
    
    def _pack_int4(self, tensor: torch.Tensor) -> torch.Tensor:
        """Pack INT4 values into INT8."""
        
        # Ensure even number of elements
        flat = tensor.reshape(-1)
        if flat.shape[0] % 2 != 0:
            flat = torch.cat([flat, torch.zeros(1, dtype=flat.dtype)])
        
        # Pack pairs
        low = flat[0::2] & 0x0F
        high = (flat[1::2] & 0x0F) << 4
        packed = (low | high).to(torch.uint8)
        
        return packed
    
    def dequantize(
        self,
        packed: torch.Tensor,
        scales: torch.Tensor,
        original_shape: tuple
    ) -> torch.Tensor:
        """Dequantize INT4 to float."""
        
        # Unpack
        unpacked = self._unpack_int4(packed)
        
        # Reshape and apply scales
        unpacked = unpacked.reshape(-1, self.group_size).float()
        scales = scales.reshape(-1, 1)
        
        dequantized = unpacked * scales
        
        return dequantized.reshape(original_shape)
    
    def _unpack_int4(self, packed: torch.Tensor) -> torch.Tensor:
        """Unpack INT4 values from INT8."""
        
        low = (packed & 0x0F).to(torch.int8)
        high = ((packed >> 4) & 0x0F).to(torch.int8)
        
        # Sign extend
        low = torch.where(low > 7, low - 16, low)
        high = torch.where(high > 7, high - 16, high)
        
        # Interleave
        unpacked = torch.stack([low, high], dim=-1).reshape(-1)
        
        return unpacked

class QuantizedLinear(nn.Module):
    """Quantized linear layer."""
    
    def __init__(
        self,
        in_features: int,
        out_features: int,
        config: QuantizationConfig
    ):
        super().__init__()
        
        self.in_features = in_features
        self.out_features = out_features
        self.config = config
        
        # Quantized weights (stored as INT8 or packed INT4)
        self.register_buffer("weight_quantized", None)
        self.register_buffer("weight_scales", None)
        self.register_buffer("bias", None)
    
    @classmethod
    def from_linear(
        cls,
        linear: nn.Linear,
        config: QuantizationConfig
    ) -> "QuantizedLinear":
        """Create from existing linear layer."""
        
        layer = cls(linear.in_features, linear.out_features, config)
        
        # Quantize weights
        if config.quant_type == QuantizationType.INT8:
            quantizer = Int8Quantizer()
            quantized, scales = quantizer.quantize(linear.weight.data)
        elif config.quant_type == QuantizationType.INT4:
            quantizer = Int4Quantizer(config.group_size)
            quantized, scales = quantizer.quantize(linear.weight.data)
        else:
            raise ValueError(f"Unsupported quantization: {config.quant_type}")
        
        layer.weight_quantized = quantized
        layer.weight_scales = scales
        
        if linear.bias is not None:
            layer.bias = linear.bias.data.clone()
        
        return layer
    
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """Forward pass with dequantization."""
        
        # Dequantize weights
        if self.config.quant_type == QuantizationType.INT8:
            quantizer = Int8Quantizer()
            weight = quantizer.dequantize(
                self.weight_quantized,
                self.weight_scales
            )
        elif self.config.quant_type == QuantizationType.INT4:
            quantizer = Int4Quantizer(self.config.group_size)
            weight = quantizer.dequantize(
                self.weight_quantized,
                self.weight_scales,
                (self.out_features, self.in_features)
            )
        else:
            weight = self.weight_quantized.float()
        
        # Compute
        output = torch.nn.functional.linear(x, weight, self.bias)
        
        return output

Speculative Decoding

from dataclasses import dataclass
from typing import Any, Optional
import torch

@dataclass
class SpeculativeConfig:
    """Speculative decoding configuration."""
    
    num_speculative_tokens: int = 4
    temperature: float = 0.0
    top_p: float = 1.0

class SpeculativeDecoder:
    """Speculative decoding for faster inference."""
    
    def __init__(
        self,
        target_model: Any,
        draft_model: Any,
        config: SpeculativeConfig
    ):
        self.target_model = target_model
        self.draft_model = draft_model
        self.config = config
    
    async def generate(
        self,
        prompt: str,
        max_tokens: int
    ) -> str:
        """Generate with speculative decoding."""
        
        generated_tokens = []
        current_prompt = prompt
        
        while len(generated_tokens) < max_tokens:
            # Draft phase: generate speculative tokens
            draft_tokens = await self._draft_phase(current_prompt)
            
            # Verify phase: check with target model
            accepted_tokens = await self._verify_phase(
                current_prompt,
                draft_tokens
            )
            
            # Update state
            generated_tokens.extend(accepted_tokens)
            current_prompt += "".join(accepted_tokens)
            
            # Check for EOS
            if any(t in ["<|endoftext|>", ""] for t in accepted_tokens):
                break
        
        return "".join(generated_tokens)
    
    async def _draft_phase(self, prompt: str) -> list[str]:
        """Generate draft tokens with small model."""
        
        draft_tokens = []
        current = prompt
        
        for _ in range(self.config.num_speculative_tokens):
            token = await self.draft_model.generate_next_token(
                current,
                temperature=self.config.temperature
            )
            draft_tokens.append(token)
            current += token
        
        return draft_tokens
    
    async def _verify_phase(
        self,
        prompt: str,
        draft_tokens: list[str]
    ) -> list[str]:
        """Verify draft tokens with target model."""
        
        # Get target model probabilities for all positions
        full_sequence = prompt + "".join(draft_tokens)
        target_probs = await self.target_model.get_token_probabilities(
            full_sequence
        )
        
        # Get draft model probabilities
        draft_probs = await self.draft_model.get_token_probabilities(
            full_sequence
        )
        
        accepted = []
        
        for i, token in enumerate(draft_tokens):
            # Get probabilities at position
            p_target = target_probs[len(prompt) + i].get(token, 0)
            p_draft = draft_probs[len(prompt) + i].get(token, 0)
            
            # Acceptance criterion
            if p_draft > 0:
                acceptance_prob = min(1, p_target / p_draft)
            else:
                acceptance_prob = 1 if p_target > 0 else 0
            
            # Accept or reject
            import random
            if random.random() < acceptance_prob:
                accepted.append(token)
            else:
                # Sample from adjusted distribution
                adjusted_token = self._sample_adjusted(
                    target_probs[len(prompt) + i],
                    draft_probs[len(prompt) + i]
                )
                accepted.append(adjusted_token)
                break  # Stop at first rejection
        
        return accepted
    
    def _sample_adjusted(
        self,
        target_probs: dict[str, float],
        draft_probs: dict[str, float]
    ) -> str:
        """Sample from adjusted distribution."""
        
        # Compute adjusted probabilities
        adjusted = {}
        for token in target_probs:
            p_t = target_probs.get(token, 0)
            p_d = draft_probs.get(token, 0)
            adjusted[token] = max(0, p_t - p_d)
        
        # Normalize
        total = sum(adjusted.values())
        if total > 0:
            adjusted = {k: v / total for k, v in adjusted.items()}
        else:
            adjusted = target_probs
        
        # Sample
        import random
        r = random.random()
        cumsum = 0
        for token, prob in adjusted.items():
            cumsum += prob
            if r < cumsum:
                return token
        
        return list(adjusted.keys())[0]

class MedusaDecoder:
    """Medusa-style parallel decoding."""
    
    def __init__(
        self,
        model: Any,
        num_heads: int = 4,
        num_candidates: int = 5
    ):
        self.model = model
        self.num_heads = num_heads
        self.num_candidates = num_candidates
    
    async def generate(
        self,
        prompt: str,
        max_tokens: int
    ) -> str:
        """Generate with Medusa decoding."""
        
        generated = []
        current = prompt
        
        while len(generated) < max_tokens:
            # Generate candidates from all heads
            candidates = await self._generate_candidates(current)
            
            # Verify candidates
            accepted = await self._verify_candidates(current, candidates)
            
            generated.extend(accepted)
            current += "".join(accepted)
            
            if not accepted:
                # Fall back to single token
                token = await self.model.generate_next_token(current)
                generated.append(token)
                current += token
        
        return "".join(generated)
    
    async def _generate_candidates(
        self,
        prompt: str
    ) -> list[list[str]]:
        """Generate candidate sequences from Medusa heads."""
        
        # Each head predicts tokens at different positions
        candidates = []
        
        for head_idx in range(self.num_heads):
            head_candidates = await self.model.medusa_head_predict(
                prompt,
                head_idx,
                self.num_candidates
            )
            candidates.append(head_candidates)
        
        return candidates
    
    async def _verify_candidates(
        self,
        prompt: str,
        candidates: list[list[str]]
    ) -> list[str]:
        """Verify candidate sequences."""
        
        # Build candidate trees
        trees = self._build_candidate_trees(candidates)
        
        # Verify with single forward pass
        best_sequence = []
        
        for tree in trees:
            verified = await self.model.verify_sequence(prompt, tree)
            if len(verified) > len(best_sequence):
                best_sequence = verified
        
        return best_sequence
    
    def _build_candidate_trees(
        self,
        candidates: list[list[str]]
    ) -> list[list[str]]:
        """Build candidate trees from head predictions."""
        
        # Simplified: just return top candidates
        trees = []
        for head_candidates in candidates:
            for candidate in head_candidates[:self.num_candidates]:
                trees.append([candidate])
        return trees

Production Inference Service

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from enum import Enum
import asyncio

app = FastAPI()

# Initialize components
batcher = None  # Would initialize DynamicBatcher
kv_cache = None  # Would initialize KVCacheManager

class OptimizationLevel(str, Enum):
    NONE = "none"
    BASIC = "basic"  # Batching only
    FULL = "full"  # Batching + KV cache + quantization

class GenerateRequest(BaseModel):
    prompt: str
    max_tokens: int = 100
    temperature: float = 0.7
    stream: bool = False
    optimization: OptimizationLevel = OptimizationLevel.FULL

class GenerateResponse(BaseModel):
    text: str
    tokens_generated: int
    latency_ms: float
    optimization_used: str

class BatchGenerateRequest(BaseModel):
    prompts: list[str]
    max_tokens: int = 100
    temperature: float = 0.7

class BatchGenerateResponse(BaseModel):
    results: list[GenerateResponse]
    total_latency_ms: float
    throughput_tokens_per_sec: float

@app.post("/v1/generate")
async def generate(request: GenerateRequest) -> GenerateResponse:
    """Generate text with optimization."""
    
    import time
    import uuid
    
    start = time.time()
    
    # Create inference request
    req = InferenceRequest(
        id=str(uuid.uuid4()),
        prompt=request.prompt,
        max_tokens=request.max_tokens,
        temperature=request.temperature
    )
    
    # Submit to batcher
    if batcher:
        result = await batcher.infer(req)
        text = result.text
        tokens = result.tokens_generated
    else:
        # Direct inference (placeholder)
        text = "Generated text placeholder"
        tokens = 10
    
    latency = (time.time() - start) * 1000
    
    return GenerateResponse(
        text=text,
        tokens_generated=tokens,
        latency_ms=latency,
        optimization_used=request.optimization.value
    )

@app.post("/v1/generate/batch")
async def batch_generate(request: BatchGenerateRequest) -> BatchGenerateResponse:
    """Batch generate text."""
    
    import time
    import uuid
    
    start = time.time()
    
    # Create requests
    requests = [
        InferenceRequest(
            id=str(uuid.uuid4()),
            prompt=prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature
        )
        for prompt in request.prompts
    ]
    
    # Submit all
    if batcher:
        tasks = [batcher.infer(req) for req in requests]
        results = await asyncio.gather(*tasks)
    else:
        results = [
            InferenceResult(
                request_id=req.id,
                text="Placeholder",
                tokens_generated=10,
                latency_ms=100
            )
            for req in requests
        ]
    
    total_latency = (time.time() - start) * 1000
    total_tokens = sum(r.tokens_generated for r in results)
    throughput = total_tokens / (total_latency / 1000) if total_latency > 0 else 0
    
    return BatchGenerateResponse(
        results=[
            GenerateResponse(
                text=r.text,
                tokens_generated=r.tokens_generated,
                latency_ms=r.latency_ms,
                optimization_used="batch"
            )
            for r in results
        ],
        total_latency_ms=total_latency,
        throughput_tokens_per_sec=throughput
    )

class CacheStatsResponse(BaseModel):
    total_entries: int
    total_size_mb: float
    hit_rate: float

@app.get("/v1/cache/stats")
async def cache_stats() -> CacheStatsResponse:
    """Get KV cache statistics."""
    
    if kv_cache:
        return CacheStatsResponse(
            total_entries=len(kv_cache._caches),
            total_size_mb=kv_cache._current_size / 1e6,
            hit_rate=0.85  # Would track actual hit rate
        )
    
    return CacheStatsResponse(
        total_entries=0,
        total_size_mb=0,
        hit_rate=0
    )

@app.delete("/v1/cache")
async def clear_cache():
    """Clear KV cache."""
    
    if kv_cache:
        kv_cache._caches.clear()
        kv_cache._current_size = 0
    
    return {"status": "cleared"}

class ModelInfoResponse(BaseModel):
    model_name: str
    quantization: Optional[str]
    batch_size: int
    max_sequence_length: int

@app.get("/v1/model/info")
async def model_info() -> ModelInfoResponse:
    """Get model information."""
    
    return ModelInfoResponse(
        model_name="llama-7b",
        quantization="int4",
        batch_size=32,
        max_sequence_length=4096
    )

@app.get("/health")
async def health():
    return {"status": "healthy"}

@app.get("/metrics")
async def metrics():
    """Get inference metrics."""
    
    return {
        "requests_total": 1000,
        "requests_per_second": 50,
        "avg_latency_ms": 150,
        "p99_latency_ms": 500,
        "tokens_per_second": 2000,
        "batch_utilization": 0.75,
        "cache_hit_rate": 0.85
    }

References

• vLLM: https://github.com/vllm-project/vllm
• PagedAttention: https://arxiv.org/abs/2309.06180
• Speculative Decoding: https://arxiv.org/abs/2211.17192
• GPTQ Quantization: https://arxiv.org/abs/2210.17323
• Medusa: https://arxiv.org/abs/2401.10774

Conclusion

Inference optimization determines whether LLM applications are economically viable at scale. Start with dynamic batching—grouping requests together maximizes GPU utilization and amortizes the cost of loading model weights. Implement KV caching to avoid recomputing attention for previously processed tokens; paged attention (vLLM-style) enables efficient memory management for variable-length sequences. Quantize models to INT8 or INT4 to reduce memory footprint and increase throughput, with minimal quality degradation for most tasks. Use speculative decoding with a small draft model to generate multiple tokens per forward pass of the large model. Continuous batching handles streaming scenarios where requests arrive and complete at different times. Monitor key metrics: tokens per second, batch utilization, cache hit rate, and latency percentiles. The key insight is that inference optimization is about maximizing useful computation per GPU cycle—every optimization technique reduces waste, whether it's redundant computation (KV cache), memory bandwidth (quantization), or idle time (batching). Combine these techniques for multiplicative improvements in throughput and cost efficiency.