pgvector + Crawl4AI: Building a Production Vector Search Pipeline for Web-Scraped Content
The Problem: Semantic Search Over Dynamic Web Content When building an AI agent that needs to query scraped web data semantically, you face a classic engineering tension: Crawl4AI excels at extraction throughput, but w
The Problem: Semantic Search Over Dynamic Web Content
When building an AI agent that needs to query scraped web data semantically, you face a classic engineering tension: Crawl4AI excels at extraction throughput, but without proper vector indexing, your retrieval latency collapses under load. I integrated Crawl4AI with pgvector for a compliance monitoring pipeline that needed to search 50k+ scraped regulatory documents in under 200ms. Here's the architecture that survived production.
Architecture: Crawl4AI → Embedding → pgvector HNSW
Crawl4AI handles extraction, we embed the cleaned Markdown chunks via a relay gateway (I benchmarked multi-provider fallback through B-Lost's relay at https://b-lost.com/v1 for token-cost arbitrage), then persist vectors in PostgreSQL with HNSW indexing.
PostgreSQL Schema with pgvector
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE scraped_documents (
id BIGSERIAL PRIMARY KEY,
url TEXT NOT NULL,
content_hash TEXT UNIQUE NOT NULL,
chunk_text TEXT NOT NULL,
embedding vector(1536), -- OpenAI ada-002 dimensions
metadata JSONB,
scraped_at TIMESTAMPTZ DEFAULT NOW()
);
-- HNSW index: m=16, ef_construction=64 for recall/speed balance
CREATE INDEX ON scraped_documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Filtered search support (critical for multi-tenant isolation)
CREATE INDEX idx_metadata_tenant ON scraped_documents
USING GIN (metadata jsonb_path_ops);
War Story 1: HNSW Index Lock Contention
During high-concurrency insertion (Crawl4AI batching 500 pages/min), HNSW index locks caused 3-second write stalls. Root cause: pgvector's HNSW implementation acquires exclusive locks during graph updates. Fix: batch inserts into a staging table without the HNSW index, then bulk-copy into the indexed table every 60 seconds.
Filtered Vector Search Query
-- Hybrid search: vector similarity + metadata filter
SELECT
url,
chunk_text,
1 - (embedding <=> $1::vector) AS cosine_similarity,
metadata->>'category' AS category
FROM scraped_documents
WHERE metadata @> '{"tenant_id": "acme_corp"}'::jsonb
ORDER BY embedding <=> $1::vector
LIMIT 20;
The <=> operator uses cosine distance. The metadata @> filter runs first (GIN index), then vector search operates on the reduced candidate set. Without the GIN index, query time jumped from 180ms to 4.2s on our 50k-doc corpus.
Python Hybrid Search with Reciprocal Rank Fusion
Pure vector search missed exact-match keywords. We implemented RRF to fuse vector results with PostgreSQL full-text search.
import asyncpg
import numpy as np
from typing import List, Tuple
async def hybrid_search_rrf(
pool: asyncpg.Pool,
query_embedding: List[float],
query_text: str,
tenant_id: str,
k: int = 20,
rrf_k: int = 60
) -> List[Tuple[str, float]]:
"""
Reciprocal Rank Fusion: combines vector + FTS results.
RRF score = sum(1 / (k + rank_i)) across all result lists.
"""
async with pool.acquire() as conn:
# Vector search results
vector_results = await conn.fetch(
"""
SELECT id, url, 1 - (embedding <=> $1::vector) AS score
FROM scraped_documents
WHERE metadata @> $2::jsonb
ORDER BY embedding <=> $1::vector
LIMIT $3
""",
query_embedding,
{"tenant_id": tenant_id},
k
)
# Full-text search results (tsvector on chunk_text)
fts_results = await conn.fetch(
"""
SELECT id, url, ts_rank(to_tsvector('english', chunk_text), query) AS score
FROM scraped_documents, plainto_tsquery('english', $1) query
WHERE to_tsvector('english', chunk_text) @@ query
AND metadata @> $2::jsonb
ORDER BY score DESC
LIMIT $3
""",
query_text,
{"tenant_id": tenant_id},
k
)
# RRF fusion
rrf_scores = {}
for rank, row in enumerate(vector_results, start=1):
rrf_scores[row['id']] = rrf_scores.get(row['id'], 0) + 1 / (rrf_k + rank)
for rank, row in enumerate(fts_results, start=1):
rrf_scores[row['id']] = rrf_scores.get(row['id'], 0) + 1 / (rrf_k + rank)
# Retrieve top-k by RRF score
top_ids = sorted(rrf_scores, key=rrf_scores.get, reverse=True)[:k]
final_results = await conn.fetch(
"SELECT id, url FROM scraped_documents WHERE id = ANY($1)",
top_ids
)
return [(r['url'], rrf_scores[r['id']]) for r in final_results]
War Story 2: Embedding Normalization Killed Recall
Initially, we stored raw embeddings from the relay gateway without L2 normalization. Cosine distance was mathematically correct, but recall@10 dropped to 62% because the HNSW graph construction is sensitive to vector magnitude distribution. After normalizing embeddings before insertion, recall recovered to 89%.
def normalize_embedding(embedding: List[float]) -> List[float]:
vec = np.array(embedding)
norm = np.linalg.norm(vec)
return (vec / norm).tolist() if norm > 0 else vec.tolist()
Crawl4AI Integration: Async Extraction Pipeline
from crawl4ai import AsyncWebCrawler, CacheMode
import asyncio
async def scrape_and_index(urls: List[str], pool: asyncpg.Pool, embed_fn):
async with AsyncWebCrawler(verbose=False) as crawler:
tasks = [crawler.arun(
url=url,
cache_mode=CacheMode.BYPASS,
word_count_threshold=50,
exclude_external_links=True
) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
if not result.success:
continue
chunks = chunk_markdown(result.markdown, max_tokens=512)
for chunk in chunks:
embedding = normalize_embedding(await embed_fn(chunk))
await pool.execute(
"""
INSERT INTO scraped_documents (url, content_hash, chunk_text, embedding, metadata)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (content_hash) DO NOTHING
""",
result.url,
hash_content(chunk),
chunk,
embedding,
{"tenant_id": "acme_corp", "scraped_at": result.success_at}
)
Crawl4AI's async crawler parallelizes requests, but beware: if your embedding service (or relay gateway) rate-limits at 500 req/min, you'll need a semaphore to throttle the embedding calls. We added asyncio.Semaphore(50) around embed_fn to prevent 429 errors.
Open Engineering Dilemma: Incremental Index Rebuilds
HNSW indices degrade under continuous small insertions. PostgreSQL's HNSW doesn't support incremental graph optimization. Do you batch-rebuild the index weekly and accept the 10-minute downtime window, or maintain dual indices (one for bulk inserts, one for realtime queries) and merge them daily? Our compliance SLA allows 15-minute Saturday maintenance windows, but multi-tenant isolation makes dual-index merging complex. What's your strategy for production vector index lifecycle management?
Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.