Semantic Caching in Elixir: Cut LLM Costs with pgvector

Every AI feature in production ends up answering the same question many times. “How do I reset my password?”, “how can I reset my password”, “reset password steps”: three prompts, three paid completions, three multi-second waits, and one answer. A plain cache keyed on the exact string catches none of the second and third. Semantic caching does: you embed the prompt, look for a previous prompt that means the same thing, and return its stored answer instead of calling the model.

This post builds semantic caching in Elixir from scratch with Postgres, pgvector, Ecto and Req. It is part of our series on building AI apps with Elixir and Phoenix, and it reuses the same pgvector setup, so if you already have RAG running you are halfway there.

By the end you will have:

• a two-layer cache (exact match first, then nearest neighbour by cosine distance)
• a scope key so one tenant’s answers never leak to another
• a threshold you tuned on your own data instead of copied from a blog
• telemetry that tells you the hit rate and what the cache saved
• a clear list of prompts you should never cache

Why semantic caching, and when it pays

A chat completion is the most expensive and slowest call in most AI features. An embedding call is a small fraction of its price and usually returns quickly, because the model is tiny and there is no output to generate. So the trade is: pay for one cheap embedding on every request, and skip the expensive completion on every hit.

That trade pays when your traffic repeats itself. Good candidates:

• Support and FAQ bots, where most questions are paraphrases of a few dozen.
• Document Q&A over a fixed corpus, where many users ask the same thing about the same PDF.
• Classification and extraction prompts that see near-identical inputs (the same invoice template, the same product description with a typo fixed).
• Public demo endpoints, where visitors all try the same three example questions.

It does not pay for open-ended creative generation, long multi-turn conversations, or anything personalised. We will come back to that list, because getting it wrong is how a cache starts serving the wrong answer.

Semantic caching vs provider prompt caching

OpenAI and Anthropic both offer prompt caching. That feature discounts the input tokens of a repeated prefix (a long system prompt, a big document) but still runs the model and still bills the output. A semantic cache skips the model entirely. They stack well: prompt caching makes misses cheaper, semantic caching makes hits free.

The design in one paragraph

On each request, normalise the prompt and compute a hash. Look up an exact match for that hash within the request’s scope. If there is none, embed the prompt and ask pgvector for the nearest stored prompt in the same scope. If its cosine distance is under your threshold, return the stored response. Otherwise call the model, store the prompt, its embedding and the response, and return the fresh answer. Expire entries on a TTL and whenever the inputs that shaped the answer change.

Setting up pgvector in Phoenix

Add the dependencies. pgvector gives you the Ecto type and query helpers, and req is the HTTP client we use for OpenAI.

# mix.exs
defp deps do
  [
    {:pgvector, "~> 0.3"},
    {:req, "~> 0.5"},
    # ...your existing deps
  ]
end

Postgrex needs to know about the vector type. Define a types module and point the repo at it:

# lib/my_app/postgrex_types.ex
Postgrex.Types.define(
  MyApp.PostgrexTypes,
  Pgvector.extensions() ++ Ecto.Adapters.Postgres.extensions(),
  []
)

# config/config.exs
config :my_app, MyApp.Repo, types: MyApp.PostgrexTypes

The Postgres server needs the extension installed. Most managed Postgres providers ship it; locally the pgvector/pgvector Docker image is the quickest route.

The cache table

defmodule MyApp.Repo.Migrations.CreateLlmCacheEntries do
  use Ecto.Migration

  def change do
    execute "CREATE EXTENSION IF NOT EXISTS vector", "DROP EXTENSION IF EXISTS vector"

    create table(:llm_cache_entries) do
      add :scope, :string, null: false
      add :prompt_hash, :binary, null: false
      add :prompt, :text, null: false
      add :embedding, :vector, size: 1536, null: false
      add :response, :text, null: false
      add :model, :string, null: false
      add :hit_count, :integer, null: false, default: 0
      add :expires_at, :utc_datetime, null: false

      timestamps(type: :utc_datetime)
    end

    create unique_index(:llm_cache_entries, [:scope, :prompt_hash])
    create index(:llm_cache_entries, [:expires_at])

    execute """
            CREATE INDEX llm_cache_entries_embedding_idx
            ON llm_cache_entries USING hnsw (embedding vector_cosine_ops)
            """,
            "DROP INDEX llm_cache_entries_embedding_idx"
  end
end

size: 1536 matches OpenAI’s text-embedding-3-small. If you use a different embedding model, change it to that model’s dimension. The HNSW index uses vector_cosine_ops because we will query by cosine distance.

The schema

defmodule MyApp.AI.CacheEntry do
  use Ecto.Schema

  schema "llm_cache_entries" do
    field :scope, :string
    field :prompt_hash, :binary
    field :prompt, :string
    field :embedding, Pgvector.Ecto.Vector
    field :response, :string
    field :model, :string
    field :hit_count, :integer, default: 0
    field :expires_at, :utc_datetime

    timestamps(type: :utc_datetime)
  end
end

Embeddings with Req

One small module wraps the embeddings endpoint. It returns {:ok, list} or {:error, reason} so a flaky API never crashes the caller; on an embedding failure we simply skip the cache and call the model.

defmodule MyApp.AI.Embeddings do
  @model "text-embedding-3-small"

  def embed(text) do
    case Req.post("https://api.openai.com/v1/embeddings",
           auth: {:bearer, api_key()},
           json: %{model: @model, input: text},
           receive_timeout: 10_000
         ) do
      {:ok, %{status: 200, body: %{"data" => [%{"embedding" => vector} | _]}}} ->
        {:ok, vector}

      {:ok, %{status: status, body: body}} ->
        {:error, {:http, status, body}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp api_key, do: Application.fetch_env!(:my_app, :openai_api_key)
end

The cache module

This is the heart of it. fetch/3 takes a scope, the prompt, and a zero-arity function that calls the model. The caller does not need to know whether the answer came from the cache.

defmodule MyApp.AI.SemanticCache do
  import Ecto.Query
  import Pgvector.Ecto.Query

  alias MyApp.Repo
  alias MyApp.AI.{CacheEntry, Embeddings}

  @threshold 0.08
  @ttl_seconds 7 * 24 * 3600

  def fetch(scope, prompt, generate_fun) when is_function(generate_fun, 0) do
    normalised = normalise(prompt)
    hash = :crypto.hash(:sha256, normalised)

    case exact(scope, hash) do
      %CacheEntry{} = entry ->
        hit(entry, :exact, 0.0)

      nil ->
        semantic_fetch(scope, normalised, hash, generate_fun)
    end
  end

  defp semantic_fetch(scope, normalised, hash, generate_fun) do
    case Embeddings.embed(normalised) do
      {:ok, vector} ->
        case nearest(scope, vector) do
          {%CacheEntry{} = entry, distance} when distance <= @threshold ->
            hit(entry, :semantic, distance)

          _miss ->
            miss(scope, normalised, hash, vector, generate_fun)
        end

      {:error, _reason} ->
        # The cache is an optimisation; never let it block an answer.
        with {:ok, response, _model} <- generate_fun.(), do: {:ok, response}
    end
  end

  defp exact(scope, hash) do
    now = DateTime.utc_now()

    Repo.one(
      from e in CacheEntry,
        where: e.scope == ^scope and e.prompt_hash == ^hash and e.expires_at > ^now
    )
  end

  defp nearest(scope, vector) do
    now = DateTime.utc_now()
    vector = Pgvector.new(vector)

    Repo.one(
      from e in CacheEntry,
        where: e.scope == ^scope and e.expires_at > ^now,
        order_by: cosine_distance(e.embedding, ^vector),
        limit: 1,
        select: {e, cosine_distance(e.embedding, ^vector)}
    )
  end

  defp hit(entry, kind, distance) do
    Repo.update_all(from(e in CacheEntry, where: e.id == ^entry.id), inc: [hit_count: 1])
    emit(:hit, kind, distance)
    {:ok, entry.response}
  end

  defp miss(scope, normalised, hash, vector, generate_fun) do
    emit(:miss, :semantic, nil)

    with {:ok, response, model} <- generate_fun.() do
      now = DateTime.utc_now() |> DateTime.truncate(:second)

      Repo.insert(
        %CacheEntry{
          scope: scope,
          prompt_hash: hash,
          prompt: normalised,
          embedding: Pgvector.new(vector),
          response: response,
          model: model,
          expires_at: DateTime.add(now, @ttl_seconds)
        },
        on_conflict: :nothing,
        conflict_target: [:scope, :prompt_hash]
      )

      {:ok, response}
    end
  end

  defp normalise(prompt) do
    prompt
    |> String.downcase()
    |> String.replace(~r/\s+/u, " ")
    |> String.trim()
  end

  defp emit(result, kind, distance) do
    :telemetry.execute(
      [:my_app, :semantic_cache, :lookup],
      %{count: 1, distance: distance || -1.0},
      %{result: result, kind: kind}
    )
  end
end

A few decisions worth explaining:

• Exact match first. A hash lookup on a unique index costs nothing and skips the embedding call entirely for literal repeats, which are more common than you would expect (retries, double submits, the demo button everyone clicks).
• The embedding is computed once. On a miss we reuse the vector we already paid for when we store the new entry.
• on_conflict: :nothing. Two users asking the same new question at the same moment will both miss and both generate. The second insert quietly does nothing instead of raising.
• Failure is transparent. If the embeddings API is down, the request still gets an answer. It just costs full price.

generate_fun returns {:ok, response, model} (or {:error, reason}, which with passes straight back to the caller) so we can record which model produced each cached answer. When you upgrade models, that column tells you which entries are stale.

Scope: the part that keeps you out of trouble

The scope string decides which cached answers a request is allowed to see. Get it too broad and one customer sees an answer computed from another customer’s data. Get it too narrow and the hit rate collapses.

Build it from everything that changes the answer other than the question itself:

def scope(%{tenant_id: tenant_id}, opts) do
  Enum.join(
    [
      "tenant:#{tenant_id}",
      "model:#{opts[:model]}",
      "prompt:v#{opts[:prompt_version]}",
      "corpus:#{opts[:corpus_version] || "none"}"
    ],
    "|"
  )
end

• Tenant. Anything that reads private data is scoped per tenant. Only a truly public FAQ can share one scope across everybody.
• Model. A cheaper model’s answer should not be served to a request that paid for the stronger one.
• System prompt version. Bump it whenever you edit the system prompt, and every old answer falls out of scope at once, with no delete required.
• Corpus version. For RAG, the answer depends on the documents. A hash of the document ids and their updated_at values works, or a counter you bump on every upload.

A pgvector detail about filtered queries

HNSW finds approximate neighbours first and applies your WHERE clause afterwards. With many small scopes in one big table, the index can return candidates from other scopes, filter them all out, and hand you no result even though a good match exists. Three fixes, in order of effort: raise hnsw.ef_search for the session, enable iterative index scans (pgvector 0.8 added hnsw.iterative_scan), or create a partial index per large scope. For a table of a few hundred thousand rows, a plain index plus a sensible ef_search is usually enough. Measure before you get clever.

Tuning the threshold on your own data

0.08 above is a starting point, not an answer. Cosine distance between embeddings depends on the embedding model and on your domain, and the cost of a wrong hit is much higher than the cost of a miss. So tune it.

Collect a few dozen pairs of real prompts from your logs and label each pair as “same answer” or “different answer”. Then measure:

defmodule MyApp.AI.ThresholdCheck do
  alias MyApp.AI.Embeddings

  def run(pairs) do
    for {a, b, label} <- pairs do
      {:ok, va} = Embeddings.embed(a)
      {:ok, vb} = Embeddings.embed(b)
      {label, Float.round(cosine_distance(va, vb), 4), a, b}
    end
    |> Enum.sort_by(fn {_label, d, _a, _b} -> d end)
  end

  defp cosine_distance(a, b) do
    dot = Enum.zip_with(a, b, &(&1 * &2)) |> Enum.sum()
    norm = fn v -> :math.sqrt(Enum.reduce(v, 0.0, &(&1 * &1 + &2))) end
    1.0 - dot / (norm.(a) * norm.(b))
  end
end

Sort by distance and look for the smallest distance at which a “different answer” pair appears. Set your threshold comfortably below it. The dangerous pairs are the ones that differ by one word: “how do I cancel my Pro plan” and “how do I cancel my Free plan” can sit very close together while needing different answers. If your domain is full of those, keep the threshold tight and lean on the exact-match layer.

Using it from a Phoenix LiveView

The cache fits naturally into a LiveView that already calls a model asynchronously. On a hit the answer appears almost immediately; on a miss the user waits as before.

def handle_event("ask", %{"question" => question}, socket) do
  scope = MyApp.AI.scope(socket.assigns.current_scope, model: "gpt-4o-mini", prompt_version: 3)

  socket =
    socket
    |> assign(:asking, true)
    |> start_async(:answer, fn ->
      MyApp.AI.SemanticCache.fetch(scope, question, fn ->
        MyApp.AI.Chat.complete(question, model: "gpt-4o-mini")
      end)
    end)

  {:noreply, socket}
end

def handle_async(:answer, {:ok, {:ok, answer}}, socket) do
  {:noreply, assign(socket, asking: false, answer: answer)}
end

def handle_async(:answer, _failure, socket) do
  {:noreply, assign(socket, asking: false, error: "Something went wrong. Try again.")}
end

MyApp.AI.Chat.complete/2 stands in for your existing completion call; it only has to return {:ok, text, model}.

If you stream responses token by token (see streaming OpenAI responses in Phoenix LiveView), keep streaming on a miss and collect the full text as it arrives so you can store it when the stream ends. On a hit you can render the whole answer at once. Faking a typing effect for a cached answer is a product choice; most users prefer the instant answer.

Expiry and invalidation

Three mechanisms keep the cache honest:

1. TTL. Every entry has expires_at. Lookups ignore expired rows, and a nightly job deletes them:

defmodule MyApp.AI.PruneCache do
  use Oban.Worker, queue: :maintenance

  import Ecto.Query

  @impl Oban.Worker
  def perform(_job) do
    now = DateTime.utc_now()
    MyApp.Repo.delete_all(from e in MyApp.AI.CacheEntry, where: e.expires_at < ^now)
    :ok
  end
end

2. Scope versioning. Editing the system prompt or re-indexing documents bumps a version inside the scope, so old entries simply stop matching and age out.

3. Targeted deletes. When someone reports a bad answer, delete by prompt_hash or by scope. Give support a button for it; you will use it.

Measuring what the cache saves

Attach a handler to the telemetry event and send it to wherever your metrics live. With Telemetry.Metrics in a Phoenix app:

# lib/my_app_web/telemetry.ex
def metrics do
  [
    counter("my_app.semantic_cache.lookup.count", tags: [:result, :kind]),
    distribution("my_app.semantic_cache.lookup.distance",
      tags: [:kind],
      reporter_options: [buckets: [0.02, 0.05, 0.08, 0.12, 0.2]]
    )
  ]
end

Hit rate is hits / (hits + misses). Multiply the hits by your average completion cost to get the money saved, and subtract the embedding cost of every non-exact lookup. The distance distribution shows whether your threshold sits in a sensible place: a big cluster of hits right at the threshold edge means you are probably serving borderline matches.

The hit_count column is useful too. Sort by it to see your most common questions. That list is often a better FAQ page than the one you wrote.

What you should never cache

Be strict here. A semantic cache that returns a confidently wrong answer is worse than no cache.

• Multi-turn conversations. The meaning of “and what about the second one?” depends on the conversation, not the text. Either skip the cache after the first turn or include a summary of the context in the prompt you embed.
• Personalised answers. Anything that reads the user’s own account, orders or files. If you must cache it, scope it per user, which usually kills the hit rate.
• Time-sensitive questions. “What’s the status of my deploy”, “what’s new this week”. Use a short TTL or skip them.
• Tool-calling agents with side effects. Caching the final answer of an agent that sent an email means the second user gets told an email was sent when nothing happened.
• Anything where one word changes the answer and your threshold tuning showed you cannot separate the cases.

A simple way to enforce this is to make caching opt-in per call site: only the FAQ bot and the document Q&A pass through SemanticCache.fetch/3; everything else calls the model directly.

Where this fits in a real app

Semantic caching sits in front of whatever AI feature you already have. If you are building document Q&A, the natural home is next to your retrieval code: the pdfai template already wires OpenAI into a Phoenix LiveView app with pgvector and Stripe, so adding a cache table and one module is an afternoon’s work rather than a week of setup.

If your AI feature is also a paid feature, the cache changes your unit economics in a way worth tracking per customer: hits cost you almost nothing while the customer still gets an answer. If you meter AI usage for billing, decide deliberately whether a cached answer counts against the customer’s quota. Aurora Meter meters that kind of usage for Phoenix apps, and the decision is one line either way.

Summary

• Semantic caching returns a stored LLM answer when a new prompt means the same thing as an old one, and skips the completion call entirely.
• In Elixir it is a Postgres table with a vector column, an HNSW index on cosine distance, an exact-match hash in front, and one module that wraps your model call.
• Scope every entry by tenant, model, system prompt version and corpus version, so answers never cross boundaries they should not.
• Tune the threshold on labelled pairs from your own traffic, and keep it tight where one word changes the answer.
• Expire by TTL, invalidate by bumping versions, and measure hit rate and distance with telemetry.
• Only cache prompts that are stateless, impersonal and stable.

If you would rather start from a working AI app than a blank mix phx.new, the Builder Pass gives you every PhxTemplates starter, including the AI ones, for a single lifetime price.