RAG in Elixir: Semantic Search with pgvector and Postgres

Most tutorials for retrieval-augmented generation assume you are writing Python. You reach for a dedicated vector database, a framework with a hundred abstractions, and a stack that has nothing to do with the app you actually ship. If your product runs on Phoenix, that is a lot of moving parts to bolt on for one feature.

The good news: you do not need any of it. A pgvector Elixir RAG pipeline runs entirely on the Postgres database you already have, queried with the Ecto you already know, on a runtime built for exactly this kind of concurrent, IO-bound work. This guide walks through the whole thing end to end: what RAG is, how to install pgvector, how to generate embeddings with the OpenAI API through Req, how to store and query vectors with Ecto, and how to wire the results into a LiveView so users can ask questions of your own documents.

This post is part of our guide to building AI apps with Elixir and Phoenix. If you have already read how to stream OpenAI responses in Phoenix LiveView, this is the natural next step: give the model your data before it answers.

What RAG actually is (and why it matters)

Retrieval-augmented generation is a plain idea wrapped in an intimidating acronym. A large language model only knows what was in its training data. It does not know your internal docs, your product changelog, or the support tickets you closed last week. RAG fixes that in three steps:

1. Retrieve. Take the user’s question, find the most relevant chunks of your content.
2. Augment. Paste those chunks into the prompt as context.
3. Generate. Ask the model to answer using that context.

The retrieval step is where vectors come in. You convert each chunk of your content into an embedding, which is a list of floating-point numbers that captures its meaning. You do the same to the question. Chunks whose vectors sit closest to the question’s vector are, semantically, the most relevant. That is semantic search: matching on meaning rather than keywords, so “how do I cancel my plan” finds a document titled “Ending your subscription” even though they share no words.

You need somewhere to store those vectors and a fast way to find the nearest ones. That is what pgvector gives Postgres.

Why Postgres and pgvector, not a dedicated vector database

The reflex is to add Pinecone, Weaviate, or Qdrant. For most Phoenix apps that is premature. pgvector is a Postgres extension that adds a native vector column type and nearest-neighbor operators. That means:

• One database. Your vectors live next to your documents and users tables. You can join them, filter by user_id, and wrap writes in the same transaction. No second system to sync, back up, or pay for.
• Real indexes. pgvector supports HNSW and IVFFlat indexes for approximate nearest-neighbor search that stays fast into the millions of rows.
• Ecto all the way down. Queries are just Ecto with a custom operator. No new query language.

A dedicated vector database earns its keep at very large scale or when you need features Postgres lacks. Be honest about where you are: if you have fewer than a few million chunks, pgvector on the Postgres you are already running is the simplest thing that works, and Elixir’s concurrency model means the embedding calls that dominate your latency run in parallel for free.

Step 1: Install the pgvector extension

First, make sure the extension is available in your Postgres build. On a managed host like Fly.io Postgres, Supabase, or Neon it is usually one click or already present. Locally with Homebrew:

brew install pgvector

Then enable it inside your database with an Ecto migration. This runs CREATE EXTENSION and is safe to run once per database:

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

  def up do
    execute "CREATE EXTENSION IF NOT EXISTS vector"
  end

  def down do
    execute "DROP EXTENSION vector"
  end
end

Add the pgvector Elixir library to your deps so Ecto understands the type:

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

Register the type with your repo’s Postgrex types module so Ecto can load and dump vector columns:

# 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

Step 2: A migration and schema for document chunks

Store your content in chunks, not whole documents. A chunk of a few hundred words gives the model tight, relevant context and keeps you under token limits. Each chunk gets a vector column sized to your embedding model’s dimensions. OpenAI’s text-embedding-3-small returns 1536 dimensions, so:

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

  def change do
    create table(:document_chunks) do
      add :content, :text, null: false
      add :embedding, :vector, size: 1536
      add :document_id, references(:documents, on_delete: :delete_all)
      add :metadata, :map, default: %{}

      timestamps()
    end

    create index(:document_chunks, [:document_id])
  end
end

Add an HNSW index for fast approximate search. vector_cosine_ops matches the cosine-distance operator you will query with:

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

  def change do
    execute(
      "CREATE INDEX document_chunks_embedding_idx ON document_chunks
       USING hnsw (embedding vector_cosine_ops)",
      "DROP INDEX document_chunks_embedding_idx"
    )
  end
end

The Ecto schema uses Pgvector.Ecto.Vector as the field type:

defmodule MyApp.Knowledge.DocumentChunk do
  use Ecto.Schema
  import Ecto.Changeset

  schema "document_chunks" do
    field :content, :string
    field :embedding, Pgvector.Ecto.Vector
    field :metadata, :map, default: %{}
    belongs_to :document, MyApp.Knowledge.Document

    timestamps()
  end

  def changeset(chunk, attrs) do
    chunk
    |> cast(attrs, [:content, :embedding, :metadata, :document_id])
    |> validate_required([:content, :embedding])
  end
end

Step 3: Generate embeddings with the OpenAI API and Req

You do not need an SDK. The OpenAI embeddings endpoint is one HTTP POST, and Req makes it clean. Here is a small client that turns text into a vector:

defmodule MyApp.Embeddings do
  @endpoint "https://api.openai.com/v1/embeddings"
  @model "text-embedding-3-small"

  @doc "Return the embedding vector for a single string of text."
  def embed(text) when is_binary(text) do
    case embed_all([text]) do
      {:ok, [vector]} -> {:ok, vector}
      other -> other
    end
  end

  @doc "Embed a list of strings in one API call. Returns vectors in order."
  def embed_all(texts) when is_list(texts) do
    Req.post(@endpoint,
      auth: {:bearer, api_key()},
      json: %{model: @model, input: texts}
    )
    |> case do
      {:ok, %{status: 200, body: body}} ->
        vectors =
          body["data"]
          |> Enum.sort_by(& &1["index"])
          |> Enum.map(& &1["embedding"])

        {:ok, vectors}

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

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

  defp api_key, do: System.fetch_env!("OPENAI_API_KEY")
end

Batching matters. The embeddings endpoint accepts many strings per request, so embed_all/1 embeds a whole document’s chunks in one round trip instead of one call per chunk. When you do have independent documents to embed, fan them out with Task.async_stream/3 and let the BEAM run the network calls concurrently:

def embed_documents(documents) do
  documents
  |> Task.async_stream(&embed_document/1,
    max_concurrency: 8,
    timeout: 30_000
  )
  |> Enum.to_list()
end

This concurrency is where Elixir quietly wins. In a single-threaded runtime you either block on each request or reach for extra machinery to parallelize IO. On the BEAM it is one function call, and the scheduler keeps every request in flight without you thinking about threads.

Step 4: Chunk and ingest your content

Before embedding, split each document into overlapping chunks. Overlap keeps a sentence that straddles a boundary from losing its context. A naive but effective splitter:

defmodule MyApp.Chunker do
  @chunk_size 1000
  @overlap 200

  def split(text) do
    text
    |> String.graphemes()
    |> Enum.chunk_every(@chunk_size, @chunk_size - @overlap, :discard)
    |> Enum.map(&Enum.join/1)
  end
end

For real documents (Markdown, HTML, or PDF text) you will want to split on paragraph or heading boundaries rather than raw character counts, but the shape is the same. Ingestion then embeds each chunk and inserts it:

defmodule MyApp.Knowledge do
  alias MyApp.{Repo, Chunker, Embeddings}
  alias MyApp.Knowledge.DocumentChunk

  def ingest(document_id, text) do
    chunks = Chunker.split(text)

    with {:ok, vectors} <- Embeddings.embed_all(chunks) do
      rows =
        Enum.zip(chunks, vectors)
        |> Enum.map(fn {content, vector} ->
          %{
            content: content,
            embedding: vector,
            document_id: document_id,
            inserted_at: {:placeholder, :now},
            updated_at: {:placeholder, :now}
          }
        end)

      {count, _} =
        Repo.insert_all(DocumentChunk, rows,
          placeholders: %{now: DateTime.utc_now()}
        )

      {:ok, count}
    end
  end
end

insert_all/3 writes every chunk in a single statement, which is far faster than inserting them one at a time.

Step 5: Query the nearest chunks with Ecto

Now the payoff. To answer a question, embed it, then ask Postgres for the chunks whose vectors are closest. pgvector exposes cosine distance through the <=> operator, and the pgvector Elixir library gives you Pgvector.Ecto.Query helpers so you can express it in Ecto:

defmodule MyApp.Knowledge do
  import Ecto.Query
  import Pgvector.Ecto.Query
  alias MyApp.{Repo, Embeddings}
  alias MyApp.Knowledge.DocumentChunk

  @doc "Return the top-k most relevant chunks for a question."
  def search(question, k \\ 5) do
    with {:ok, query_vector} <- Embeddings.embed(question) do
      results =
        from(c in DocumentChunk,
          order_by: cosine_distance(c.embedding, ^query_vector),
          limit: ^k,
          select: %{content: c.content, metadata: c.metadata}
        )
        |> Repo.all()

      {:ok, results}
    end
  end
end

Because this is ordinary Ecto, you can add a where to scope results to one tenant, one document, or one user. That single line is the reason keeping vectors in Postgres pays off: multi-tenant filtering and semantic search in the same query, no data leaving the database.

from(c in DocumentChunk,
  where: c.document_id in ^allowed_ids,
  order_by: cosine_distance(c.embedding, ^query_vector),
  limit: ^k
)

Step 6: Build the augmented prompt and generate an answer

With the top chunks in hand, assemble a prompt that instructs the model to answer only from the provided context, then call the chat completions endpoint. Grounding the model in retrieved context is the single most effective way to cut hallucinations:

defmodule MyApp.RAG do
  alias MyApp.{Knowledge}

  def answer(question) do
    with {:ok, chunks} <- Knowledge.search(question, 5) do
      context =
        chunks
        |> Enum.map(& &1.content)
        |> Enum.join("\n\n---\n\n")

      messages = [
        %{role: "system", content: system_prompt()},
        %{role: "user", content: build_prompt(context, question)}
      ]

      chat(messages)
    end
  end

  defp system_prompt do
    "You are a helpful assistant. Answer using only the context provided. " <>
      "If the answer is not in the context, say you do not know."
  end

  defp build_prompt(context, question) do
    """
    Context:
    #{context}

    Question: #{question}
    """
  end

  defp chat(messages) do
    Req.post("https://api.openai.com/v1/chat/completions",
      auth: {:bearer, System.fetch_env!("OPENAI_API_KEY")},
      json: %{model: "gpt-4o-mini", messages: messages}
    )
    |> case do
      {:ok, %{status: 200, body: body}} ->
        {:ok, get_in(body, ["choices", Access.at(0), "message", "content"])}

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

      error ->
        error
    end
  end
end

Step 7: Wire it into a LiveView

The last piece is a UI. A LiveView is a great fit because retrieval and generation are IO-bound, and you want the interface responsive while they run. Kick the work off in an async assign so the process is never blocked:

defmodule MyAppWeb.AskLive do
  use MyAppWeb, :live_view
  alias MyApp.RAG

  def mount(_params, _session, socket) do
    {:ok, assign(socket, question: "", answer: nil, loading: false)}
  end

  def handle_event("ask", %{"question" => question}, socket) do
    socket =
      socket
      |> assign(loading: true, answer: nil)
      |> start_async(:answer, fn -> RAG.answer(question) end)

    {:noreply, socket}
  end

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

  def handle_async(:answer, {:ok, {:error, _reason}}, socket) do
    {:noreply,
     socket
     |> assign(loading: false)
     |> put_flash(:error, "Something went wrong. Please try again.")}
  end

  def render(assigns) do
    ~H"""
    <form phx-submit="ask">
      <input type="text" name="question" placeholder="Ask a question..." />
      <button disabled={@loading}>{if @loading, do: "Thinking...", else: "Ask"}</button>
    </form>

    <div :if={@answer} class="answer">{@answer}</div>
    """
  end
end

start_async and handle_async keep the LiveView process free while OpenAI works, and the page updates the moment the answer lands. If you want the answer to appear token by token instead of all at once, combine this with the approach in our post on streaming OpenAI responses in Phoenix LiveView.

Production notes and honest tradeoffs

A working demo is not a production system. A few things to plan for:

• Cost and caching. Every question costs one embedding call plus one chat call. Embeddings are cheap, but cache the embeddings of your content (you only compute them once at ingest) and consider caching answers to common questions.
• Re-embedding on model changes. If you switch embedding models, the dimensions and the vector space change. You must re-embed your whole corpus. Store the model name in metadata so you know what a vector was built with.
• Index tuning. HNSW has build-time parameters (m, ef_construction) and a query-time ef_search that trade recall for speed. The defaults are fine to start; tune only when you can measure.
• Chunking quality. Retrieval is only as good as your chunks. Splitting on semantic boundaries beats fixed character counts, and adding a title or source to each chunk’s text often improves relevance.
• When to graduate. If you cross tens of millions of vectors or need specialized filtering at scale, a dedicated vector database may be worth the operational cost. Below that, pgvector keeps winning on simplicity.

Skip the plumbing

Everything above is a few hundred lines you can own. If you would rather start from a working AI app and adapt it, that is exactly what our templates ship. pdfai is a Phoenix app that already reads documents, embeds them, and answers questions in a LiveView, and phx_ai_document extends the same pattern to structured document AI. Both save you the ingestion, embedding, and retrieval wiring so you can focus on your own content. If you want to see how the pieces fit before writing your own, read our introduction to pdfai.

PhxTemplates is the AI-native, use-case-specific Phoenix starter, at a lifetime price. A Builder Pass unlocks every template, including the AI ones, so you can ship a real RAG feature this week instead of assembling it from scratch.

Wrapping up

A pgvector Elixir RAG pipeline is less work than the Python-first tutorials suggest, and it fits Phoenix perfectly. You installed one Postgres extension, added a vector column, embedded your content with Req in concurrent batches, queried the nearest chunks with plain Ecto, and served answers from a LiveView that stays responsive under load. No extra database, no heavy framework, no new query language. The whole retrieval layer is Postgres you already run, and the concurrency is the BEAM doing what it was built to do. Point it at your own docs and you have semantic search and grounded answers, all in the stack you already ship.