Bygg en produktionsklar RAG-pipeline med pgvector och Claude i Python (2026)

Retrieval-Augmented Generation (RAG) har blivit standardmönstret för att grunda LLM:er i din egen data – och ärligt talat, det finns en anledning till varför alla pratar om det just nu. I den här handledningen bygger vi en produktionsklar RAG-pipeline med pgvector i PostgreSQL, embeddings från en lokal modell och Claude API som svarsmotor. Allt i Python, och utan att blanda in en separat vektordatabas.

Vi täcker det som tutorials från 2024–2025 ofta missar: HNSW-tuning för låg latens, hybrid sökning med BM25, korrekt placering av cache breakpoints i Claude och hur du undviker den nya fem-minuters TTL-fällan i prompt caching som introducerades i början av 2026 (jag gick själv på den fällan i ett kundprojekt i februari, så det här avsnittet är skrivet med ärr).

Varför pgvector + Claude i 2026?

I april 2026 har pgvector passerat 8 miljoner installationer och är, så vitt jag kan se, de facto-standarden för vektorsökning i Postgres. Anledningen är ganska enkel: du behöver inte ytterligare en databas i din stack. Du kan filtrera på tenant_id, datum eller permissions med vanlig SQL och köra ANN-sökning i samma transaktion.

Claude (Sonnet 4.6 och Opus 4.7) hanterar långa kontexter exceptionellt bra och har förstklassig prompt caching. För RAG betyder det att du kan cacha både systemprompten och hämtade dokument – vilket sänker input-kostnaden med 60–90 % på återkommande anrop. Det är inte småpengar när du kör i produktion.

När pgvector är rätt val

• Du har redan PostgreSQL i produktion.
• Datasetet är under ~10 miljoner vektorer per tabell.
• Du behöver hybrid filter (vektor + WHERE-klausuler på metadata).
• Multi-tenant SaaS där du vill partitionera per kund.

När du bör titta på alternativ

• Hundratals miljoner vektorer eller sub-millisekunds-latens i stor skala – då passar Qdrant, Milvus eller Weaviate bättre.
• Du behöver inbyggd cross-encoder re-ranking i samma databas.

Arkitekturöversikt

Pipelinen har fem lager. Inget revolutionerande, men ordningen spelar roll.

1. Inläsning – läs in PDF:er, Markdown eller webbsidor.
2. Chunkning – dela upp i ~500–800 token långa bitar med overlap.
3. Embedding – generera 768- eller 1024-dimensionella vektorer.
4. Indexering – lagra i pgvector med HNSW-index.
5. Generering – hämta top-K, skicka till Claude med prompt caching.

Förutsättningar

• Python 3.11 eller senare
• PostgreSQL 16 eller senare
• En Anthropic API-nyckel (ANTHROPIC_API_KEY)
• Grundläggande SQL-kunskaper

Steg 1: Installera PostgreSQL och pgvector

På macOS via Homebrew:

brew install postgresql@16 pgvector
brew services start postgresql@16
createdb ragdemo

På Ubuntu 24.04:

sudo apt install postgresql-16 postgresql-16-pgvector
sudo -u postgres createdb ragdemo

Aktivera tillägget i databasen:

psql ragdemo -c "CREATE EXTENSION IF NOT EXISTS vector;"
psql ragdemo -c "CREATE EXTENSION IF NOT EXISTS pg_trgm;"

pg_trgm behöver vi senare för BM25-liknande textsökning i steg 6. Glöm det inte – det är lätt att missa och svårt att lägga till i efterhand utan att stöka till indexeringen.

Steg 2: Konfigurera Python-miljön

python -m venv .venv
source .venv/bin/activate

pip install \
    "anthropic>=0.40" \
    "psycopg[binary,pool]>=3.2" \
    "pgvector>=0.3" \
    "sentence-transformers>=3.0" \
    "tiktoken>=0.7" \
    "python-dotenv"

Skapa en .env med dina nycklar:

ANTHROPIC_API_KEY=sk-ant-...
DATABASE_URL=postgresql://localhost/ragdemo

Steg 3: Definiera schema och HNSW-index

Vi använder BAAI/bge-small-en-v1.5 som ger 384-dimensionella embeddings. För svenska texter byter du till intfloat/multilingual-e5-base (768 dimensioner) – byt då också vector(384) till vector(768) nedan. Skapa schemat:

CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    source      TEXT NOT NULL,
    chunk_index INT  NOT NULL,
    content     TEXT NOT NULL,
    tokens      INT  NOT NULL,
    metadata    JSONB DEFAULT '{}'::jsonb,
    embedding   vector(384) NOT NULL,
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX documents_hnsw_idx
    ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

CREATE INDEX documents_trgm_idx
    ON documents
    USING gin (content gin_trgm_ops);

CREATE INDEX documents_metadata_idx
    ON documents
    USING gin (metadata);

Vad gör HNSW-parametrarna?

• m – antal grannar per nod i grafen. Högre värde = bättre recall men större index. 16 är en bra startpunkt.
• ef_construction – ansträngning vid indexbygget. 64 räcker för <1 M rader; höj till 128–256 för större dataset.
• ef_search – sätts vid query-tid (SET hnsw.ef_search = 80;). Styr precision/latens-balansen runtime.

Steg 4: Chunkning och embeddings

Skapa ingest.py. Vi använder en token-medveten splitter med 600 tokens per chunk och 80 tokens overlap – tillräckligt för att bevara meningsgränser utan att skicka för stora bitar till Claude.

import os, json
from pathlib import Path

import psycopg
from psycopg.rows import dict_row
from pgvector.psycopg import register_vector
from sentence_transformers import SentenceTransformer
import tiktoken

EMBED_MODEL = "BAAI/bge-small-en-v1.5"
CHUNK_TOKENS = 600
OVERLAP = 80

enc = tiktoken.get_encoding("cl100k_base")
model = SentenceTransformer(EMBED_MODEL)

def chunk_text(text: str) -> list[str]:
    tokens = enc.encode(text)
    chunks = []
    step = CHUNK_TOKENS - OVERLAP
    for start in range(0, len(tokens), step):
        piece = enc.decode(tokens[start:start + CHUNK_TOKENS])
        chunks.append(piece)
        if start + CHUNK_TOKENS >= len(tokens):
            break
    return chunks

def ingest_file(conn, path: Path):
    text = path.read_text(encoding="utf-8")
    chunks = chunk_text(text)
    # bge-modeller fungerar bäst med passage-prefix
    payloads = [f"passage: {c}" for c in chunks]
    vectors = model.encode(payloads, normalize_embeddings=True)

    with conn.cursor() as cur:
        for i, (chunk, vec) in enumerate(zip(chunks, vectors)):
            cur.execute(
                "INSERT INTO documents (source, chunk_index, content, "
                "tokens, embedding, metadata) "
                "VALUES (%s, %s, %s, %s, %s, %s)",
                (str(path), i, chunk, len(enc.encode(chunk)),
                 vec, json.dumps({"filename": path.name})),
            )
    conn.commit()

if __name__ == "__main__":
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        register_vector(conn)
        for p in Path("docs").glob("**/*.md"):
            print(f"Indexerar {p}")
            ingest_file(conn, p)

Notera normalize_embeddings=True. Det är vad vector_cosine_ops förväntar sig och ger snabbare jämförelser i pgvector. Hoppar du över det får du fortfarande svar – men prestandan blir lidande, och det är en av de saker som är otroligt lätt att glömma vid en migrering.

Steg 5: Indexera dokument

Lägg dina källfiler i ./docs/ och kör:

python ingest.py

Du kan verifiera resultatet:

psql ragdemo -c "SELECT count(*), avg(tokens) FROM documents;"

Steg 6: Hybrid sökning – kombinera BM25 och vektorer

Så, nu till den intressanta delen. Ren vektorsökning missar ofta exakta termer som produktnamn, error-koder eller datum. I 2026 är hybrid sökning best practice för produktion – det är inte längre en "nice to have". Vi kombinerar pgvector-cosine med trigram-likhet (en BM25-approximation som följer med Postgres):

SQL = (
    "WITH vec AS ("
    "  SELECT id, content, source, "
    "         1 - (embedding <=> %s) AS vec_score "
    "  FROM documents "
    "  ORDER BY embedding <=> %s "
    "  LIMIT %s"
    "), kw AS ("
    "  SELECT id, content, source, "
    "         similarity(content, %s) AS kw_score "
    "  FROM documents "
    "  WHERE content %% %s "
    "  ORDER BY kw_score DESC "
    "  LIMIT %s"
    ") "
    "SELECT id, content, source, "
    "       COALESCE(v.vec_score, 0) * %s "
    "     + COALESCE(k.kw_score, 0) * (1 - %s) AS score "
    "FROM vec v FULL OUTER JOIN kw k USING (id, content, source) "
    "ORDER BY score DESC "
    "LIMIT %s"
)

def hybrid_search(conn, query: str, k: int = 8, alpha: float = 0.6):
    qvec = model.encode([f"query: {query}"], normalize_embeddings=True)[0]
    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute("SET LOCAL hnsw.ef_search = 80")
        cur.execute(
            SQL,
            (qvec, qvec, k * 2, query, query, k * 2, alpha, alpha, k),
        )
        return cur.fetchall()

Operatorn <=> är cosine-distans i pgvector. Vi konverterar till likhet med 1 - distance. alpha styr balansen: 0.6 ger något tyngre vikt åt semantisk likhet, vilket fungerar bra för de flesta Q&A-fall. Om du har mycket teknisk dokumentation med exakta termer kan du behöva sänka alpha till 0.4–0.5.

Steg 7: Anropa Claude med prompt caching

Här ligger den största kostnadsbesparingen. Sedan februari 2026 isoleras cachen per workspace, och standard-TTL är 5 minuter. Det betyder att du måste strukturera prompten så att statiska delar kommer först och cache_control sätts på sista statiska blocket. (Det här är inte uppenbart i dokumentationen – jag fick gräva i SDK-källkoden för att förstå exakt hur breakpoints räknas.)

import os
from anthropic import Anthropic

client = Anthropic()
MODEL = "claude-sonnet-4-6"

SYSTEM_PROMPT = (
    "Du är en teknisk assistent som svarar på frågor strikt baserat "
    "på dokumentationen som tillhandahålls. Citera alltid källan med "
    "[source]-syntax. Om svaret saknas, säg det rakt ut – gissa aldrig."
)

def build_context(chunks: list[dict]) -> str:
    parts = []
    for c in chunks:
        parts.append(f"[source: {c['source']}#{c['id']}]\n{c['content']}")
    return "\n\n---\n\n".join(parts)

def answer(question: str, chunks: list[dict]) -> str:
    context = build_context(chunks)
    response = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},
            },
            {
                "type": "text",
                "text": f"<documentation>\n{context}\n</documentation>",
                "cache_control": {"type": "ephemeral"},
            },
        ],
        messages=[{"role": "user", "content": question}],
    )
    return response.content[0].text

Varför två cache_control-block?

Du får använda upp till 4 cache_control-brytpunkter per request. Genom att sätta en på systemprompten och en på dokumentationen kan Claude återanvända systemprompten även när hämtade chunks ändras – vilket händer hela tiden i en RAG-applikation.

Minimitröskeln för Sonnet 4.6 är 2 048 tokens; Opus 4.7 kräver 4 096. Om dina retrieved chunks är mindre än så lönar sig caching av dokument-blocket inte. Cacha då bara systemprompten.

Steg 8: Sätt ihop hela pipen

import os
import psycopg
from pgvector.psycopg import register_vector

def rag(question: str, k: int = 6) -> str:
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        register_vector(conn)
        chunks = hybrid_search(conn, question, k=k)
    return answer(question, chunks)

if __name__ == "__main__":
    print(rag("Hur konfigurerar jag HNSW ef_search per query?"))

Optimera HNSW-parametrar för din arbetsbörda

Standardparametrarna är ofta inte optimala. Använd den här mätrutinen för att hitta rätt balans:

-- Kör för olika ef_search-värden och mät
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, content
FROM documents
ORDER BY embedding <=> '[...]'::vector
LIMIT 10;

Dataset	m	ef_construction	ef_search
< 100 K vektorer	16	64	40
100 K – 1 M	16	128	80
1 M – 10 M	32	200	120

Vanliga fallgropar

Cache invalideras av tidsstämplar

Sätt aldrig "Aktuell tid: 2026-04-30T14:32:15Z" i den cachade systemprompten. Lägg dynamiska värden i user-meddelandet, eller trunkera till dagsnivå om du behöver dem alls. Det här är förmodligen den vanligaste anledningen till att människor klagar på att deras cache "inte fungerar".

Ej normaliserade embeddings

Om du glömmer normalize_embeddings=True kommer vector_cosine_ops fortfarande fungera, men frågorna blir 2–3× långsammare. Verifiera med stickprov att vektornormen är ~1.

För stora chunks

Chunks över 1 200 tokens drunknar i Claudes uppmärksamhet och försämrar svarskvalitet. Håll dig till 400–800 tokens med 10–15 % overlap.

Indexbygget tar lång tid

HNSW-bygget är O(n × log n). På en 4-kärnig maskin tar 1 M vektorer cirka 6–10 minuter. Sätt maintenance_work_mem till minst 2GB innan du kör CREATE INDEX – annars sitter du och tittar på en framstegsstapel som inte rör sig.

Prestanda och kostnader (april 2026)

• Embedding (lokal bge-small): ~3 ms per chunk på CPU.
• Vektorsökning (1 M rader, HNSW): 4–9 ms per query.
• Claude Sonnet 4.6: $3 / 1 M input-tokens, $15 / 1 M output – cache-läs ~$0.30 / 1 M.
• Total RAG-latens: ofta 700–1 200 ms inklusive nätverk till Claude.

Med prompt caching aktiverad sjunker en typisk produktionsräkning från ~$50 till ~$15–20 per miljon förfrågningar, förutsatt att du har träff inom TTL. Det är en av de få optimeringar där du faktiskt kan se siffrorna förändras i fakturan nästa månad.

Nästa steg

• Lägg till en cross-encoder re-ranker (t.ex. BAAI/bge-reranker-v2-m3) ovanpå hybrid-resultaten för 5–15 % bättre precision.
• Partitionera documents per tenant_id i multi-tenant-applikationer.
• Kombinera detta med ett multi-agentsystem i LangGraph där en agent gör retrieval och en annan validerar svaret.

FAQ

Är pgvector tillräckligt snabbt för produktion?

Ja, för datasets upp till ~10 miljoner vektorer presterar HNSW i pgvector på enkelsiffriga millisekunder. Över det börjar specialiserade vektordatabaser som Qdrant eller Milvus visa fördelar i minneseffektivitet och horisontell skalning.

Vad är skillnaden mellan HNSW och IVFFlat?

HNSW är grafbaserat och ger låg latens med hög recall men kostar mer minne. IVFFlat partitionerar vektorrymden, bygger snabbare och använder mindre minne men har ofta lägre recall vid samma sökansträngning. För <100 K vektorer eller batch-uppdateringar är IVFFlat ofta tillräckligt; för latenskritisk produktion väljer du HNSW.

Hur många chunks ska jag skicka till Claude?

Vanligen 4–8 chunks à 500–800 tokens. Fler chunks ökar recall men späder ut uppmärksamheten och kostar mer. Mät svarskvalitet (t.ex. med en LLM-as-judge) på olika k-värden snarare än att gissa – min erfarenhet är att 6 är en överraskande bra default.

Kan jag använda pgvector tillsammans med min befintliga Postgres-data?

Absolut – det är en av huvudfördelarna. Du kan göra JOIN mellan vektor-tabeller och dina vanliga relations-tabeller och filtrera med WHERE innan ANN-sökningen körs. Det är svårt eller omöjligt i en separat vektordatabas.

Hur hanterar jag uppdateringar av dokument?

Lagra ett content_hash per chunk. När ett källdokument ändras, kör en diff: ta bort gamla rader vars hash försvunnit och infoga nya. HNSW stöder UPDATE/DELETE i pgvector ≥ 0.5, men prestanda försämras gradvis – planera in en REINDEX CONCURRENTLY kvartalsvis eller efter större batchimporter.

Avslutning

Du har nu en RAG-pipeline som tål produktion: pgvector med HNSW för snabb retrieval, hybrid sökning för precision på exakta termer, och Claude med prompt caching för låg kostnad.

Den största hävstången framåt är inte att byta vektordatabas. Det är att mäta retrieval-kvaliteten på riktiga frågor och iterera på chunkning, k-värden och re-ranking. Den som mäter, vinner.