AI Semantic Caching

Intercept costly LLM calls by checking SwarmV for semantically similar prompts first.

Overview

Instead of paying for identical or very similar queries to your LLM, you can use SwarmV to cache completions. Because SwarmV is a vector database, it can match queries that mean the same thing, even if the exact phrasing differs.

TypeScript / OpenAI SDK

import { SwarmV, SemanticCache } from "@swarmv/sdk";
import OpenAI from "openai";

const swarmv = new SwarmV();   // backend: SWARMV_API_KEY from the environment
const openai = new OpenAI();

const embed = async (text: string) =>
  (await openai.embeddings.create({ model: "text-embedding-3-small", input: text })).data[0].embedding;

// threshold = minimum cosine similarity for a hit. Responses are stored with their vectors
// (up to ~8 KB each), so every instance sharing the index shares the cache.
// Optional: { store: myRedisStore } keeps responses elsewhere instead.
const cache = new SemanticCache(swarmv, embed, 0.95);

const answer = await cache.call("How do I reset my password?", async (prompt) => {
  const completion = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: prompt }],
  });
  return completion.choices[0].message.content ?? "";
});

Distances and similarity

If you build your own cache logic, note that search returns a distance, not a similarity score.

TypeScript

// Search returns { id, distance, metadata } for the closest vectors (lower distance = closer).
// For the default cosine index, convert a distance to similarity like this:
import { cosineSimilarity } from "@swarmv/sdk";

const [best] = await swarmv.search(promptVector, { topK: 1 });
if (best && cosineSimilarity(best.distance) >= 0.95) {
  // close enough: use what you stored with it, e.g. best.metadata?.cached_response
}

Python

from swarmv import cosine_similarity

best = swarmv.search(prompt_vector, top_k=1)
if best and cosine_similarity(best[0].distance) >= 0.95:
    ...  # close enough: use (best[0].metadata or {}).get("cached_response")

Databricks (MosaicML / Model Serving)

Cache Foundation Model API calls across your Databricks environments.

Python / Databricks (MLflow)

from mlflow.deployments import get_deploy_client
from swarmv import SwarmV, SemanticCache

client = get_deploy_client("databricks")
index = SwarmV()  # backend: SWARMV_API_KEY from the environment

def databricks_llm(prompt: str) -> str:
    res = client.predict(
        endpoint="databricks-meta-llama-3-70b-instruct",
        inputs={"messages": [{"role": "user", "content": prompt}]},
    )
    return res["choices"][0]["message"]["content"]

def embed(prompt: str):
    res = client.predict(endpoint="databricks-bge-large-en", inputs={"input": prompt})
    return res["data"][0]["embedding"]

# threshold = minimum cosine similarity. Responses are stored with their vectors, so the cache
# is shared by every process using this index. Optional: store=<obj with get/set>.
cache = SemanticCache(index, embed, threshold=0.95)

# Databricks is only called when SwarmV has no similar prompt cached
response = cache.call("Explain semantic caching", databricks_llm)