System Design Patterns for AI: Batch Inference, Caching, and Async Pipelines | FromZero2AI Blog

Building AI applications in production requires a shift in how we approach software architecture. Traditional web servers process requests in milliseconds, but Large Language Model (LLM) invocations can take seconds or even minutes. Furthermore, API tokens represent a significant operational expense, and third-party APIs are subject to tight rate limits.

To build responsive, reliable, and cost-effective AI systems, engineers must implement specific system design patterns: Semantic Caching, Asynchronous Task Pipelines, and Offline Batch Processing. This article breaks down these architectural patterns.

AI System Design Patterns: Semantic Caching and Asynchronous Worker Pipeline AI System Design Patterns: Semantic Caching and Asynchronous Worker Pipeline

1. Semantic Prompt Caching

Traditional caching strategies (like Redis key-value stores) rely on exact string matching. If a user asks "How do I connect to a database in Python?" and another asks "What is the code to connect to a Python database?", an exact-match cache fails to recognize that these questions have the same meaning.

Semantic Caching solves this by storing cache entries in a vector database.

The Workflow

The incoming prompt is passed to an embedding model to generate a vector representation.
A vector similarity search (cosine similarity or Euclidean distance) is performed against the cache database.
If the nearest neighbor has a similarity score above a set threshold (e.g., 0.96), the system returns the cached response, bypassing the LLM.
If not, the prompt is sent to the LLM, and the response is subsequently stored in the vector database for future matches.

2. Asynchronous Task Workers (The 202 Pattern)

When processing complex agentic flows or multi-step reasoning chains, running the pipeline inside a synchronous HTTP request-response cycle is a recipe for timeouts and poor user experience. Instead, engineers use the Asynchronous Task Queue pattern.

The Architecture

Web Server: Receives the request, validates the payload, generates a unique job ID, writes the task to a message queue (e.g., Redis via BullMQ or Celery), and immediately returns a 202 Accepted status to the client with the job ID.
Worker Pool: Background workers pull jobs from the queue, execute the heavy LLM pipelines, write intermediary updates, and save the final output to a database.
Status Delivery: The client retrieves updates either by polling a status endpoint (e.g., /jobs/:id) or by listening to a WebSocket channel.

3. Offline Batch Queuing

For non-real-time workloads—such as generating daily product descriptions, analyzing system logs, or run-time evaluation datasets—real-time APIs are unnecessarily expensive and risk hitting rate limits.

Modern LLM providers offer Batch APIs (e.g., OpenAI Batch API, Anthropic Message Batches) that accept a list of tasks in a single file, run them asynchronously on their infrastructure, and return the results within 24 hours. In exchange, providers offer a 50% discount on token costs. System designs should isolate batch tasks from real-time queues to leverage these savings.

Python Implementation: A Simple Semantic Cache

Below is a complete, runnable Python class demonstrating how to build a semantic prompt cache using sentence-transformers and cosine similarity.

import numpy as np
from sentence_transformers import SentenceTransformer

class SemanticCache:
    def __init__(self, threshold: float = 0.92):
        # Load a lightweight, fast embedding model
        self.encoder = SentenceTransformer("all-MiniLM-L6-v2")
        self.threshold = threshold
        # In-memory storage for vectors and corresponding text responses
        self.cache_vectors = []
        self.cache_data = []

    def _cosine_similarity(self, a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def get(self, prompt: str) -> str | None:
        """Looks up the prompt in the semantic cache."""
        if not self.cache_vectors:
            return None
        
        # 1. Embed the query prompt
        query_vector = self.encoder.encode(prompt)
        
        # 2. Search for the closest match in the cache
        best_score = -1.0
        best_index = -1
        
        for idx, cache_vector in enumerate(self.cache_vectors):
            score = self._cosine_similarity(query_vector, cache_vector)
            if score > best_score:
                best_score = score
                best_index = idx
                
        # 3. Return cache hit if similarity is above the threshold
        if best_score >= self.threshold:
            print(f"[CACHE HIT] Similarity: {best_score:.4f}")
            return self.cache_data[best_index]["response"]
            
        print(f"[CACHE MISS] Best similarity: {best_score:.4f}")
        return None

    def set(self, prompt: str, response: str):
        """Saves a new prompt-response pair to the cache."""
        vector = self.encoder.encode(prompt)
        self.cache_vectors.append(vector)
        self.cache_data.append({"prompt": prompt, "response": response})
        print(f"[CACHE WRITE] Saved prompt: '{prompt}'")

# --- Demo Usage ---
if __name__ == "__main__":
    cache = SemanticCache(threshold=0.90)
    
    # First search: Cache Miss
    prompt1 = "How do I write a file in Python?"
    if not (res := cache.get(prompt1)):
        # Simulate LLM Call
        llm_response = "Use: with open('file.txt', 'w') as f: f.write('hello')"
        cache.set(prompt1, llm_response)
        res = llm_response
    print(f"Result: {res}\n")
    
    # Second search (Semantically similar prompt): Cache Hit
    prompt2 = "What is the syntax to write files using Python?"
    cached_res = cache.get(prompt2)
    print(f"Result from cache: {cached_res}")

Pattern Matrix

Implementing these patterns in your system architecture ensures that your application remains fast, stable, and cost-efficient even under heavy user demand.