AI Inference Optimization Techniques
Reduce latency and cost of AI model inference with batching, quantization, caching, and request routing strategies.
What you'll learn
- ✓How batching reduces per-request inference cost
- ✓When and how to apply model quantization
- ✓Request routing strategies across model tiers
- ✓Practical caching patterns for repeated queries
Prerequisites
None — this post is self-contained.
Running AI models in production means balancing three competing priorities: latency, cost, and quality. A naive deployment that sends every request to the largest model with no optimization burns through budgets and frustrates users with slow responses. This guide covers the practical techniques that production teams use to make inference fast and affordable without sacrificing output quality.
Request Batching
Most inference backends process batched requests more efficiently than individual ones. GPUs excel at parallel computation, and batching amortizes the overhead of model loading and memory allocation across multiple inputs.
import asyncio
from dataclasses import dataclass, field
from time import time
@dataclass
class BatchProcessor:
max_batch_size: int = 8
max_wait_seconds: float = 0.1
_queue: list = field(default_factory=list)
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def process_single(self, input_text: str) -> str:
future = asyncio.get_event_loop().create_future()
async with self._lock:
self._queue.append((input_text, future))
if len(self._queue) >= self.max_batch_size:
await self._flush_batch()
# Start timer for partial batch
asyncio.get_event_loop().call_later(
self.max_wait_seconds, lambda: asyncio.ensure_future(self._flush_batch())
)
return await future
async def _flush_batch(self):
async with self._lock:
if not self._queue:
return
batch = self._queue[:self.max_batch_size]
self._queue = self._queue[self.max_batch_size:]
# Process batch together
inputs = [item[0] for item in batch]
results = await self._run_inference_batch(inputs)
for (_, future), result in zip(batch, results):
if not future.done():
future.set_result(result)
async def _run_inference_batch(self, inputs: list[str]) -> list[str]:
# Replace with actual batch inference call
return [f"Result for: {inp}" for inp in inputs]
The two key parameters are max_batch_size and max_wait_seconds. Larger batches improve throughput but increase latency for the first items in the batch. The wait timeout ensures requests are not held indefinitely when traffic is low.
Model Quantization
Quantization reduces model precision from 32-bit or 16-bit floating point to 8-bit or 4-bit integers. This shrinks model size and speeds up inference with minimal quality loss for most tasks.
# Using llama.cpp for quantized inference
from llama_cpp import Llama
# Load a 4-bit quantized model (GGUF format)
llm = Llama(
model_path="./models/llama-3-8b-instruct-Q4_K_M.gguf",
n_ctx=4096,
n_gpu_layers=35, # Offload layers to GPU
n_batch=512,
)
def generate(prompt: str, max_tokens: int = 256) -> str:
output = llm(
prompt,
max_tokens=max_tokens,
temperature=0.7,
stop=["</s>", "\n\n"],
)
return output["choices"][0]["text"]
Quantization levels and their trade-offs:
| Format | Size Reduction | Quality Impact | Best For |
|---|---|---|---|
| FP16 | 2x vs FP32 | Negligible | Default for GPU inference |
| INT8 | 4x vs FP32 | Minimal | Production serving |
| Q4_K_M | 8x vs FP32 | Small for most tasks | Edge deployment, cost reduction |
| Q2_K | 16x vs FP32 | Noticeable | Extreme resource constraints |
For API-based models, you cannot control quantization directly. Instead, choose the right model tier — smaller models like GPT-4o-mini or Claude Haiku serve many tasks at a fraction of the cost.
Tiered Model Routing
Not every request needs the most capable model. A routing layer classifies incoming requests by complexity and sends them to the appropriate model tier.
from enum import Enum
class ModelTier(Enum):
SMALL = "gpt-4o-mini"
MEDIUM = "gpt-4o"
LARGE = "o3"
def classify_complexity(prompt: str) -> ModelTier:
"""Route based on task complexity heuristics."""
word_count = len(prompt.split())
# Simple factual questions
if word_count < 30 and any(
prompt.lower().startswith(q)
for q in ["what is", "who is", "when did", "define"]
):
return ModelTier.SMALL
# Code generation, analysis, complex reasoning
complexity_signals = [
"analyze", "compare", "implement", "debug",
"architecture", "trade-off", "optimize",
]
if any(signal in prompt.lower() for signal in complexity_signals):
return ModelTier.LARGE
return ModelTier.MEDIUM
async def route_request(prompt: str, client) -> str:
tier = classify_complexity(prompt)
response = await client.chat.completions.create(
model=tier.value,
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content
A more sophisticated approach uses a small classifier model to evaluate complexity. Train this classifier on historical request-response pairs where you know which model tier produced acceptable results.
Response Caching
Many applications receive repeated or semantically similar queries. Caching avoids redundant inference calls entirely.
Exact match caching works for deterministic queries:
import hashlib
import json
from functools import lru_cache
def cache_key(model: str, messages: list[dict], temperature: float) -> str:
content = json.dumps({"model": model, "messages": messages, "temp": temperature})
return hashlib.sha256(content.encode()).hexdigest()
class InferenceCache:
def __init__(self):
self._cache: dict[str, str] = {}
def get(self, key: str) -> str | None:
return self._cache.get(key)
def set(self, key: str, value: str, ttl_seconds: int = 3600):
self._cache[key] = value
async def cached_inference(self, client, model, messages, temperature=0):
key = cache_key(model, messages, temperature)
cached = self.get(key)
if cached:
return cached
response = await client.chat.completions.create(
model=model, messages=messages, temperature=temperature,
)
result = response.choices[0].message.content
self.set(key, result)
return result
For production systems, replace the dictionary with Redis or Memcached and add TTL expiration. Set temperature to 0 for cached requests to ensure deterministic outputs.
Streaming for Perceived Latency
Streaming does not reduce actual inference time, but it dramatically improves perceived latency. Users see the first tokens within milliseconds instead of waiting for the full response.
async def stream_response(client, messages: list[dict]):
stream = await client.chat.completions.create(
model="gpt-4o",
messages=messages,
stream=True,
)
full_response = []
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
full_response.append(delta)
yield delta # Send to client immediately
return "".join(full_response)
Combine streaming with the other optimizations. Route to the right model, check the cache first, and stream the response if the cache misses.
Measuring Optimization Impact
Track these metrics before and after applying optimizations:
import time
from dataclasses import dataclass
@dataclass
class InferenceMetrics:
total_requests: int = 0
cache_hits: int = 0
total_tokens: int = 0
total_latency_ms: float = 0
@property
def cache_hit_rate(self) -> float:
return self.cache_hits / max(self.total_requests, 1)
@property
def avg_latency_ms(self) -> float:
return self.total_latency_ms / max(self.total_requests, 1)
@property
def cost_per_request(self) -> float:
avg_tokens = self.total_tokens / max(self.total_requests, 1)
return avg_tokens * 0.000005 # Adjust per model pricing
Focus on the metrics that matter for your application. Chatbots need low latency. Batch processing pipelines need high throughput. Cost-sensitive applications need efficient routing.
Summary
Inference optimization is not a single technique but a combination of strategies applied together. Start with model routing to avoid overspending on simple queries. Add response caching for repeated patterns. Use batching for throughput-heavy workloads. Apply quantization when self-hosting models. Layer streaming on top for user-facing applications. Measure before and after each change to confirm the optimization actually helps for your specific workload.
Related articles
- AI Semantic Caching for LLM Applications
Reduce LLM API costs and latency by caching responses based on semantic similarity rather than exact string matching.
- Go Go Profiling with pprof: CPU, Memory, and Goroutines
Learn how to profile Go applications using pprof to find CPU bottlenecks, memory leaks, and goroutine issues with practical examples.
- React React Performance: Memo, useMemo, and useCallback Guide
Optimize React performance with React.memo, useMemo, and useCallback. Learn when to use each, how to profile, and avoid common pitfalls.
- Python Python Profiling: Find and Fix Performance Bottlenecks
Learn how to profile Python code with cProfile, line_profiler, memory_profiler, and timeit to identify slow functions, memory leaks, and optimize runtime performance.