DEV Community

BAOFUFAN
BAOFUFAN

Posted on

Semantic Caching in RAG: Slash LLM API Calls by 90% (From 500K to 50K)

#programming #python

I was jolted awake at 3 AM by an alert. Bleary‑eyed, I opened the dashboard and saw OpenAI API calls skyrocketing—five minutes had burned through our entire daily budget. The reason was simple: a trending topic overwhelmed our customer support RAG system. Users were asking the same question in many different ways, but our cache relied on exact string matching, completely missing these paraphrased queries.

Problem breakdown: it’s not that we had no cache—our cache was just too “dumb”

If you run a RAG pipeline, you know the biggest LLM cost factor isn’t token consumption—it’s repeated answers. A user asking “What’s the return policy?” and “How do I send something back?” is essentially the same intent. Traditional caching uses string hashing, so even a slight phrasing change causes a miss and fires off another expensive LLM call. Our situation looked like this:

  • Daily call volume: 500K, and 80% were semantic duplicates.
  • We only had simple Redis MD5 caching — hit rate under 15%.
  • Even with a prompt‑normalizing layer, tons of “different words, same meaning” queries still leaked through.

The root cause was obvious: we needed a cache that understands semantic similarity, not just string‑level deduplication. Standard solutions like Redis exact‑key matching or Bloom filters only catch duplicate strings, powerless against paraphrasing and synonym swaps. Running brute‑force embedding comparisons in the application layer would have crushed latency and memory under production traffic.

Architecture: why we chose LangChain + Redis vector search

The goal was simple: put a semantic cache right at the front of the RAG pipeline. Every incoming question is checked against previously seen questions. If we find a near match, we return the stored answer and skip the LLM entirely.

We evaluated a few paths:

  • FAISS + in‑memory dictionary: Great for a single machine, but in a multi‑instance microservice setup the cache can’t be shared, and rolling updates wipe data.
  • Vector databases like Milvus / Pinecone: Overkill. Introducing a brand‑new dependency just for caching adds too much operational overhead.
  • Redis + RediSearch vector index: Our team already used Redis for basic caching. Enabling the vector module lets us store embeddings and run KNN searches with sub‑millisecond latency—perfect for the high‑throughput, low‑latency demands of a semantic cache.

By implementing a custom SemanticRedisCache that plugs into LangChain’s BaseCache interface, we can store prompt embeddings in a Redis vector index. At query time we perform an approximate nearest‑neighbor search. If the similarity exceeds a threshold, we return the cached answer immediately — LLM calls drop off a cliff.

Core implementation: custom LangChain cache + Redis vector index

Step 1: Set up the Redis vector index

The following command creates a Redis index that supports vector search. It stores the prompt and answer as hash fields and uses the embedding field for similarity lookups.

# 使用 Redis Stack 镜像（自带 RediSearch）
docker run -d --name redis-stack -p 6379:6379 redis/redis-stack:latest

redis-cli FT.CREATE idx:semantic_cache ON HASH PREFIX 1 cache: SCHEMA \
  prompt TEXT \
  answer TEXT \
  embedding VECTOR FLAT 6 TYPE FLOAT32 DIM 384 DISTANCE_METRIC COSINE
Enter fullscreen mode Exit fullscreen mode

We chose FLAT brute‑force search because a cache size on the order of hundreds of thousands can easily handle K=1 queries. DIM 384 matches the output dimension of the all-MiniLM-L6-v2 model we use later.

Step 2: Custom LangChain cache class

This code shows how to make LangChain look up the cache by semantic meaning rather than by string equality. It inherits from BaseCache, overrides lookup and update, and performs a vector nearest‑neighbor search in Redis for every prompt.

import hashlib
import json
import numpy as np
from typing import Optional
from langchain_core.caches import BaseCache
from langchain_core.outputs import Generation
from sentence_transformers import SentenceTransformer
import redis
from redis.commands.search.query import Query

class SemanticRedisCache(BaseCache):
    def __init__(self, redis_url="redis://localhost:6379", threshold=0.9):
        self.redis = redis.from_url(redis_url)
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.threshold = threshold
        self.index_name = "idx:semantic_cache"
        self.dim = 384

    def _embed(self, text: str) -> bytes:
        """Generate the vector and convert it to the binary format required by Redis"""
        vec = self.model.encode(text, normalize_embeddings=True)
        return vec.astype(np.float32).tobytes()

    def lookup(self, prompt: str, llm_string: str = "") -> Optional[list[Generation]]:
        """
        Semantic lookup: convert the prompt into a vector, query the nearest neighbour
        in the Redis index, and return the cached answer if cosine similarity >= threshold.
        """
        query_vector = self._embed(prompt)
        # RediSearch KNN query: K=1, return similarity score and answer
        q = (
            Query("*=>[KNN 1 @embedding $vec AS score]")
            .sort_by("score")
            .return_fields("answer", "score")
            .dialect(2)
        )
        results = self.redis.ft(self.index_name)
Enter fullscreen mode Exit fullscreen mode

In practice, you would execute the query against the index, compare the returned score against the threshold, and either serve the cached answer or let the request proceed to the LLM—all without touching OpenAI until truly necessary. This one change took our LLM call volume from 500K to 50K and slashed costs by 90%.

Top comments (0)