Building a personalised recommendation engine with embeddings
Why Embeddings Beat Traditional Collaborative Filtering
Collaborative filtering is the workhorse of most recommendation systems. It works by finding users who behaved similarly to you in the past and assuming you share future preferences. The approach has two well-known failure modes: the cold-start problem (new items and new users have no history) and sparsity (most users interact with a tiny fraction of the catalogue).
Embedding-based recommendations sidestep both problems by representing items and users as dense vectors in a shared semantic space. A new product can be embedded immediately from its description. A new user can be embedded from their stated preferences or even demographic signals. Similarity is then just a cosine distance computation.
Modern embedding models — whether text encoders like OpenAI text-embedding-3-small, open-source alternatives like nomic-embed-text, or domain-specific fine-tunes — produce 768-to-3072-dimensional vectors that capture nuanced semantic relationships. Two products described in completely different words will have similar embeddings if they serve the same need, something keyword matching can never achieve.
The pipeline covered in this article has four stages:
- Embed your item catalogue
- Embed user profiles or interaction history
- Store everything in pgvector with the right index
- Query at inference time with ANN (approximate nearest neighbour) search
Setting Up pgvector for Production
PostgreSQL with the pgvector extension is the most practical choice for teams already running Postgres. It keeps your recommendation data alongside your application data, eliminating a separate vector-database operational surface.
-- Enable the extension (requires pgvector installed on the server)
CREATE EXTENSION IF NOT EXISTS vector;
-- Items table with a 1536-dim embedding (OpenAI text-embedding-3-small)
CREATE TABLE items (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
title TEXT NOT NULL,
description TEXT,
category TEXT,
embedding VECTOR(1536),
created_at TIMESTAMPTZ DEFAULT NOW()
);
-- User preference profiles
CREATE TABLE user_profiles (
user_id UUID PRIMARY KEY,
embedding VECTOR(1536),
updated_at TIMESTAMPTZ DEFAULT NOW()
);
-- Interaction log — used to update user embeddings
CREATE TABLE interactions (
id BIGSERIAL PRIMARY KEY,
user_id UUID NOT NULL REFERENCES user_profiles(user_id),
item_id UUID NOT NULL REFERENCES items(id),
event_type TEXT NOT NULL CHECK (event_type IN ('view','click','purchase','skip')),
weight FLOAT NOT NULL DEFAULT 1.0,
occurred_at TIMESTAMPTZ DEFAULT NOW()
);
The HNSW index is the right choice for most production workloads. It offers sub-millisecond query times with recall rates above 95% for typical ef_search settings:
-- HNSW index for cosine similarity (inner product with normalised vectors)
CREATE INDEX items_embedding_hnsw
ON items
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
CREATE INDEX user_profiles_embedding_hnsw
ON user_profiles
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
The m parameter controls the number of bi-directional links per node (higher = better recall, more memory). ef_construction controls build-time quality. For a catalogue under 500k items, m=16, ef_construction=64 is a safe default. Benchmark your specific workload before tuning further.
Generating and Storing Item Embeddings
The embedding pipeline runs as a batch job on catalogue ingest and whenever items are updated. Python with the openai library and psycopg2 is a natural fit:
import openai
import psycopg2
import psycopg2.extras
from typing import List, Dict
client = openai.OpenAI()
def embed_texts(texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
"""Embed a batch of texts. OpenAI supports up to 2048 items per call."""
response = client.embeddings.create(input=texts, model=model)
return [item.embedding for item in response.data]
def build_item_text(item: Dict) -> str:
"""Construct a rich text representation of an item for embedding."""
parts = [item["title"]]
if item.get("description"):
parts.append(item["description"])
if item.get("category"):
parts.append(f"Category: {item['category']}")
if item.get("tags"):
parts.append(f"Tags: {', '.join(item['tags'])}")
return " | ".join(parts)
def upsert_item_embeddings(conn, items: List[Dict]) -> None:
texts = [build_item_text(item) for item in items]
embeddings = embed_texts(texts)
rows = [
(item["id"], item["title"], item.get("description"), item.get("category"), emb)
for item, emb in zip(items, embeddings)
]
with conn.cursor() as cur:
psycopg2.extras.execute_values(
cur,
"""
INSERT INTO items (id, title, description, category, embedding)
VALUES %s
ON CONFLICT (id) DO UPDATE
SET title = EXCLUDED.title,
description = EXCLUDED.description,
category = EXCLUDED.category,
embedding = EXCLUDED.embedding
""",
rows,
template="(%s, %s, %s, %s, %s::vector)"
)
conn.commit()
Run this in batches of 100-500 items depending on your average text length. OpenAI's rate limits and your database connection pool are the practical constraints. A simple queue with exponential backoff handles transient failures:
import time
def embed_catalogue_with_retry(conn, items: List[Dict], batch_size: int = 200) -> None:
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
for attempt in range(3):
try:
upsert_item_embeddings(conn, batch)
print(f"Embedded {i + len(batch)}/{len(items)} items")
break
except openai.RateLimitError:
wait = 2 ** attempt
print(f"Rate limited, retrying in {wait}s...")
time.sleep(wait)
Building and Updating User Embeddings
User embeddings represent the centroid of items a user has positively engaged with, weighted by interaction type. Purchases carry more signal than views; skips can carry negative weight.
The key design decision is how frequently to update user embeddings. Batch nightly updates work for low-activity systems. High-activity systems benefit from event-driven updates triggered on each interaction.
INTERACTION_WEIGHTS = {
"purchase": 3.0,
"click": 1.0,
"view": 0.5,
"skip": -0.5,
}
def compute_user_embedding(conn, user_id: str) -> List[float]:
"""
Compute a weighted average of item embeddings for all interactions
in the last 90 days, giving more recent interactions higher weight.
"""
with conn.cursor() as cur:
cur.execute(
"""
SELECT
i.embedding,
int.event_type,
-- Recency decay: interactions from today have weight 1.0,
-- from 90 days ago have weight ~0.05
EXP(-0.033 * EXTRACT(DAY FROM NOW() - int.occurred_at)) AS recency
FROM interactions int
JOIN items i ON i.id = int.item_id
WHERE int.user_id = %s
AND int.occurred_at > NOW() - INTERVAL '90 days'
""",
(user_id,)
)
rows = cur.fetchall()
if not rows:
return None
import numpy as np
weighted_sum = np.zeros(1536)
total_weight = 0.0
for embedding_str, event_type, recency in rows:
# pgvector returns embeddings as Python lists when using psycopg2
emb = np.array(embedding_str)
base_weight = INTERACTION_WEIGHTS.get(event_type, 0.5)
w = base_weight * recency
weighted_sum += w * emb
total_weight += abs(w)
if total_weight == 0:
return None
user_emb = weighted_sum / total_weight
# Normalise to unit length for cosine similarity
norm = np.linalg.norm(user_emb)
return (user_emb / norm).tolist() if norm > 0 else None
def refresh_user_embedding(conn, user_id: str) -> None:
embedding = compute_user_embedding(conn, user_id)
if embedding is None:
return
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO user_profiles (user_id, embedding, updated_at)
VALUES (%s, %s::vector, NOW())
ON CONFLICT (user_id) DO UPDATE
SET embedding = EXCLUDED.embedding,
updated_at = NOW()
""",
(user_id, embedding)
)
conn.commit()
Serving Recommendations at Query Time
With embeddings stored and indexed, retrieval is a single SQL query. The most common patterns are user-to-item (recommend items similar to the user's profile) and item-to-item (recommend items similar to a specific item):
def recommend_for_user(
conn,
user_id: str,
limit: int = 20,
exclude_interacted: bool = True,
category_filter: str | None = None
) -> List[Dict]:
"""Return top-N items nearest to the user's embedding."""
filters = ["up.user_id = %s"]
params = [user_id]
if category_filter:
filters.append("i.category = %s")
params.append(category_filter)
exclude_clause = ""
if exclude_interacted:
exclude_clause = """
AND i.id NOT IN (
SELECT item_id FROM interactions WHERE user_id = %s
)
"""
params.append(user_id)
params.append(limit)
query = f"""
SELECT
i.id,
i.title,
i.category,
1 - (i.embedding <=> up.embedding) AS similarity_score
FROM user_profiles up
CROSS JOIN items i
WHERE {' AND '.join(filters)}
{exclude_clause}
ORDER BY i.embedding <=> up.embedding
LIMIT %s
"""
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(query, params)
return cur.fetchall()
def recommend_similar_items(
conn,
item_id: str,
limit: int = 10
) -> List[Dict]:
"""Return items most similar to a given item (item-to-item)."""
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
"""
SELECT
i.id,
i.title,
i.category,
1 - (i.embedding <=> ref.embedding) AS similarity_score
FROM items ref
CROSS JOIN items i
WHERE ref.id = %s
AND i.id != ref.id
ORDER BY i.embedding <=> ref.embedding
LIMIT %s
""",
(item_id, limit)
)
return cur.fetchall()
The <=> operator is pgvector's cosine distance operator. Subtracting from 1 gives cosine similarity in [0,1]. For fast approximate results the HNSW index is used automatically; no query hints needed.
Production Concerns: Freshness, Diversity, and Business Rules
Raw nearest-neighbour results suffer from filter bubbles and ignore business constraints. A production system layers additional logic on top:
def apply_business_rules(
recommendations: List[Dict],
max_per_category: int = 3,
diversity_penalty: float = 0.1
) -> List[Dict]:
"""
Re-rank to enforce category diversity and apply any business rules.
Uses Maximal Marginal Relevance (MMR) lite: penalise items from
already-selected categories.
"""
selected = []
category_counts: Dict[str, int] = {}
for item in recommendations:
cat = item.get("category", "other")
count = category_counts.get(cat, 0)
# Apply diversity penalty for over-represented categories
adjusted_score = float(item["similarity_score"])
if count >= max_per_category:
adjusted_score *= (1.0 - diversity_penalty * count)
selected.append({**item, "adjusted_score": adjusted_score})
category_counts[cat] = count + 1
return sorted(selected, key=lambda x: x["adjusted_score"], reverse=True)
For cold-start users (no interaction history), fall back to item-to-item similarity on their stated preferences or use popularity-weighted items in their declared interests:
def recommend_cold_start(
conn,
interest_texts: List[str],
limit: int = 20
) -> List[Dict]:
"""
For new users with no interaction history.
Embed their stated interests and find nearest items.
"""
if not interest_texts:
# Ultimate fallback: return trending items
return get_trending_items(conn, limit)
combined = " | ".join(interest_texts)
embeddings = embed_texts([combined])
user_emb = embeddings[0]
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
cur.execute(
"""
SELECT id, title, category,
1 - (embedding <=> %s::vector) AS similarity_score
FROM items
ORDER BY embedding <=> %s::vector
LIMIT %s
""",
(user_emb, user_emb, limit)
)
return cur.fetchall()
Evaluating Recommendation Quality
Offline evaluation metrics matter but correlate imperfectly with business outcomes. Track these during development:
- Recall@K: fraction of items the user eventually interacted with that appeared in the top-K recommendations
- NDCG@K: normalised discounted cumulative gain, rewards recommendations that appear higher in the list
- Coverage: fraction of the catalogue that ever appears in any recommendation (guards against popularity bias)
- Intra-list diversity: average pairwise cosine distance among recommendations in a single list
import numpy as np
from typing import Set
def recall_at_k(recommended_ids: List[str], relevant_ids: Set[str], k: int) -> float:
top_k = set(recommended_ids[:k])
hits = len(top_k & relevant_ids)
return hits / len(relevant_ids) if relevant_ids else 0.0
def ndcg_at_k(recommended_ids: List[str], relevant_ids: Set[str], k: int) -> float:
dcg = 0.0
for rank, item_id in enumerate(recommended_ids[:k], start=1):
if item_id in relevant_ids:
dcg += 1.0 / np.log2(rank + 1)
# Ideal DCG: all relevant items at the top
ideal_dcg = sum(1.0 / np.log2(i + 2) for i in range(min(len(relevant_ids), k)))
return dcg / ideal_dcg if ideal_dcg > 0 else 0.0
Online A/B testing is the definitive measure. Split users into control (existing algorithm) and treatment (embedding-based) groups, then measure click-through rate, conversion rate, and revenue-per-session over a statistically significant window. Two weeks is the minimum for most catalogues to account for day-of-week variation.
The combination of dense embeddings with a well-indexed vector store gives you a recommendation system that handles cold-start, scales to millions of items, and can be built on infrastructure most teams already operate. The critical implementation details — recency weighting in user profiles, diversity enforcement in ranking, and correct HNSW index parameters — are what separate a prototype from a production system.