ragdedup

A lightweight semantic cache for LLM and RAG pipelines. One class, one method, zero required dependencies.

ragdedup recognizes when a new request means roughly the same thing as one you already answered, even if the wording is different, and returns the cached response instead of calling the model again. Wrap any embedding model and any vector store behind a single call:

result = cache.get_or_compute(text, llm_fn)

It embeds text, checks similarity against everything already cached, and either returns the cached response or calls llm_fn(text), stores the result, and returns that instead. This is the pattern behind cutting manual email handling time by roughly 75 percent in a production lead generation pipeline: near duplicate messages like "any updates?", "just checking in", and "following up on this" stopped triggering a fresh LLM call every single time.

Why this exists, and where the real competition is

There is already a mature, actively maintained project doing semantic caching for LLMs: GPTCache (by Zilliz). It is the right call if you want maximum configurability.

What ragdedup actually offers that is worth knowing about:

1. Zero required dependencies. The default in-memory backend is pure Python cosine similarity, no numpy needed. Add openai, sentence-transformers, or faiss only when you actually want them.
2. One class, one method. SemanticCache(embedder, vector_store).get_or_compute(text, llm_fn). No cache_manager or similarity_evaluation objects to assemble first.
3. Sync and async from day one. get_or_compute and aget_or_compute both ship in v0.1.0, and aget_or_compute accepts either a sync or an async llm_fn.
4. Built in cost reporting. stats.estimated_cost_saved(cost_per_call) gives you a dollar figure you can put directly into a report.
5. Plugs straight into other LLMClient style interfaces. Since llm_fn is just Callable[[str], str], you can pass any object's .complete method in directly, including agentic_rag_toolkit's LLMClient.

If you need multi-backend flexibility, a large existing community, or out of the box LangChain and LlamaIndex hooks, use GPTCache. Use ragdedup when you want the core idea in the smallest possible footprint.

Install

pip install ragdedup                          # core only, zero extra dependencies
pip install "ragdedup[openai]"                # + real OpenAI embeddings
pip install "ragdedup[sentence-transformers]" # + free local embeddings
pip install "ragdedup[faiss]"                 # + FAISS backed vector store for scale

Quickstart

from ragdedup import SemanticCache, HashEmbedder

def my_llm(text: str) -> str:
    ...  # call your real model here, return the text response

cache = SemanticCache(embedder=HashEmbedder(dim=64), similarity_threshold=0.85)

result = cache.get_or_compute("Any updates on my order?", my_llm)
print(result.was_cached, result.response)

result = cache.get_or_compute("Just checking in on my order", my_llm)
print(result.was_cached)   # True, second call is a semantic match, no LLM call made

print(cache.stats.hit_rate, cache.stats.estimated_cost_saved(cost_per_call=0.002))

HashEmbedder is dependency free and deterministic, good for trying the package out or for tests, but it is not a real semantic model. Swap in OpenAIEmbedder or SentenceTransformerEmbedder for production use. See examples/basic_usage.py (no API keys needed) and examples/openai_faiss_example.py (production shaped, real OpenAI embeddings plus FAISS).

API Reference

SemanticCache, CacheResult, CacheStats (module: ragdedup.cache)

Constructor

SemanticCache(
    embedder: Embedder,
    vector_store: VectorStore | None = None,
    similarity_threshold: float = 0.92,
)

Methods

.get_or_compute(text: str, llm_fn: Callable[[str], str]) -> CacheResult

Raises ValueError if text is empty or blank. Embeds text, checks the vector store for a match at or above similarity_threshold. On a match, returns the cached response without calling llm_fn. On no match, calls llm_fn(text), stores the result, and returns it.

.aget_or_compute(text: str, llm_fn) -> CacheResult

Same behavior, async. llm_fn may be a regular function or an async def function; both are detected and handled automatically.

.clear(reset_stats: bool = False) -> None

Empties the vector store. Pass reset_stats=True to also zero out .stats.

.size() -> int

Number of entries currently cached.

CacheResult fields: response: str, was_cached: bool, similarity: float | None (the match score, None on a miss), matched_text: str | None (the original text that matched, None on a miss).

CacheStats (available as cache.stats): hits: int, misses: int, total (property), hit_rate (property, hits / total), and estimated_cost_saved(cost_per_call: float) -> float.

from ragdedup import SemanticCache, HashEmbedder

cache = SemanticCache(embedder=HashEmbedder(), similarity_threshold=0.9)
result = cache.get_or_compute("what is your refund policy", my_llm)
print(result.response, result.was_cached, result.similarity)
print(cache.stats.hits, cache.stats.hit_rate, cache.stats.estimated_cost_saved(0.002))

Embedder (Protocol), HashEmbedder, OpenAIEmbedder, SentenceTransformerEmbedder (module: ragdedup.embedders)

Any object with .embed(text: str) -> list[float] satisfies Embedder automatically, no subclassing needed.

from ragdedup import OpenAIEmbedder

embedder = OpenAIEmbedder(model="text-embedding-3-small")
vector = embedder.embed("hello world")

VectorStore (Protocol), InMemoryVectorStore, FAISSVectorStore, StoreMatch, cosine_similarity (module: ragdedup.stores)

Any object with .add(), .query(), .size(), .clear() satisfies VectorStore automatically.

InMemoryVectorStore(max_size: int | None = 10_000, ttl_seconds: float | None = None)

Thread safe, dependency free. max_size evicts the oldest entry first once exceeded (this is what makes the cache "rolling"). ttl_seconds expires entries older than that many seconds, checked lazily on the next query. Both raise ValueError if set to a non-positive number. Pass None to disable either limit.

FAISSVectorStore(dim: int)

Requires pip install faiss-cpu. Faster and able to hold far more entries than InMemoryVectorStore. Known limitation: FAISS's flat index does not support deleting individual entries cheaply, so this store has no max_size or ttl_seconds and grows unbounded. Monitor .size() yourself at real scale, or wrap your own Chroma/Pinecone client the same way (see examples/) if you need automatic expiry.

store = InMemoryVectorStore(max_size=5000, ttl_seconds=86400)   # rolling 24 hour cache, max 5000 entries

Methods on both: .add(id, vector, text, metadata=None), .query(vector, top_k=1) -> list[StoreMatch], .size() -> int, .clear() -> None.

StoreMatch fields: id: str, score: float, text: str, metadata: dict.

cosine_similarity(a: list[float], b: list[float]) -> float is the plain function InMemoryVectorStore uses internally, exported in case you want it directly. Raises ValueError if the two vectors are different lengths.

Known limitations

Under concurrent access, two near simultaneous calls with the same brand new (not yet cached) text can both miss the cache and both call llm_fn once, before either result gets stored. This does not cause incorrect behavior, it is just an occasional missed cache hit under heavy concurrent load on genuinely new queries. If you need strict single flight behavior (guarantee llm_fn runs at most once per unique input, even under a race), add your own per key lock in front of get_or_compute.

FAISSVectorStore has no automatic eviction, see above.

HashEmbedder is not a real semantic model. If you use it in production instead of just for demos and tests, you will get false cache hits and misses that a real embedding model would not.

Development

python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
pytest --cov=ragdedup tests/ -v

License

MIT, see LICENSE.