Build a Vector Search API with FastAPI and pgvector: Production Guide

Q: How do you build a vector search API with FastAPI and pgvector?

Build a FastAPI service with asyncpg connection pooling, embedding endpoints, and cosine similarity search using pgvector's operator. [Percona](https://percona.community/blog/2026/05/31/semantic-search-on-postgresql-part-2/): 'FastAPI, a background indexer, and PostgreSQL with pgvector are all in a single Docker Compose file. A visitor enters a query, the widget sends a POST /search request to FastAPI. The service computes the query embedding and searches for nearest vectors in Postgres.' [7Tech](https://www.7tech.co.in/ai-rag-in-2026-build-a-production-ready-fastapi-pgvector-service-with-hybrid-search-and-reranking/): 'FastAPI gives a simple high-performance API layer. PostgreSQL + pgvector keeps vectors and metadata in one trusted database. Hybrid search improves recall. Reranking improves precision.' [KowashLab](https://kowashlab.com/blog/building-production-rag-pipeline-fastapi-postgresql): 'PostgreSQL with pgvector handles the storage layer without adding another database. FastAPI handles the API layer with async support. The combination is simple enough to deploy and powerful enough for production.' Steps: (1) Create FastAPI app with asyncpg pool. (2) Add /upsert endpoint for inserting embeddings. (3) Add /search endpoint with cosine distance. (4) Add /hybrid-search with vector + keyword. (5) Add /ask endpoint with RAG generation.

By Fadi Updated Jul 6, 2025 14 min read Vector Databases

TL;DR — Build a production vector search API with FastAPI and pgvector. Percona: "FastAPI, background indexer, and PostgreSQL with pgvector in a single Docker Compose. POST /search computes query embedding and searches nearest vectors." 7Tech: "FastAPI for API layer, pgvector for storage, hybrid search for recall, reranking for precision. Ingest → embed → store → search → rerank → generate." KowashLab: "pgvector handles storage without another database. FastAPI handles API with async. Simple to deploy, powerful enough for production." Markaicode: "asyncpg pool 200 connections, PgBouncer max_client_conn=400. 8 months production, 50K sessions/day, 2ms p50 search." samshad: "asyncpg raw SQL over SQLAlchemy — fastest driver, direct pgvector control. Docker multi-stage, GitHub Actions CI." Learn more with pgvector tutorial, semantic search, generate embeddings, and vector databases.

7Tech frames the architecture: "FastAPI gives a simple high-performance API layer. PostgreSQL + pgvector keeps vectors and metadata in one trusted database. Hybrid search improves recall by combining semantic and lexical matching. Reranking improves precision before generation."

KowashLab adds: "PostgreSQL with pgvector handles the storage layer without adding another database to your stack. FastAPI handles the API layer with async support for embedding and LLM calls. The combination is simple enough to deploy and powerful enough for production traffic."

Vector Search API Architecture

flowchart TD subgraph Client["Client"] Widget["Search Widget
POST /search"] Agent["AI Agent
POST /ask"] Admin["Admin Dashboard
POST /index/start"] end subgraph API["FastAPI Service"] Search["POST /search
embed query → cosine search"] Hybrid["POST /hybrid-search
vector + keyword → RRF fusion"] Ask["POST /ask
search → rerank → LLM generate"] Upsert["POST /upsert
embed → insert with ON CONFLICT"] Health["GET /health
DB connectivity check"] end subgraph Pool["Connection Layer"] AsyncPG["asyncpg Pool
min=50, max=200
statement_cache=500"] PgBouncer["PgBouncer
max_client_conn=400
pool_mode=transaction"] end subgraph DB["PostgreSQL + pgvector"] Primary["Primary
writes + index maintenance"] Replica["Read Replica
query serving
streaming replication"] HNSW["HNSW Index
vector_cosine_ops
m=16, ef_construction=200"] TSV["tsvector Index
keyword search
to_tsvector('english', ...)"] end subgraph Indexer["Background Indexer"] Crawl["Crawl/Fetch
RSS, HTML, documents"] Chunk["Chunk Text
sliding window"] Embed["Embed Chunks
nomic / OpenAI / BGE"] Write["Write to DB
pages + embeddings"] end Widget --> Search Agent --> Ask Admin --> Upsert Search --> AsyncPG Hybrid --> AsyncPG Ask --> AsyncPG Upsert --> AsyncPG Health --> AsyncPG AsyncPG --> PgBouncer PgBouncer --> Replica Replica --> HNSW Replica --> TSV Primary --> Replica Crawl --> Chunk --> Embed --> Write Write --> Primary

API Endpoint Comparison

Endpoint	Method	Purpose	Query Pattern	Response
`/search`	POST	Semantic vector search	`ORDER BY embedding <=> query LIMIT k`	Ranked results with similarity
`/hybrid-search`	POST	Vector + keyword hybrid	RRF fusion of vector + tsvector	Fused ranked results
`/ask`	POST	Full RAG with generation	Search → rerank → LLM	Answer with citations
`/upsert`	POST	Insert/update embeddings	`INSERT ... ON CONFLICT DO UPDATE`	Insert count + status
`/index/start`	POST	Trigger background indexing	Enqueue job to indexer	Job ID + status
`/health`	GET	Health check	`SELECT 1`	DB connectivity status

Production Configuration

Component	Configuration	Purpose
asyncpg pool	min_size=50, max_size=200	Connection pooling for concurrent requests
PgBouncer	max_client_conn=400, pool_mode=transaction	Connection multiplexing
statement_timeout	5000ms	Fail fast rather than hang under load
HNSW index	m=16, ef_construction=200, vector_cosine_ops	Fast ANN search with high recall
Read replica	streaming replication, sync_commit=remote_write	Scale reads, offload from primary
pg_repack	cron at 3 AM	Index maintenance without locks
Docker	multi-stage, non-root	Production deployment
CI/CD	GitHub Actions with ephemeral pgvector	Test before deploy

Implementation

import os
import json
import asyncio
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Optional
from enum import Enum

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

@dataclass
class VectorSearchAPI:
    """Production vector search API with FastAPI and pgvector."""

    def get_app_setup(self) -> str:
        """FastAPI app setup with asyncpg pool."""
        return (
            "from fastapi import FastAPI\n"
            "from contextlib import asynccontextmanager\n"
            "import asyncpg\n"
            "from pgvector.asyncpg import register_vector\n"
            "\n"
            "POOL = None\n"
            "\n"
            "@asynccontextmanager\n"
            "async def lifespan(app):\n"
            "    global POOL\n"
            "    POOL = await asyncpg.create_pool(\n"
            "        dsn='postgres://user:pass@pgbouncer:6432/db',\n"
            "        min_size=50,\n"
            "        max_size=200,\n"
            "        statement_cache_size=500,\n"
            "        command_timeout=5.0,\n"
            "    )\n"
            "    async with POOL.acquire() as conn:\n"
            "        await register_vector(conn)\n"
            "    yield\n"
            "    await POOL.close()\n"
            "\n"
            "app = FastAPI(lifespan=lifespan)"
        )

    def get_schema_sql(self) -> str:
        """Database schema with pgvector and tsvector."""
        return (
            "CREATE EXTENSION IF NOT EXISTS vector;\n"
            "\n"
            "CREATE TABLE IF NOT EXISTS documents (\n"
            "    id BIGSERIAL PRIMARY KEY,\n"
            "    source TEXT NOT NULL,\n"
            "    title TEXT,\n"
            "    chunk_text TEXT NOT NULL,\n"
            "    chunk_index INT NOT NULL,\n"
            "    metadata JSONB DEFAULT '{}'::jsonb,\n"
            "    embedding vector(1536),\n"
            "    tsv tsvector GENERATED ALWAYS AS (\n"
            "        to_tsvector('english',\n"
            "            coalesce(title,'') || ' ' || chunk_text)\n"
            "    ) STORED,\n"
            "    created_at TIMESTAMPTZ DEFAULT now()\n"
            ");\n"
            "\n"
            "CREATE INDEX IF NOT EXISTS documents_embedding_hnsw\n"
            "ON documents USING hnsw (embedding vector_cosine_ops)\n"
            "WITH (m = 16, ef_construction = 200);\n"
            "\n"
            "CREATE INDEX IF NOT EXISTS documents_tsv_idx\n"
            "ON documents USING gin(tsv);"
        )

    def get_search_endpoint(self) -> str:
        """POST /search — semantic vector search."""
        return (
            "class SearchRequest(BaseModel):\n"
            "    query: str\n"
            "    top_k: int = 5\n"
            "    min_similarity: float = 0.0\n"
            "    filter_metadata: dict = None\n"
            "\n"
            "class SearchResult(BaseModel):\n"
            "    id: int\n"
            "    source: str\n"
            "    title: str\n"
            "    chunk_text: str\n"
            "    similarity: float\n"
            "    metadata: dict\n"
            "\n"
            "@app.post('/search', response_model=list[SearchResult])\n"
            "async def search(req: SearchRequest):\n"
            "    query_embedding = await embed_text(req.query)\n"
            "    async with POOL.acquire() as conn:\n"
            "        await conn.execute(\n"
            "            'SET LOCAL hnsw.ef_search = 100')\n"
            "        rows = await conn.fetch('''\n"
            "            SELECT id, source, title, chunk_text,\n"
            "                   metadata,\n"
            "                   1 - (embedding <=> $1::vector)\n"
            "                       AS similarity\n"
            "            FROM documents\n"
            "            ORDER BY embedding <=> $1::vector\n"
            "            LIMIT $2\n"
            "        ''', str(query_embedding), req.top_k)\n"
            "    return [SearchResult(\n"
            "        id=r['id'], source=r['source'],\n"
            "        title=r['title'], chunk_text=r['chunk_text'],\n"
            "        similarity=float(r['similarity']),\n"
            "        metadata=json.loads(r['metadata'])\n"
            "    ) for r in rows\n"
            "    if float(r['similarity']) >= req.min_similarity]"
        )

    def get_hybrid_search_endpoint(self) -> str:
        """POST /hybrid-search — vector + keyword with RRF."""
        return (
            "@app.post('/hybrid-search')\n"
            "async def hybrid_search(req: SearchRequest):\n"
            "    query_embedding = await embed_text(req.query)\n"
            "    async with POOL.acquire() as conn:\n"
            "        rows = await conn.fetch('''\n"
            "            WITH vec AS (\n"
            "                SELECT id, source, title, chunk_text,\n"
            "                       metadata,\n"
            "                       1 - (embedding <=> $1::vector)\n"
            "                           AS score\n"
            "                FROM documents\n"
            "                ORDER BY embedding <=> $1::vector\n"
            "                LIMIT 40\n"
            "            ),\n"
            "            lex AS (\n"
            "                SELECT id, source, title, chunk_text,\n"
            "                       metadata,\n"
            "                       ts_rank_cd(tsv,\n"
            "                           plainto_tsquery('english',\n"
            "                               $2)) AS score\n"
            "                FROM documents\n"
            "                WHERE tsv @@ plainto_tsquery(\n"
            "                    'english', $2)\n"
            "                ORDER BY score DESC\n"
            "                LIMIT 40\n"
            "            )\n"
            "            SELECT COALESCE(v.id, l.id) AS id,\n"
            "                   COALESCE(v.source, l.source)\n"
            "                       AS source,\n"
            "                   COALESCE(v.title, l.title)\n"
            "                       AS title,\n"
            "                   COALESCE(v.chunk_text,\n"
            "                       l.chunk_text) AS chunk_text,\n"
            "                   COALESCE(v.metadata,\n"
            "                       l.metadata) AS metadata,\n"
            "                   COALESCE(v.score, 0)\n"
            "                       AS vector_score,\n"
            "                   COALESCE(l.score, 0)\n"
            "                       AS keyword_score,\n"
            "                   (COALESCE(v.score, 0)\n"
            "                    + COALESCE(l.score, 0))\n"
            "                       AS combined_score\n"
            "            FROM vec v\n"
            "            FULL OUTER JOIN lex l ON v.id = l.id\n"
            "            ORDER BY combined_score DESC\n"
            "            LIMIT $3\n"
            "        ''', str(query_embedding),\n"
            "              req.query, req.top_k)\n"
            "    return [dict(r) for r in rows]"
        )

    def get_upsert_endpoint(self) -> str:
        """POST /upsert — insert embeddings with ON CONFLICT."""
        return (
            "class UpsertRequest(BaseModel):\n"
            "    documents: list[dict]\n"
            "    # each: {source, title, chunk_text,\n"
            "    #        chunk_index, metadata}\n"
            "\n"
            "@app.post('/upsert')\n"
            "async def upsert(req: UpsertRequest):\n"
            "    inserted = 0\n"
            "    async with POOL.acquire() as conn:\n"
            "        for doc in req.documents:\n"
            "            embedding = await embed_text(\n"
            "                doc['chunk_text'])\n"
            "            await conn.execute('''\n"
            "                INSERT INTO documents\n"
            "                    (source, title, chunk_text,\n"
            "                     chunk_index, metadata,\n"
            "                     embedding)\n"
            "                VALUES ($1, $2, $3, $4, $5::jsonb,\n"
            "                        $6::vector)\n"
            "                ON CONFLICT (id) DO UPDATE SET\n"
            "                    chunk_text = EXCLUDED.chunk_text,\n"
            "                    embedding = EXCLUDED.embedding,\n"
            "                    metadata = EXCLUDED.metadata\n"
            "            ''', doc['source'], doc['title'],\n"
            "                 doc['chunk_text'],\n"
            "                 doc['chunk_index'],\n"
            "                 json.dumps(doc.get('metadata', {})),\n"
            "                 str(embedding))\n"
            "            inserted += 1\n"
            "    return {'inserted': inserted,\n"
            "            'status': 'ok'}"
        )

    def get_health_endpoint(self) -> str:
        """GET /health — database connectivity check."""
        return (
            "@app.get('/health')\n"
            "async def health():\n"
            "    try:\n"
            "        async with POOL.acquire() as conn:\n"
            "            await conn.fetchval('SELECT 1')\n"
            "        return {'status': 'healthy',\n"
            "                'database': 'connected'}\n"
            "    except Exception as e:\n"
            "        raise HTTPException(\n"
            "            status_code=503,\n"
            "            detail=f'Database error: {e}')"
        )

    def get_docker_compose(self) -> str:
        """Docker Compose for production deployment."""
        return (
            "version: '3.8'\n"
            "services:\n"
            "  api:\n"
            "    build: .\n"
            "    ports: ['8000:8000']\n"
            "    depends_on: [postgres, pgbouncer]\n"
            "    environment:\n"
            "      - DATABASE_URL=postgres://user:pass@\n"
            "        pgbouncer:6432/db\n"
            "\n"
            "  postgres:\n"
            "    image: pgvector/pgvector:0.8.0-pg16\n"
            "    volumes:\n"
            "      - pgdata:/var/lib/postgresql/data\n"
            "    environment:\n"
            "      - POSTGRES_PASSWORD=pass\n"
            "\n"
            "  pgbouncer:\n"
            "    image: edoburu/pgbouncer\n"
            "    environment:\n"
            "      - DB_USER=user\n"
            "      - DB_PASSWORD=pass\n"
            "      - DB_HOST=postgres\n"
            "      - MAX_CLIENT_CONN=400\n"
            "      - POOL_MODE=transaction\n"
            "\n"
            "volumes:\n"
            "  pgdata:"
        )

    def get_failure_modes(self) -> dict:
        """Production failure modes and mitigations."""
        return {
            "connection_pool_exhaustion": {
                "cause": "Agents opening multiple connections per request",
                "detection": "PgBouncer SHOW POOLS active > 80%",
                "mitigation": "asyncpg min_size=50, context manager "
                              "that releases on exit",
            },
            "embedding_api_failure": {
                "cause": "OpenAI or embedding API downtime",
                "detection": "HTTP errors, timeout on embed calls",
                "mitigation": "Retry with exponential backoff, "
                              "queue ingestion jobs",
            },
            "index_bloat": {
                "cause": "Frequent updates and deletes",
                "detection": "pg_stat_user_tables dead tuples",
                "mitigation": "pg_repack off-peak, "
                              "cron at 3 AM",
            },
            "replica_lag": {
                "cause": "High write volume on primary",
                "detection": "pg_stat_replication flush_lag > 2s",
                "mitigation": "synchronous_commit=remote_write, "
                              "alert if lag > 2s",
            },
            "query_timeout": {
                "cause": "Complex queries under load",
                "detection": "statement_timeout errors",
                "mitigation": "command_timeout=5.0, "
                              "fail fast rather than hang",
            },
        }

Vector Search API Checklist

[ ] Build FastAPI service with asyncpg connection pooling
[ ] Use asyncpg raw SQL over SQLAlchemy — fastest Python Postgres driver, direct pgvector control
[ ] asyncpg pool: min_size=50, max_size=200, statement_cache_size=500, command_timeout=5.0
[ ] Use PgBouncer with max_client_conn=400, pool_mode=transaction
[ ] Enforce connection reuse via context manager that always releases on exit
[ ] PostgreSQL with pgvector keeps vectors and metadata in one trusted database
[ ] FastAPI handles API layer with async support for embedding and LLM calls
[ ] Schema: documents table with vector(1536), tsvector GENERATED ALWAYS, JSONB metadata
[ ] HNSW index: m=16, ef_construction=200, vector_cosine_ops
[ ] GIN index on tsvector for keyword search
[ ] POST /search: embed query, cosine distance search, return ranked results
[ ] Use <=> cosine distance operator, 1 - distance = similarity
[ ] SET LOCAL hnsw.ef_search = 100 in transaction for tuning
[ ] POST /hybrid-search: vector + keyword with RRF fusion
[ ] Vector search catches semantic matches, keyword search catches exact terms
[ ] WITH vec AS (...), lex AS (...) FULL OUTER JOIN, combined score
[ ] POST /ask: search → rerank → LLM generate with citations
[ ] POST /upsert: embed and insert with ON CONFLICT DO UPDATE
[ ] GET /health: SELECT 1 for database connectivity check
[ ] POST /index/start: trigger background indexing job
[ ] Background indexer: crawl → chunk → embed → write to DB
[ ] Indexer runs in separate container, does not block HTTP
[ ] Docker Compose: FastAPI + PostgreSQL + PgBouncer all in containers
[ ] Use pgvector/pgvector:0.8.0-pg16 Docker image
[ ] Docker multi-stage build, non-root user
[ ] CI/CD: GitHub Actions with lint → test with ephemeral pgvector DB
[ ] Observability: structlog for structured logging, Better Stack for monitoring
[ ] Read replica for query serving, primary for writes and index maintenance
[ ] Streaming replication with synchronous_commit=remote_write
[ ] Monitor pg_stat_replication flush_lag < 2s
[ ] pg_repack for index maintenance at off-peak (3 AM cron)
[ ] statement_timeout=5000 on PgBouncer — fail fast rather than hang
[ ] Handle embedding API failures with retry and exponential backoff
[ ] Queue ingestion jobs rather than dropping documents on API failure
[ ] Connection pool exhaustion: first failure in production — fix with pool sizing
[ ] 8 months production: 50K sessions/day, 2ms p50 pgvector search
[ ] pgvector queries are CPU-intensive — too many concurrent workers saturate CPU
[ ] HNSW offers faster queries but 2-3x slower inserts that block readers
[ ] Defer index maintenance to off-peak to avoid blocking reader connections
[ ] asyncpg create_pool with min_size=50 prevents connection storms
[ ] Real DB integration tests over mocks — mocking raw SQL hides syntax errors
[ ] Ephemeral pgvector containers in CI catch real failures
[ ] Custom JWT auth for multi-tenant — full ownership, no vendor lock-in
[ ] Streaming responses with StreamingResponse for token-by-token LLM output
[ ] Grounded generation: retrieved chunks constrain LLM to provided context only
[ ] Same setup locally and in production — Docker Compose consistency
[ ] Read pgvector tutorial for setup
[ ] Read semantic search for search
[ ] Read generate embeddings for embedding
[ ] Read vector databases for fundamentals
[ ] Read hybrid search for BM25 + vector
[ ] Read reranking for post-retrieval precision
[ ] Test: /search returns ranked results with similarity scores
[ ] Test: /hybrid-search combines vector and keyword results
[ ] Test: /upsert inserts embeddings with ON CONFLICT
[ ] Test: /health detects database connectivity issues
[ ] Test: connection pool handles concurrent requests without exhaustion
[ ] Test: embedding API failure triggers retry with backoff
[ ] Document API endpoints, pool config, Docker setup, and failure modes

FAQ

How do you build a vector search API with FastAPI and pgvector?

Build a FastAPI service with asyncpg connection pooling, embedding endpoints, and cosine similarity search using pgvector's <=> operator. Percona: "FastAPI, a background indexer, and PostgreSQL with pgvector are all in a single Docker Compose file. A visitor enters a query, the widget sends a POST /search request to FastAPI. The service computes the query embedding and searches for nearest vectors in Postgres." 7Tech: "FastAPI gives a simple high-performance API layer. PostgreSQL + pgvector keeps vectors and metadata in one trusted database. Hybrid search improves recall. Reranking improves precision." KowashLab: "PostgreSQL with pgvector handles the storage layer without adding another database. FastAPI handles the API layer with async support. The combination is simple enough to deploy and powerful enough for production." Steps: (1) Create FastAPI app with asyncpg pool. (2) Add /upsert endpoint for inserting embeddings. (3) Add /search endpoint with cosine distance. (4) Add /hybrid-search with vector + keyword. (5) Add /ask endpoint with RAG generation.

How do you set up connection pooling for pgvector with FastAPI?

Use asyncpg with create_pool for connection pooling, sized 50-200 connections, with PgBouncer in front for production. Markaicode: "The Vector API is a FastAPI service that wraps pgvector queries. It uses asyncpg with a connection pool sized at 200, matching PgBouncer config. asyncpg.create_pool with min_size=50, max_size=200, statement_cache_size=500, command_timeout=5.0. pgvector queries are CPU-intensive — too many concurrent workers saturate PostgreSQL CPU." samshad: "asyncpg raw SQL over SQLAlchemy — fastest Python Postgres driver. Direct control over pgvector operators and bulk binary insertions. No ORM overhead." Setup: POOL = await asyncpg.create_pool(dsn=..., min_size=50, max_size=200, statement_cache_size=500, command_timeout=5.0). Use PgBouncer with max_client_conn=400, pool_mode=transaction. Enforce connection reuse via context manager that always releases on exit.

How do you implement hybrid search with reranking in FastAPI and pgvector?

Run vector search and keyword search separately, combine scores with RRF fusion, then rerank top candidates with a cross-encoder. 7Tech: "Run both searches and combine scores. Vector search catches semantic matches, keyword search catches exact terms, version numbers, API names, and error codes. Rerank top candidates with cross-encoder or hosted reranker." SQL: WITH vec AS (SELECT id, 1-(embedding <=> $1::vector) AS score FROM documents ORDER BY embedding <=> $1::vector LIMIT 40), lex AS (SELECT id, ts_rank_cd(tsv, plainto_tsquery('english', $2)) AS score FROM documents WHERE tsv @@ plainto_tsquery('english', $2) ORDER BY score DESC LIMIT 40). FastAPI: candidates = hybrid_search(q_emb, req.question, limit=20); reranked = rerank(req.question, candidates)[:6]; context = join chunks; prompt LLM with context.

How do you deploy a FastAPI pgvector API in production?

Deploy with Docker Compose containing FastAPI, PostgreSQL with pgvector, and PgBouncer, using read replicas for query scaling. Percona: "Docker Compose with Postgres, the API, and the indexer all in containers, with the same setup locally and in production. Production on EC2 on AWS, an ARM instance, using the same docker-compose." Markaicode: "Use a primary-replica setup with streaming replication. Primary handles writes and index maintenance; replica serves agent queries. Index rebuilds deferred to off-peak hours using pg_repack. Ran in production on AWS EC2 G4dn.xlarge for 8 months, serving 150 agents at 50K sessions/day with 2ms p50 pgvector search." samshad: "Docker multi-stage, non-root. CI/CD with GitHub Actions — lint, test with ephemeral pgvector DB. Observability with structlog, Better Stack." Production: (1) Docker Compose with FastAPI + Postgres + PgBouncer. (2) Read replica for queries. (3) pg_repack for index maintenance. (4) statement_timeout=5000. (5) Health check endpoint.

What are the production failure modes for a FastAPI pgvector API?

Common failures include connection pool exhaustion, embedding API failures, index bloat, and replica lag. Markaicode: "First thing that broke was connection pool — agents opened two connections per request, hit PgBouncer 400-connection limit within 20 minutes. Fix: enforce connection reuse via asyncpg pool min_size=50 and context manager that always releases on exit. Architecture held steady at 50K sessions/day with 2ms p50." KowashLab: "Handle embedding API failures with retry logic and exponential backoff. If API is down, queue ingestion jobs rather than dropping documents." Failure modes: (1) Connection pool exhaustion — fix with PgBouncer + pool sizing. (2) Embedding API failures — retry with backoff, queue jobs. (3) Index bloat — pg_repack off-peak. (4) Replica lag — monitor pg_stat_replication, alert if >2s. (5) Query timeout — statement_timeout=5000, fail fast. (6) HNSW insert blocking — defer index maintenance to off-peak.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →