Building Context-Aware Search in Python with LLM Embeddings + Metadata
This article walks through building a context-aware semantic search engine that combines embedding-based similarity with structured metadata filtering, covering everything from generating embeddings to persisting the index.
Building Context-Aware Search in Python with LLM Embeddings + Metadata
Building Context-Aware Search in Python with LLM Embeddings + Metadata
In this article, you will learn how to build a context-aware semantic search engine in Python that combines embedding-based similarity with structured metadata filtering.
Topics we will cover include:
How sentence embeddings and cosine similarity work together to find semantically relevant documents.
How to build a metadata-aware search index that filters by team, status, priority, and date before scoring candidates.
How to persist the index to disk so embeddings are computed only once and reloaded efficiently on subsequent runs.
Building Context-Aware Search in Python with LLM Embeddings + Metadata
Introduction
Keyword search breaks the moment a user types something a document doesn’t literally say. A support engineer searching for “login keeps failing” won’t find a ticket titled “OAuth2 token refresh race condition”, even though that’s exactly what they need. This is the core problem that context-aware semantic search aims to solve.
Semantic search solves this by converting text into dense vector representations called embeddings, where meaning determines proximity rather than exact word overlap. Layer structured metadata filters on top — by date, status, team, priority — and you get a system that understands what someone is asking while respecting contextual constraints at the same time.
This article walks through building that system end-to-end: embeddings from a local pretrained model, a metadata-aware index, cosine similarity ranking, and an index that persists across restarts without requiring re-encoding.
You can get the code on GitHub.
What You Will Build
A simple context-aware search engine over a corpus of engineering support tickets. By the end you will have:
384-dimensional embeddings generated locally from a pretrained model, no API key required
A search index that filters by team, status, priority, and date before scoring
Cosine similarity ranking over the filtered candidate pool
A persisted index that reloads without re-encoding
Prerequisites: Python 3.8+, basic familiarity with NumPy and working with lists of dictionaries.
Install dependencies:
1
pip install sentence-transformers numpy
Understanding How Semantic Search Works
A sentence embedding model takes a string and returns a fixed-length vector of floating-point numbers. The model is trained so that sentences with similar meanings produce vectors pointing in similar directions in high-dimensional space.
Cosine similarity measures the angle between two vectors:
\[
\text{cosine similarity}(A, B) =
\frac{A \cdot B}{\|A\| \, \|B\|}
\]
When vectors are unit-normalized — meaning their length equals 1.0 — this simplifies to the dot product: A · B. Scores range from -1 (opposite) to 1 (identical). In practice, unrelated documents score around 0.1–0.25, and strong matches score above 0.6.
So why does metadata filtering matter? Embedding models encode semantic content. They do not encode who wrote a document, what team owns it, or when it was created. These attributes live outside the text and must be handled separately. Combining both signals — semantic score and metadata constraints — is what makes search useful in real systems.
Setting Up the Dataset
We’ll work with 20 engineering support tickets across three teams — infrastructure, backend, and frontend — with four priority levels, two statuses, and a two-month date window.
Each ticket is a plain dictionary. The text field is what gets embedded; everything else is metadata for filtering.
To keep things concise, a truncated list is shown here instead of the full code block. The complete set of tickets is available in this GitHub gist.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from datetime import date
tickets = [
{"id": "T-101", "team": "infrastructure", "status": "open", "priority": "high",
"created": date(2025, 11, 3),
"text": "Kubernetes pod keeps crashing with OOMKilled — memory limits on the ML inference container are set too low for the model it loads at runtime."},
{"id": "T-102", "team": "infrastructure", "status": "open", "priority": "high",
"created": date(2025, 11, 8),
"text": "Nginx ingress returning 502 after rotating TLS certificate. Chain is valid per openssl verify but the backend handshake fails immediately."},
{"id": "T-103", "team": "infrastructure", "status": "resolved", "priority": "medium",
"created": date(2025, 10, 14),
"text": "Terraform state file locked in S3 — a team member force-applied a plan without releasing the DynamoDB lock first."},
...
{"id": "T-401", "team": "infrastructure", "status": "open", "priority": "medium",
"created": date(2025, 11, 11),
"text": "CI pipeline fails on ARM64 runners — base Docker image has no ARM variant, exec format error at build stage."},
{"id": "T-402", "team": "infrastructure", "status": "resolved", "priority": "high",
"created": date(2025, 10, 9),
"text": "VPN gateway latency spikes at peak hours — BGP route flapping between two peers causing intermittent packet loss across the private subnet."},
]
A quick check on the shape of the corpus before moving on:
1
2
3
open_ct = sum(1 for t in tickets if t["status"] == "open")
resolved_ct = sum(1 for t in tickets if t["status"] == "resolved")
print(f"{len(tickets)} tickets | {open_ct} open | {resolved_ct} resolved")
Output:
1
20 tickets | 14 open | 6 resolved
Running the snippet confirms the distribution: 20 tickets total, 14 open and 6 resolved, spread across the three teams.
Step 1: Generating Embeddings
all-MiniLM-L6-v2 maps any sentence to a 384-dimensional vector. It runs entirely on CPU, downloads once from Hugging Face (~22 MB), is cached locally after that, and requires no API key.
1
2
3
4
5
6
7
8
9
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer("all-MiniLM-L6-v2")
texts = [t["text"] for t in tickets]
embeddings = model.encode(texts, normalize_embeddings=True, show_progress_bar=True)
print(f"Shape: {embeddings.shape} | norm[0]: {np.linalg.norm(embeddings[0]):.4f}")
We pass normalize_embeddings=True so each output vector comes out with L2 norm exactly 1.0. Once vectors sit on the unit hypersphere, cosine similarity between any two of them is just their dot product, so no division is needed at query time. That means scoring the entire candidate pool reduces to a single matrix multiplication.
Output:
Sentence Embeddings for 20 Tickets
We get back a (20, 384) float32 matrix — one row per ticket. The norm of 1.0 confirms the normalization worked.
Step 2: Building the Index
The index stores the embedding matrix alongside the associated metadata and exposes a search method that accepts optional keyword arguments for every metadata field.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class ContextAwareIndex:
def init(self, embeddings: np.ndarray, documents: list):
self.embeddings = embeddings # (N, D), L2-normalized
self.documents = documents
def search(
self,
query: str,
top_k: int = 5,
team: str = None,
status: str = None,
priority: str = None,
after: "date" = None,
before: "date" = None,
min_score: float = 0.0,
) -> list[dict]:
Embed the query into the same vector space as the documents
q_vec = model.encode([query], normalize_embeddings=True)[0]
Build a boolean mask — False for any document that fails a filter condition
mask = np.ones(len(self.documents), dtype=bool)
for i, doc in enumerate(self.documents):
if team and doc["team"] != team: mask[i] = False
if status and doc["status"] != status: mask[i] = False
if priority and doc["priority"] != priority: mask[i] = False
if after and doc["created"] before: mask[i] = False
candidate_idx = np.where(mask)[0]
if len(candidate_idx) == 0:
return []
Score only the candidates that passed the filter
scores = self.embeddings[candidate_idx] @ q_vec
Drop anything below the minimum score threshold, sort, return top-k
valid = np.where(scores >= min_score)[0]
if len(valid) == 0:
return []
top_local = np.argsort(scores[valid])[::-1][:top_k]
top_global = candidate_idx[valid[top_local]]
return [
{**self.documents[i], "score": float(scores[valid[top_local[j]]])}
for j, i in enumerate(top_global)
]
index = ContextAwareIndex(embeddings, tickets)
The key design decision here is filtering before scoring, not after. Post-hoc filtering wastes dot-product compute on documents you’d discard anyway. Filtering first also ensures min_score can drop irrelevant results instead of returning noisy low-confidence matches.
Step 3: Running Queries
We’ll run three queries to show different aspects of the system: semantic search alone, the same query with metadata filters, and a cross-team query scoped by priority.
First, a small helper that formats results consistently across all three examples.
Query 1: Searching Without Filters
To establish a baseline, we search without any metadata constraints, letting the embedding model rank the full corpus on semantic similarity alone.
1
2
results = index.search("authentication token expiry and session management", top_k=4)
show("'authentication token expiry and session management' (no filters)", results)
Running this against the full 20-ticket corpus returns the following four backend tickets:
1
2
3
4
5
6
7
8
9
Query: 'authentication token expiry and session management' (no filters)
[0.6133] T-207 backend open high 2025-11-03
Session cookie persists after logout — token blacklist check is missing from the midd...
[0.4958] T-201 backend open high 2025-11-05
OAuth2 token refresh fails intermittently — race condition in the token cache where t...
[0.3459] T-203 backend open medium 2025-11-01
JWT signature verification fails intermittently — clock skew of 4 seconds between the...
[0.1714] T-206 backend open high 2025-11-13
Rate limiting not scoping per user — middleware uses a shared Redis key derived from ...
Query 2: Filtering by Status and Date
The query text is identical to the previous one. What changes is the candidate pool: this time we restrict to open tickets created before November 10th, 2025, simulating a workflow where a team wants only unresolved issues within a certain window.
1
2
3
4
5
6
7
results = index.search(
"authentication token expiry and session management",
top_k=4,
status="open",
before=date(2025, 11, 10),
)
show("same query [status=open, before=2025-11-10]", results)
Output:
1
2
3
4
5
6
7
8
9
Query: same query [status=open, before=2025-11-10]
[0.6133] T-207 backend open high 2025-11-03
Session cookie persists after logout — token blacklist check is missing from the midd...
[0.4958] T-201 backend open high 2025-11-05
OAuth2 token refresh fails intermittently — race condition in the token cache where t...
[0.3459] T-203 backend open medium 2025-11-01
JWT signature verification fails intermittently — clock skew of 4 seconds between the...
[0.1419] T-202 backend open high 2025-11-09
Database connection pool exhausted under load — pool capped at 20 connections but the...
Query 3: Searching Across Teams with a Priority Filter
Resource exhaustion appears in both infrastructure and backend tickets; they share semantic territory regardless of team ownership. This query tests whether the model groups them correctly across that boundary.
1
2
3
4
5
6
7
results = index.search(
"resource exhaustion and memory pressure under load",
top_k=2,
status="open",
priority="high",
)
show("'resource exhaustion and memory pressure' [status=open, priority=high]", results)
This outputs:
1
2
3
[truncated for AI cost control]