Using pgvector with LangChain for production RAG

Shipping a langchain pgvector production rag system means moving past the quickstart notebook. You need a Postgres schema that survives writes, an embedding pipeline that batches, and query paths that stay fast under concurrency. This guide lays out an ordered path we’ve used to put RAG on Postgres into production.

1. Provision the database and extension

The foundation of any langchain pgvector production rag deployment is the table and index. Enable the extension and define a narrow schema:

CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
  id uuid PRIMARY KEY,
  content text,
  embedding vector(1536),
  metadata jsonb
);

For production retrieval, build an HNSW index with cosine distance. pgvector 0.5+ supports HNSW, which avoids the training step IVFFlat requires and gives better recall:

CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

HNSW build still takes a brief exclusive lock on the table. Run it during a migration window or on a freshly loaded table before promoting to primary. Keep the embedding dimension locked to your model—vector(1536) matches OpenAI text-embedding-3-small; changing models later means a new column.

2. Embedding and ingestion pipeline

LangChain’s PGVector.from_documents is fine for prototypes, but it inserts row-by-row. For production, batch embeddings and use parameterized multi-value inserts.

Chunking before embed

Split source text with RecursiveCharacterTextSplitter:

from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
chunks = splitter.split_text(raw_doc)

Overlap recovers context at boundaries but increases row count linearly—budget storage accordingly.

Batch embed and insert

from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectors = embeddings.embed_documents(chunks)  # single batch API call

import psycopg2, uuid
conn = psycopg2.connect(dsn)
cur = conn.cursor()
for txt, vec in zip(chunks, vectors):
    cur.execute(
        "INSERT INTO documents (id, content, embedding, metadata) VALUES (%s,%s,%s,%s)",
        (str(uuid.uuid4()), txt, "[" + ",".join(map(str, vec)) + "]", {"src": "docs"})
    )
conn.commit()

Cast the vector list to the Postgres literal explicitly. Mismatched dimensions throw at insert, but a silent model swap that keeps the same dim will corrupt similarity.

3. Retrieval with LangChain

Point the LangChain vector store at the existing table:

from langchain.vectorstores import PGVector
store = PGVector(
    connection_string=dsn,
    embedding_function=embeddings,
    collection_name="docs",
    pre_delete_collection=False,
)

Use a score threshold to drop low-relevance hits instead of blindly returning top-k:

retriever = store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 8, "score_threshold": 0.78}
)

pgvector’s cosine operator <=> returns distance where 0 is identical. Threshold tuning is dataset-specific; log scores during eval.

Distance metric traps

If you indexed with vector_cosine_ops, query with <=>. Using L2 <-> on that column forces a sequential scan. Keep opclass and operator aligned, or you’ll watch latency climb on every request.

4. Tuning for concurrency

HNSW exposes m (graph connections per node) and ef_construction (search width at build). Default m=16 is okay; raise to 32 for higher recall at memory cost:

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

Per-query recall uses ef_search:

SET LOCAL hnsw.ef_search = 40;
SELECT id, content FROM documents ORDER BY embedding <=> '[...]' LIMIT 5;

LangChain’s retriever doesn’t expose session params. Wrap the connection in a helper that issues SET before the SELECT, or execute raw SQL through a pooled cursor.

Connection pooling is non-negotiable. Use PgBouncer in transaction mode. LangChain’s default PGVector opens a fresh connection per call if handed a DSN string; share a psycopg2.pool.SimpleConnectionPool instead. HNSW index memory is roughly m * dim * 4 bytes * rows plus graph overhead—for 1M rows at 1536-dim expect several GB RAM. Verify the instance class before scaling.

5. Updates, deletes, and bloat

RAG corpora change. Soft-delete by tagging metadata, then filter in the query:

SELECT id, content FROM documents
WHERE metadata->>'deleted' IS NULL
ORDER BY embedding <=> '[...]' LIMIT 5;

Periodically purge and vacuum:

DELETE FROM documents WHERE metadata->>'expired' = 'true';
VACUUM (VERBOSE, ANALYZE) documents;

HNSW indexes don’t shrink automatically. After large deletes, plan a REINDEX during off-peak or accept gradual overhead until the next full rebuild.

6. Common pitfalls in langchain pgvector production rag

• Storing full documents in the embedding table causes TOAST bloat and slows index scans. Keep content in a separate table or object storage, join by id.
• Sync drivers in async services: LangChain’s PGVector uses blocking psycopg2. In FastAPI or async handlers, run retrieval in a threadpool or use asyncpg with custom SQL.
• Embedding model drift: Changing dimensions requires a new column or table. Backfill with a job, not a live ALTER.
• No query logging: You can’t tune what you can’t see. Log vector query latency and score distributions from day one.
• IVFFlat in production: Teams pick it for “maturity” and then miss the CREATE INDEX training step, getting terrible recall. Prefer HNSW on pgvector 0.5+.

7. Serving the chain

Wire the retriever into an LCEL pipeline and stream the LLM response:

from langchain.chat_models import ChatOpenAI
from langchain.schema.runnable import RunnablePassthrough
from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer from context: {context}"),
    ("human", "{question}")
])
model = ChatOpenAI(streaming=True, temperature=0)
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
)

For resilient model access, an OpenRouter-class gateway like n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models, applies automatic fallback when a provider is rate-limited, and forwards cache-control hints—swap ChatOpenAI’s openai_api_base and keep the chain unchanged.

Monitoring

Track token usage per request and Postgres index size with \di+. Alert on p99 retrieval above 50ms and on embedding insert failures.

8. Deployment checklist

1. pgvector extension installed; HNSW index built off-peak.
2. Batch ingestion with explicit dimension checks and chunk overlap tuned.
3. Retriever uses score threshold, not raw top-k.
4. PgBouncer in front of Postgres; shared pool in the app.
5. Soft-delete plus scheduled vacuum and reindex plan.
6. LLM call behind a gateway with fallback and metering.
7. Logs for scores, latency, and token counts from day one.

That’s the core of a langchain pgvector production rag deployment that holds up under real load.