Phase 0: LLM Fast Track · 50 min · OpenAI SDK · Sentence Transformers · NumPy
Embeddings & Semantic Search
Embeddings are the spatial representation of meaning — the foundational primitive underneath every RAG system, semantic router, and memory store.
What you will learn
- Generate embeddings using OpenAI and Sentence Transformers
- Implement cosine similarity search over a collection of documents
- Choose the right embedding model for accuracy vs. speed vs. cost trade-offs
The Problem
A keyword search for "affordable car" misses a document about "budget automobile." An embedding-based search finds it instantly because it understands that "affordable" and "budget" mean the same thing in this context, and "car" and "automobile" are synonyms.
Keyword search matches strings. Semantic search matches meaning. Everything we'll build in the later phases of this course — RAG systems, agent memory, semantic routers — runs on embeddings. This lesson gives you that foundation.
What you'll build
An EmbeddingStore class that can embed documents, store them efficiently in numpy, and retrieve the most semantically similar ones at query time. You'll also benchmark it against keyword search.
What embeddings are
Text → fixed-length vector. Similar meaning → similar direction in vector space.
When you pass "The cat sat on the mat" to an embedding model, it returns a vector of ~1536 numbers (for OpenAI) or ~384 numbers (for local models). These numbers encode the semantic content of the text. The key property:
Texts with similar meaning produce vectors that point in similar directions.
The classic demonstration:
embedding("king") - embedding("man") + embedding("woman") ≈ embedding("queen")
This is not magic — it's the model having learned to place semantically related concepts nearby in vector space during training.
Why This Enables Semantic Search
Instead of asking "does this document contain the word 'affordable'?", we ask "is this document's meaning close to the query's meaning?" — measured by the angle between their vectors.
The pipeline:
Query text → embedding model → query vector
↓
compare angle with every document vector
↓
return documents with smallest angle (highest similarity)
Dimensions matter for cost and performance
OpenAI text-embedding-3-small: 1536 dimensions, ~$0.02/1M tokens, best accuracy/cost balance OpenAI text-embedding-3-large: 3072 dimensions, ~$0.13/1M tokens, highest accuracy all-MiniLM-L6-v2 (Sentence Transformers): 384 dimensions, free, offline, fast
Generating and comparing embeddings
OpenAI Embeddings API
from openai import OpenAI
client = OpenAI()
def embed_openai(texts: list[str]) -> list[list[float]]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=texts, # batch up to 2048 texts at once
)
return [item.embedding for item in response.data]
Always batch your embedding calls. Sending 100 texts in one API call costs the same tokens as 100 individual calls, but with 100x fewer round trips.
Local Embeddings with Sentence Transformers
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
def embed_local(texts: list[str]) -> list[list[float]]:
return model.encode(texts, batch_size=32).tolist()
No API key, no cost, runs offline. Quality is lower than OpenAI large but usually sufficient for many tasks.
Cosine Similarity
Cosine similarity measures the angle between two vectors, not their length. This makes it robust to documents of different lengths (longer documents have higher magnitude vectors, but cosine normalizes for that).
cosine_similarity(A, B) = (A · B) / (|A| × |B|)
Range: -1.0 (opposite) to 1.0 (identical). In practice for text, modern embedding models produce values between 0.3 and 1.0. Values above 0.8 indicate high semantic similarity.
import numpy as np
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
def batch_cosine_similarity(query: np.ndarray, corpus: np.ndarray) -> np.ndarray:
"""Compare one query against many documents efficiently."""
# corpus shape: (n_docs, n_dims)
norms = np.linalg.norm(corpus, axis=1)
query_norm = np.linalg.norm(query)
return corpus @ query / (norms * query_norm)
You have a product catalog of 500 items and need semantic search over descriptions. Which approach is appropriate?
500 vectors is tiny. A numpy matrix multiply for cosine similarity takes ~1ms. Vector databases add operational overhead (API keys, costs, network latency) that is completely unnecessary at this scale. Only move to a vector DB when you have 50k+ documents or need real-time updates at scale.
Vector databases: when you need them
For small collections (<100k docs), numpy is perfect. For larger collections, you need indexing:
| Collection Size | Approach |
|---|
| <50k docs | numpy cosine similarity (~1ms query) |
| 50k–1M docs | Chroma (local) or pgvector (Postgres) |
| 1M+ docs | Pinecone (managed) or Weaviate (self-hosted) |
| Latency-critical | Qdrant with HNSW index |
HNSW (Hierarchical Navigable Small World) is the graph-based approximate nearest-neighbor algorithm used by most vector databases. It trades a tiny amount of recall for 100–1000x faster queries at scale by building a hierarchical graph that can skip large chunks of the vector space during search.
Cosine similarity returns 0.0 for two vectors. What does this mean?
Cosine similarity measures the angle between vectors. 1.0 = same direction (identical meaning). -1.0 = opposite direction. 0.0 = 90° apart (orthogonal/unrelated). Note: in practice, modern embedding models rarely produce negative cosine similarities for text — 0.0 effectively means no semantic overlap. This is different from Euclidean distance, where 0.0 would mean identical vectors.
Build
Build it
Real embeddings come from a model, but the retrieval math is plain vector arithmetic you can verify by hand. This example hardcodes tiny 3-dimensional "embeddings" to show cosine similarity ranking a query against a small corpus.
import math
# Hand-crafted toy embeddings (3 dims) standing in for a real embedding model.
DOCS = {
"budget automobile deals": (0.9, 0.1, 0.0),
"luxury sports car review": (0.6, 0.0, 0.8),
"affordable used car guide": (0.85, 0.15, 0.05),
"vegetable soup recipe": (0.0, 0.9, 0.1),
}
def cosine_similarity(a, b):
dot = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(y * y for y in b))
return dot / (norm_a * norm_b)
query_vec = (0.88, 0.12, 0.02) # stand-in embedding for "cheap car"
scored = [(doc, cosine_similarity(query_vec, vec)) for doc, vec in DOCS.items()]
scored.sort(key=lambda pair: pair[1], reverse=True)
print("Query: 'cheap car'\n")
for rank, (doc, score) in enumerate(scored, start=1):
print(f"{rank}. [{score:.3f}] {doc}")
The code artifact implements an EmbeddingStore that works with both OpenAI and local Sentence Transformer models, with persistent storage and a keyword search benchmark.
"""
lesson_00_04_embeddings.py
EmbeddingStore class supporting:
- OpenAI text-embedding-3-small and local all-MiniLM-L6-v2
- Batch embedding with progress tracking
- Cosine similarity top-k retrieval with score threshold
- Persistent storage via .npz file (embed once, reuse)
- Benchmark: keyword search vs. semantic search on sample queries
Usage:
# With OpenAI (requires API key):
export OPENAI_API_KEY=sk-...
python lesson_00_04_embeddings.py --provider openai
# With local Sentence Transformers (no API key needed):
python lesson_00_04_embeddings.py --provider local
Requirements:
pip install openai sentence-transformers numpy
"""
import argparse
import os
import time
from pathlib import Path
from typing import Literal
import numpy as np
from openai import OpenAI
Provider = Literal["openai", "local"]
SAMPLE_DOCUMENTS = [
"How to reduce AWS EC2 costs by 60% using spot instances",
"Budget-friendly cloud computing strategies for startups",
"Machine learning model deployment on Kubernetes",
"Containerizing Python applications with Docker",
"Python asyncio and concurrent programming patterns",
"Understanding transformer architecture and attention mechanisms",
"Fine-tuning large language models for domain-specific tasks",
"Building RAG systems with vector databases",
"PostgreSQL query optimization and indexing strategies",
"Redis caching patterns for high-traffic web applications",
"Electric vehicles and sustainable transportation",
"The future of renewable energy: solar and wind",
"Deep learning for computer vision: CNNs explained",
"Natural language processing with BERT and GPT models",
"Data pipeline orchestration with Apache Airflow",
"Feature engineering for machine learning models",
"A/B testing statistical significance and sample sizes",
"Microservices architecture vs. monolithic applications",
"Serverless computing with AWS Lambda",
"Security best practices for API authentication and authorization",
]
BENCHMARK_QUERIES = [
"affordable computing for small companies",
"running ML models in production containers",
"speeding up database queries",
"training neural networks for text understanding",
"testing code reliability and correctness",
]
class EmbeddingStore:
def __init__(self, provider: Provider = "local"):
self.provider = provider
self._documents: list[str] = []
self._embeddings: np.ndarray | None = None
if provider == "openai":
self._openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self._openai_model = "text-embedding-3-small"
else:
from sentence_transformers import SentenceTransformer
self._st_model = SentenceTransformer("all-MiniLM-L6-v2")
def _embed_batch(self, texts: list[str]) -> np.ndarray:
if self.provider == "openai":
response = self._openai_client.embeddings.create(
model=self._openai_model,
input=texts,
)
vecs = [item.embedding for item in response.data]
else:
vecs = self._st_model.encode(texts, batch_size=32, show_progress_bar=False).tolist()
return np.array(vecs, dtype=np.float32)
def add_documents(self, documents: list[str], batch_size: int = 100) -> None:
print(f"Embedding {len(documents)} documents with {self.provider}...")
t0 = time.perf_counter()
all_embeddings: list[np.ndarray] = []
for i in range(0, len(documents), batch_size):
batch = documents[i : i + batch_size]
all_embeddings.append(self._embed_batch(batch))
print(f" {min(i + batch_size, len(documents))}/{len(documents)} done")
new_embeddings = np.vstack(all_embeddings)
if self._embeddings is None:
self._embeddings = new_embeddings
else:
self._embeddings = np.vstack([self._embeddings, new_embeddings])
self._documents.extend(documents)
elapsed = time.perf_counter() - t0
print(f"Embedding complete in {elapsed:.2f}s ({len(documents)/elapsed:.0f} docs/sec)")
def search(self, query: str, top_k: int = 5, threshold: float = 0.0) -> list[dict]:
if self._embeddings is None or len(self._documents) == 0:
return []
query_vec = self._embed_batch([query])[0]
corpus = self._embeddings
# Vectorized cosine similarity
norms = np.linalg.norm(corpus, axis=1)
query_norm = np.linalg.norm(query_vec)
similarities = corpus @ query_vec / (norms * query_norm + 1e-10)
# Get top-k above threshold
indices = np.argsort(similarities)[::-1]
results = []
for idx in indices:
score = float(similarities[idx])
if score < threshold:
break
results.append({
"document": self._documents[idx],
"score": round(score, 4),
"rank": len(results) + 1,
})
if len(results) >= top_k:
break
return results
def save(self, path: str) -> None:
np.savez_compressed(
path,
embeddings=self._embeddings,
documents=np.array(self._documents),
)
print(f"Saved {len(self._documents)} embeddings to {path}.npz")
def load(self, path: str) -> bool:
p = Path(path + ".npz")
if not p.exists():
return False
data = np.load(p, allow_pickle=True)
self._embeddings = data["embeddings"]
self._documents = list(data["documents"])
print(f"Loaded {len(self._documents)} embeddings from {p}")
return True
def keyword_search(query: str, documents: list[str], top_k: int = 5) -> list[dict]:
query_words = set(query.lower().split())
scored = []
for doc in documents:
doc_words = set(doc.lower().split())
overlap = len(query_words & doc_words)
scored.append({"document": doc, "score": overlap, "rank": 0})
scored.sort(key=lambda x: x["score"], reverse=True)
for i, r in enumerate(scored[:top_k]):
r["rank"] = i + 1
return scored[:top_k]
def run_benchmark(store: EmbeddingStore, documents: list[str]) -> None:
print("\n" + "=" * 70)
print("SEMANTIC vs. KEYWORD SEARCH BENCHMARK")
print("=" * 70)
for query in BENCHMARK_QUERIES:
print(f"\nQuery: '{query}'")
print(f"{'─'*65}")
semantic = store.search(query, top_k=3)
keyword = keyword_search(query, documents, top_k=3)
print(f"{'SEMANTIC (embedding)':^32} | {'KEYWORD (word overlap)':^32}")
print(f"{'─'*32}-+-{'─'*32}")
for i in range(max(len(semantic), len(keyword))):
sem_text = semantic[i]["document"][:30] if i < len(semantic) else ""
kw_text = keyword[i]["document"][:30] if i < len(keyword) else ""
sem_score = f"({semantic[i]['score']:.3f})" if i < len(semantic) else ""
print(f"{sem_text:<28} {sem_score:<4} | {kw_text}")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--provider", choices=["openai", "local"], default="local")
parser.add_argument("--cache", default="embeddings_cache", help="Cache file prefix")
args = parser.parse_args()
store = EmbeddingStore(provider=args.provider)
# Try to load cached embeddings first (avoid re-embedding on every run)
if not store.load(args.cache):
store.add_documents(SAMPLE_DOCUMENTS)
store.save(args.cache)
run_benchmark(store, SAMPLE_DOCUMENTS)
# Interactive search
print("\n\nINTERACTIVE SEARCH (type 'quit' to exit)")
print("=" * 70)
while True:
q = input("\nSearch query: ").strip()
if q.lower() in ("quit", "exit", "q"):
break
if not q:
continue
results = store.search(q, top_k=5, threshold=0.1)
for r in results:
print(f" [{r['score']:.3f}] {r['document']}")
if __name__ == "__main__":
main()
Unlock the full lesson
You've read the first 2 sections. The rest of this lesson covers Check Yourself — plus a hands-on lab, quiz, and project artifact.
Create a free account — the opening phases of 29 of 35 courses are free, no credit card.
Browse all courses · View pricing · DeVenture Academy