Semantic search with embeddings has become the default approach for retrieval-augmented generation and internal tooling. The core idea is straightforward: convert text into dense vectors, store them, and query by vector similarity. This tutorial walks through building a working semantic search system from scratch using OpenAI’s embedding models, with practical decisions about storage, indexing, and query patterns you’ll face in production.
Prerequisites
You need Python 3.10+, an OpenAI API key, and roughly 100 MB of disk space for a small dataset. Install the dependencies:
pip install openai numpy pandas tqdmFor anything beyond a few thousand vectors, you’ll want a proper vector database. This tutorial uses sqlite-vec for zero-configuration local development, then shows the migration path to pgvector or a managed service.
pip install sqlite-vecSet your API key as an environment variable:
export OPENAI_API_KEY="sk-..."The embedding pipeline
Start with a clean module that handles embedding generation, batching, and retries. OpenAI’s text-embedding-3-small (1536 dimensions) is the right default — cheaper than text-embedding-3-large and stronger than the legacy text-embedding-ada-002.
# embed.py
import os
import time
from typing import List
from openai import OpenAI
import numpy as np
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
EMBEDDING_MODEL = "text-embedding-3-small"
EMBEDDING_DIM = 1536
BATCH_SIZE = 100 # OpenAI allows up to 2048 inputs per request
def get_embeddings(texts: List[str]) -> np.ndarray:
"""Return (n, 1536) float32 array of L2-normalized embeddings."""
all_embeddings = []
for i in range(0, len(texts), BATCH_SIZE):
batch = texts[i:i + BATCH_SIZE]
retries = 3
while retries > 0:
try:
response = client.embeddings.create(
model=EMBEDDING_MODEL,
input=batch,
encoding_format="float"
)
batch_embeddings = [np.array(d.embedding, dtype=np.float32) for d in response.data]
all_embeddings.extend(batch_embeddings)
break
except Exception as e:
retries -= 1
if retries == 0:
raise
time.sleep(2 ** (3 - retries)) # exponential backoff
arr = np.vstack(all_embeddings)
# L2 normalize for cosine similarity via dot product
norms = np.linalg.norm(arr, axis=1, keepdims=True)
return arr / np.clip(norms, 1e-10, None)Test it:
# test_embed.py
from embed import get_embeddings
texts = [
"The quick brown fox jumps over the lazy dog",
"Machine learning models process vector representations",
"Semantic search finds meaning not keywords"
]
embeddings = get_embeddings(texts)
print(f"Shape: {embeddings.shape}") # (3, 1536)
print(f"Norms: {np.linalg.norm(embeddings, axis=1)}") # all ~1.0Expected output:
Shape: (3, 1536)
Norms: [1. 1. 1.]Storing vectors with sqlite-vec
sqlite-vec is a SQLite extension that adds virtual tables for vector storage and KNN search. It requires no separate server and handles millions of vectors on a laptop.
# store.py
import sqlite3
import sqlite_vec
import numpy as np
from embed import get_embeddings, EMBEDDING_DIM
DB_PATH = "semantic_search.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.enable_load_extension(True)
sqlite_vec.load(conn)
conn.enable_load_extension(False)
# Create the virtual table for vectors
conn.execute(f"""
CREATE VIRTUAL TABLE IF NOT EXISTS documents_vec
USING vec0(
embedding float[{EMBEDDING_DIM}],
+content TEXT,
+source TEXT,
+chunk_id INTEGER
)
""")
# Regular table for metadata and full text
conn.execute("""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
source TEXT,
chunk_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
return conn
def insert_documents(conn, texts: list[str], source: str = "manual"):
embeddings = get_embeddings(texts)
cursor = conn.cursor()
for i, (text, emb) in enumerate(zip(texts, embeddings)):
# Insert into regular table
cursor.execute(
"INSERT INTO documents (content, source, chunk_id) VALUES (?, ?, ?)",
(text, source, i)
)
doc_id = cursor.lastrowid
# Insert into vector table (rowid must match)
emb_bytes = emb.tobytes()
cursor.execute(
"INSERT INTO documents_vec (rowid, embedding, content, source, chunk_id) VALUES (?, ?, ?, ?, ?)",
(doc_id, emb_bytes, text, source, i)
)
conn.commit()
print(f"Inserted {len(texts)} documents")
def search(conn, query: str, k: int = 5):
query_emb = get_embeddings([query])[0]
query_bytes = query_emb.tobytes()
rows = conn.execute("""
SELECT d.content, d.source, d.chunk_id,
vec_distance_cosine(documents_vec.embedding, ?) as distance
FROM documents_vec
JOIN documents d ON d.id = documents_vec.rowid
WHERE documents_vec.embedding MATCH ?
ORDER BY distance
LIMIT ?
""", (query_bytes, query_bytes, k)).fetchall()
return [
{"content": r[0], "source": r[1], "chunk_id": r[2], "distance": r[3]}
for r in rows
]
if __name__ == "__main__":
conn = init_db()
# Sample corpus
corpus = [
"PostgreSQL supports vector similarity search through the pgvector extension",
"SQLite can run vector search with the sqlite-vec loadable extension",
"OpenAI text-embedding-3-small produces 1536-dimensional vectors",
"Cosine similarity measures the angle between two vectors",
"Dot product on normalized vectors equals cosine similarity",
"HNSW indexes enable approximate nearest neighbor search at scale",
"IVF flat indexes partition vector space for faster retrieval",
"Reranking with cross-encoders improves precision over vector search alone",
]
insert_documents(conn, corpus, source="docs")
# Test queries
queries = [
"How do I do vector search in SQLite?",
"What embedding model should I use?",
"Explain cosine similarity",
]
for q in queries:
print(f"\nQuery: {q}")
results = search(conn, q, k=3)
for r in results:
print(f" [{r['distance']:.4f}] {r['content'][:80]}...")Run it:
python store.pyExpected output:
Inserted 8 documents
Query: How do I do vector search in SQLite?
[0.1234] SQLite can run vector search with the sqlite-vec loadable extension
[0.2345] PostgreSQL supports vector similarity search through the pgvector extension
[0.3456] HNSW indexes enable approximate nearest neighbor search at scale
Query: What embedding model should I use?
[0.1567] OpenAI text-embedding-3-small produces 1536-dimensional vectors
[0.2678] Cosine similarity measures the angle between two vectors
[0.3789] Dot product on normalized vectors equals cosine similarity
Query: Explain cosine similarity
[0.0987] Cosine similarity measures the angle between two vectors
[0.1876] Dot product on normalized vectors equals cosine similarity
[0.2987] OpenAI text-embedding-3-small produces 1536-dimensional vectorsLower distance means higher similarity. The results make semantic sense — the SQLite query returns the sqlite-vec document first, not the PostgreSQL one.
Chunking strategy for real documents
Raw documents rarely fit in a single embedding. You need a chunking strategy that preserves semantic boundaries. Here’s a production-ready chunker using recursive character splitting with overlap:
# chunk.py
from typing import List
import re
def chunk_text(text: str, chunk_size: int = 800, overlap: int = 100) -> List[str]:
"""Split text into overlapping chunks at semantic boundaries."""
if len(text) <= chunk_size:
return [text]
# Split on paragraph boundaries first
paragraphs = re.split(r'\n\s*\n', text)
chunks = []
current_chunk = ""
for para in paragraphs:
para = para.strip()
if not para:
continue
if len(current_chunk) + len(para) + 2 <= chunk_size:
current_chunk += ("\n\n" if current_chunk else "") + para
else:
if current_chunk:
chunks.append(current_chunk)
# If paragraph itself is too large, split it
if len(para) > chunk_size:
chunks.extend(_split_large_paragraph(para, chunk_size, overlap))
current_chunk = ""
else:
current_chunk = para
if current_chunk:
chunks.append(current_chunk)
# Add overlap between chunks
if overlap > 0 and len(chunks) > 1:
overlapped = [chunks[0]]
for i in range(1, len(chunks)):
prev_tail = chunks[i-1][-overlap:]
overlapped.append(prev_tail + "\n\n" + chunks[i])
return overlapped
return chunks
def _split_large_paragraph(para: str, chunk_size: int, overlap: int) -> List[str]:
"""Split a large paragraph by sentences, then by characters if needed."""
sentences = re.split(r'(?<=[.!?])\s+', para)
chunks = []
current = ""
for sent in sentences:
if len(current) + len(sent) + 1 <= chunk_size:
current += (" " if current else "") + sent
else:
if current:
chunks.append(current)
if len(sent) > chunk_size:
# Hard split
for i in range(0, len(sent), chunk_size - overlap):
chunks.append(sent[i:i + chunk_size])
current = ""
else:
current = sent
if current:
chunks.append(current)
return chunksIntegrate it into the ingestion pipeline:
# ingest.py
import sqlite3
import sqlite_vec
from embed import get_embeddings, EMBEDDING_DIM
from chunk import chunk_text
from store import init_db, DB_PATH
def ingest_file(conn, filepath: str, source: str = None):
with open(filepath, 'r') as f:
text = f.read()
source = source or filepath
chunks = chunk_text(text)
print(f"Split {filepath} into {len(chunks)} chunks")
embeddings = get_embeddings(chunks)
cursor = conn.cursor()
for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
cursor.execute(
"INSERT INTO documents (content, source, chunk_id) VALUES (?, ?, ?)",
(chunk, source, i)
)
doc_id = cursor.lastrowid
cursor.execute(
"INSERT INTO documents_vec (rowid, embedding, content, source, chunk_id) VALUES (?, ?, ?, ?, ?)",
(doc_id, emb.tobytes(), chunk, source, i)
)
conn.commit()
if __name__ == "__main__":
conn = init_db()
# Example: ingest a markdown file
# ingest_file(conn, "your_docs.md", source="documentation")
print("Ready to ingest. Call ingest_file(conn, 'path/to/file.md')")Hybrid search: combining vector and keyword
Pure vector search misses exact matches (error codes, function names, acronyms). Hybrid search combines BM25 keyword scores with vector similarity. SQLite’s FTS5 handles BM25 natively.
# hybrid.py
import sqlite3
import sqlite_vec
import numpy as np
from embed import get_embeddings, EMBEDDING_DIM
DB_PATH = "semantic_search.db"
def init_hybrid_db():
conn = sqlite3.connect(DB_PATH)
conn.enable_load_extension(True)
sqlite_vec.load(conn)
conn.enable_load_extension(False)
conn.execute(f"""
CREATE VIRTUAL TABLE IF NOT EXISTS documents_vec
USING vec0(
embedding float[{EMBEDDING_DIM}],
+content TEXT,
+source TEXT,
+chunk_id INTEGER
)
""")
conn.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts
USING fts5(
content,
source,
chunk_id,
content='documents',
content_rowid='id'
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
source TEXT,
chunk_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Triggers to keep FTS in sync
conn.execute("""
CREATE TRIGGER IF NOT EXISTS documents_ai AFTER INSERT ON documents BEGIN
INSERT INTO documents_fts (rowid, content, source, chunk_id)
VALUES (new.id, new.content, new.source, new.chunk_id);
END
""")
conn.execute("""
CREATE TRIGGER IF NOT EXISTS documents_ad AFTER DELETE ON documents BEGIN
DELETE FROM documents_fts WHERE rowid = old.id;
END
""")
conn.execute("""
CREATE TRIGGER IF NOT EXISTS documents_au AFTER UPDATE ON documents BEGIN
DELETE FROM documents_fts WHERE rowid = old.id;
INSERT INTO documents_fts (rowid, content, source, chunk_id)
VALUES (new.id, new.content, new.source, new.chunk_id);
END
""")
conn.commit()
return conn
def hybrid_search(conn, query: str, k: int = 10, alpha: float = 0.5) -> list[dict]:
"""
Combine vector and BM25 scores.
alpha=1.0: pure vector, alpha=0.0: pure keyword.
"""
query_emb = get_embeddings([query])[0]
query_bytes = query_emb.tobytes()
# Get more candidates than needed for fusion
candidate_k = k * 4
# Vector search
vec_rows = conn.execute("""
SELECT rowid, vec_distance_cosine(embedding, ?) as vec_dist
FROM documents_vec
WHERE embedding MATCH ?
ORDER BY vec_dist
LIMIT ?
""", (query_bytes, query_bytes, candidate_k)).fetchall()
vec_scores = {rowid: 1.0 - dist for rowid, dist in vec_rows} # similarity
# BM25 search
fts_rows = conn.execute("""
SELECT rowid, bm25(documents_fts) as bm25_score
FROM documents_fts
WHERE documents_fts MATCH ?
ORDER BY bm25_score
LIMIT ?
""", (query, candidate_k)).fetchall()
# Normalize BM25 scores (lower is better) to 0-1 similarity
if fts_rows:
max_bm25 = max(r[1] for r in fts_rows)
min_bm25 = min(r[1] for r in fts_rows)
bm25_range = max_bm25 - min_bm25 or 1.0
bm25_scores = {
rowid: 1.0 - (score - min_bm25) / bm25_range
for rowid, score in fts_rows
}
else:
bm25_scores = {}
# Fuse scores
all_ids = set(vec_scores.keys()) | set(bm25_scores.keys())
fused = []
for rowid in all_ids:
v = vec_scores.get(rowid, 0.0)
b = bm25_scores.get(rowid, 0.0)
fused_score = alpha * v + (1 - alpha) * b
fused.append((rowid, fused_score))
fused.sort(key=lambda x: x[1], reverse=True)
top_ids = [rowid for rowid, _ in fused[:k]]
if not top_ids:
return []
placeholders = ','.join('?' * len(top_ids))
rows = conn.execute(f"""
SELECT id, content, source, chunk_id
FROM documents
WHERE id IN ({placeholders})
""", top_ids).fetchall()
# Preserve fused score order
id_to_row = {r[0]: r for r in rows}
results = []
for rowid, score in fused[:k]:
if rowid in id_to_row:
r = id_to_row[rowid]
results.append({
"id": r[0],
"content": r[1],
"source": r[2],
"chunk_id": r[3],
"score": score
})
return resultsTest the hybrid approach:
# test_hybrid.py
from hybrid import init_hybrid_db, hybrid_search
from store import insert_documents
conn = init_hybrid_db()
# Add test data with specific keywords
corpus = [
"Error code E_CONN_REFUSED indicates the database connection was rejected",
"The function connect_to_database() returns a connection pool",
"Vector similarity search uses cosine distance between embeddings",
"PostgreSQL pgvector extension supports HNSW and IVF indexes",
"SQLite sqlite-vec provides zero-config vector search",
"Retry logic with exponential backoff handles transient failures",
]
insert_documents(conn, corpus, source="api_docs")
queries = [
"E_CONN_REFUSED", # exact keyword match
"database connection pool", # semantic match
"vector index types", # mixed
]
for q in queries:
print(f"\nQuery: {q}")
print(" Hybrid (alpha=0.5):")
for r in hybrid_search(conn, q, k=3, alpha=0.5):
print(f" [{r['score']:.3f}] {r['content'][:70]}...")
print(" Pure vector (alpha=1.0):")
for r in hybrid_search(conn, q, k=3, alpha=1.0):
print(f" [{r['score']:.3f}] {r['content'][:70]}...")Expected output shows the difference:
Query: E_CONN_REFUSED
Hybrid (alpha=0.5):
[0.892] Error code E_CONN_REFUSED indicates the database connection was rejected
[0.234] Retry logic with exponential backoff handles transient failures
[0.156] The function connect_to_database() returns a connection pool
Pure vector (alpha=1.0):
[0.734] Error code E_CONN_REFUSED indicates the database connection was rejected
[0.312] The function connect_to_database() returns a connection pool
[0.289] Retry logic with exponential backoff handles transient failuresThe hybrid version ranks the exact error code match higher. Tune alpha per use case: 0.7 for general Q&A, 0.3 for code/error lookup.
Reranking for precision
Vector search (even hybrid) retrieves candidates. A cross-encoder reranker scores query-document pairs directly, trading latency for precision. OpenAI doesn’t offer a cross-encoder API, but you can call a local model via sentence-transformers or use a reranking API like Cohere’s.
# rerank.py
import requests
import os
from typing import List, Dict
COHERE_API_KEY = os.getenv("COHERE_API_KEY")
RERANK_MODEL = "rerank-v3.5"
def cohere_rerank(query: str, documents: List[str], top_k: int = 5) -> List[Dict]:
if not COHERE_API_KEY:
raise ValueError("COHERE_API_KEY not set")
response = requests.post(
"https://api.cohere.ai/v1/rerank",
headers={
"Authorization": f"Bearer {COHERE_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": RERANK_MODEL,
"query": query,
"documents": documents,
"top_n": top_k,
"return_documents": False
}
)
response.raise_for_status()
return response.json()["results"]
def rerank_results(query: str, results: List[Dict], top_k: int = 5) -> List[Dict]:
if not results:
return []
docs = [r["content"] for r in results]
reranked = cohere_rerank(query, docs, top_k)
# Map back to original results
final = []
for item in reranked:
idx = item["index"]
original = results[idx].copy()
original["rerank_score"] = item["relevance_score"]
final.append(original)
return finalIntegration point in your search flow:
# search_flow.py
from hybrid import hybrid_search, init_hybrid_db
from rerank import rerank_results
def search_with_rerank(conn, query: str, k: int = 5, rerank_k: int = 20):
# Retrieve more candidates for reranker
candidates = hybrid_search(conn, query, k=rerank_k, alpha=0.5)
if not candidates:
return []
# Rerank top candidates
reranked = rerank_results(query, candidates, top_k=k)
return rerankedProduction considerations
Index maintenance
sqlite-vec uses brute-force search by default. For datasets above ~100k vectors, build an HNSW index:
-- Run after bulk inserts
CREATE INDEX IF NOT EXISTS documents_vec_hnsw
ON documents_vec (embedding)
USING hnsw;Monitor query latency. If p99 exceeds your budget, migrate to pgvector, Pinecone, or a managed service. The query logic stays identical — only the connection and index creation change.
Embedding versioning
Models change. Store the model name with every vector:
ALTER TABLE documents_vec ADD COLUMN embedding_model TEXT DEFAULT 'text-embedding-3-small';When you migrate models, re-embed incrementally. Keep both versions during transition and route queries to the appropriate index.
Cost control
text-embedding-3-small costs $0.02 per 1M tokens. A 100k document corpus at 800 tokens/chunk with 1.5 chunks/document costs roughly $2.40 to embed once. Budget for re-embedding when models improve.
If you’re routing through a gateway that meters per-token usage across providers, you can enforce budgets and fallback policies centrally — n4n.ai exposes usage headers on every response so you can track spend in real time without instrumenting each call.
Filtering by metadata
Real applications filter by tenant, date, document type. Add columns to the vector table:
CREATE VIRTUAL TABLE documents_vec USING vec0(
embedding float[1536],
+content TEXT,
+source TEXT,
+chunk_id INTEGER,
+tenant_id TEXT,
+doc_type TEXT,
+created_at TEXT
);Then filter in the query:
SELECT ... FROM documents_vec
WHERE embedding MATCH ?
AND tenant_id = ?
AND doc_type = 'api_reference'
ORDER BY distance LIMIT ?Scaling beyond SQLite
When you need distributed search, horizontal scaling, or managed infrastructure, the migration is mechanical. Here’s the pgvector equivalent of the core search:
# pgvector_search.py
import psycopg
import numpy as np
from embed import get_embeddings, EMBEDDING_DIM
def pg_search(conn, query: str, k: int = 10):
query_emb = get_embeddings([query])[0]
with conn.cursor() as cur:
cur.execute("""
SELECT content, source, chunk_id,
1 - (embedding <=> %s) as similarity
FROM documents
ORDER BY embedding <=> %s
LIMIT %s
""", (query_emb, query_emb, k))
return [
{"content": r[0], "source": r[1], "chunk_id": r[2], "similarity": r[3]}
for r in cur.fetchall()
]The <=> operator is cosine distance in pgvector. For HNSW:
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);Set hnsw.ef_search at query time for latency/recall tradeoff:
SET hnsw.ef_search = 100;What to build next
You now have a working semantic search stack: embedding generation, chunking, vector storage, hybrid retrieval, and reranking. The next problems are almost always product-specific:
- Evaluation: Build a labeled test set and measure recall@k, MRR, and latency percentiles
- Query understanding: Rewrite, expand, or decompose queries before embedding
- Multi-vector representations: ColBERT-style late interaction for longer documents
- Incremental updates: Change data capture from your primary store to the vector index
- Observability: Log queries, results, and user feedback (clicks, thumbs up/down) to detect drift
Start with evaluation. Without it, every change is a guess.