Skip to content
Codeloom
AI

AI Inference Optimization Techniques

Reduce latency and cost of AI model inference with batching, quantization, caching, and request routing strategies.

·6 min read · By Codeloom
Intermediate 11 min read

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:

FormatSize ReductionQuality ImpactBest For
FP162x vs FP32NegligibleDefault for GPU inference
INT84x vs FP32MinimalProduction serving
Q4_K_M8x vs FP32Small for most tasksEdge deployment, cost reduction
Q2_K16x vs FP32NoticeableExtreme 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.