Search⌘ K
AI Features
Home
Courses
AI for Backend Engineers

Caching and Idempotency

Explore how to build effective caching layers and implement idempotency to handle non-deterministic AI model outputs and prevent duplicate request processing. Learn to create cache keys that reflect model and prompt versions for accurate results and distinguish between cached responses and repeated submissions to ensure consistent backend behavior.

We'll cover the following...

“Where AI Fits into Backend Work” introduced an important caching constraint earlier in this course: caching an LLM call does not offer the same straightforward benefit as caching a deterministic function because the same input can produce different valid outputs across model calls. We deferred implementing the cache because the implementation depends on the validated schema and processing pipeline developed in later lessons. This lesson implements that caching layer and also introduces idempotency, a related but distinct concern.

Already configured setup code

The block below is carried over so this file runs entirely on its own: the schema a valid result must match, the prompt builder, the code fence stripper, a call_llm that runs with no API key using a mock response by default, and the pipeline tying it all together.

import json
from enum import Enum
from pydantic import BaseModel, Field, ValidationError
from datetime import datetime, timezone
class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    NEGATIVE = "negative"
class FeedbackCategory(str, Enum):
    PRODUCT_QUALITY = "product_quality"
    CUSTOMER_SERVICE = "customer_service"
    PRICING = "pricing"
    OTHER = "other"
class Confidence(str, Enum):
    HIGH = "high"
    LOW = "low"
class Feedback(BaseModel):
    sentiment: Sentiment
    category: FeedbackCategory
    severity: int = Field(ge=1, le=5, default=1)
    confidence: Confidence
    summary: str
class StoredFeedbackResult(BaseModel):
    raw_response: str
    parsed: Feedback
    model_name: str
    prompt_version: str
    created_at: datetime
MODEL_NAME = "your-chosen-provider-model"
PROMPT_VERSION = "feedback-v3"
def build_feedback_prompt(feedback_text: str) -> str:
    return (
        "Analyze the customer feedback below. Respond with a JSON object "
        "in exactly this shape, and nothing else:\n\n"
        "{\n"
        '  "sentiment": <one of: "positive", "neutral", "negative">,\n'
        '  "category": <one of: "product_quality", "customer_service", '
        '"pricing", "other">,\n'
        '  "severity": <integer from 1 to 5>,\n'
        '  "confidence": <one of: "high", "low">,\n'
        '  "summary": <string, one short sentence>\n'
        "}\n\n"
        f'Feedback: "{feedback_text}"'
    )
def build_repair_prompt(original_prompt: str, bad_response: str, error: ValidationError) -> str:
    return (
        f"{original_prompt}\n\nYour previous response did not match the "
        f"required format. Previous response:\n{bad_response}\n\n"
        f"Validation error:\n{error}\n\nRespond again with only the "
        "corrected JSON object."
    )
def strip_code_fence(text: str) -> str:
    stripped = text.strip()
    if stripped.startswith("```"):
        lines = stripped.split("\n")[1:]
        if lines and lines[-1].strip() == "```":
            lines = lines[:-1]
        return "\n".join(lines).strip()
    return stripped
# USE_MOCK is True by default, so this file runs immediately with no
# API key. Set it to False and fill in a real provider block below to
# use an actual model.
USE_MOCK = True
def call_llm(prompt: str) -> str:
    if USE_MOCK:
        return (
            '{"sentiment": "negative", "category": "customer_service", '
            '"severity": 4, "confidence": "high", '
            '"summary": "Support took four days to respond."}'
        )
    # from openai import OpenAI
    # client = OpenAI(api_key="your-openai-api-key")
    # response = client.chat.completions.create(
    #     model="your-chosen-openai-model",
    #     messages=[{"role": "user", "content": prompt}],
    # )
    # return response.choices[0].message.content
    raise NotImplementedError("Fill in a real provider block, or set USE_MOCK back to True.")
def get_validated_feedback(feedback_text: str, max_attempts: int = 2) -> Feedback:
    prompt = build_feedback_prompt(feedback_text)
    current_prompt = prompt
    for attempt in range(1, max_attempts + 1):
        raw_response = call_llm(current_prompt)
        cleaned = strip_code_fence(raw_response)
        try:
            parsed = json.loads(cleaned)
        except json.JSONDecodeError:
            continue
        try:
            return Feedback.model_validate(parsed)
        except ValidationError as error:
            current_prompt = build_repair_prompt(prompt, cleaned, error)
    return Feedback(
        sentiment=Sentiment.NEUTRAL,
        category=FeedbackCategory.OTHER,
        severity=3,
        confidence=Confidence.LOW,
        summary="Could not be automatically assessed.",
    )
def build_stored_result(raw_response: str, parsed: Feedback) -> StoredFeedbackResult:
    return StoredFeedbackResult(
        raw_response=raw_response,
        parsed=parsed,
        model_name=MODEL_NAME,
        prompt_version=PROMPT_VERSION,
        created_at=datetime.now(timezone.utc),
    )

What a cache key needs to represent

This is what the lesson is actually about. A cache is only useful if identical future requests can find what an earlier identical request already produced. The tricky part with a paragraph of customer feedback is that the text itself, word for word, is the input, and using the entire paragraph directly as a lookup key is workable but clumsy, since keys that long are awkward to compare, store, and index efficiently.

import hashlib


def build_cache_key(text: str, model_name: str, prompt_version: str) -> str:
    combined = f"{model_name}:{prompt_version}:{text}"
    return hashlib.sha256(combined.encode("utf-8")).hexdigest()


key_one = build_cache_key(
    "Support took four days to respond to my ticket.",
    model_name=MODEL_NAME,
    prompt_version="feedback-v3",
)
key_two = build_cache_key(
    "Support took four days to respond to my ticket.",
    model_name=MODEL_NAME,
    prompt_version="feedback-v4",
)

print("Key one:", key_one)
print("Key two:", key_two)
print("Same key?", key_one == key_two)
A cache key built from more than just the text itself
...