Create Embeddings
On this page
Encode Chunks into Vectors
Your documents are now split into clean chunks. The next step is to transform them into embeddings using the mistral-embed API, then store these vectors in a vector store that will act as your search engine. This is the moment your corpus stops being plain text and becomes queryable geometry. A few engineering precautions—batching, retries, and persistence—make the difference between a fragile prototype and a reliable index.
For a small corpus, say fewer than a hundred chunks, a single direct call suffices. Extract the text from your chunk dictionaries, pass them as-is to the API, and retrieve the full matrix. The shape (87, 1024) displayed at the end of execution is your first sanity check: as many rows as chunks, and 1024 columns because that is the dimension of mistral-embed. If the row count doesn’t match, an empty chunk slipped into the batch.
import numpy as np
from mistralai.client import Mistral
client = Mistral(api_key=api_key)
# Assume chunks is the list of our prepared chunks
chunk_texts = [c["text"] for c in chunks]
# Encode in a single call
response = client.embeddings.create(
model="mistral-embed",
inputs=chunk_texts,
)
embeddings = np.array([d.embedding for d in response.data])
print(f"Embeddings: {embeddings.shape}")
# Embeddings: (87, 1024) for exampleBatch Encoding
Beyond a few dozen chunks, a single call hits API limits. The workaround is to split the list into batches and concatenate the results. The function below processes slices of 25 and displays progress. This isn’t cosmetic: on a corpus of several thousand chunks, encoding takes several minutes, and you’ll want to know where the process stands rather than staring at a silent terminal. Order is preserved, so each vector remains aligned with its corresponding chunk in your original list.
def encode_chunks(chunks, batch_size=25, model="mistral-embed"):
"""Encode a list of chunks in batches."""
texts = [c["text"] for c in chunks]
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
print(f" Encoding batch {i // batch_size + 1}/{(len(texts) - 1) // batch_size + 1} "
f"({len(batch)} chunks)...")
response = client.embeddings.create(
model=model,
inputs=batch,
)
all_embeddings.extend([d.embedding for d in response.data])
return np.array(all_embeddings)
# Encode all chunks
embeddings = encode_chunks(chunks, batch_size=25)
print(f"Total: {embeddings.shape[0]} vectors of {embeddings.shape[1]} dimensions")Surviving Network Errors
In production, an API call will eventually fail: network drop, rate limit hit, or temporary outage. Without protection, your indexing stops at batch 180 out of 400, and you’re back to square one. The robust version wraps each batch in a retry loop with exponential backoff—one second, then two, then four. This increasing spacing matters: retrying immediately after a quota rejection only makes things worse. After three consecutive failures, the function raises an explicit exception naming the faulty batch, rather than returning a truncated matrix where you’d only discover the gap during search.
import time
def encode_chunks_robust(chunks, batch_size=25, max_retries=3):
"""Robust encoding with error handling."""
texts = [c["text"] for c in chunks]
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
for attempt in range(max_retries):
try:
response = client.embeddings.create(
model="mistral-embed",
inputs=batch,
)
all_embeddings.extend([d.embedding for d in response.data])
break # Success, exit retry loop
except Exception as e:
wait_time = 2 ** attempt # Exponential backoff: 1s, 2s, 4s
print(f" Error in batch {i // batch_size + 1}, "
f"attempt {attempt + 1}/{max_retries}: {e}")
if attempt < max_retries - 1:
print(f" Waiting {wait_time}s before retry...")
time.sleep(wait_time)
else:
raise RuntimeError(
f"Failed after {max_retries} attempts for batch {i // batch_size + 1}"
)
return np.array(all_embeddings)Save the Index to Disk
Every encoding run costs time and billed API calls. Recalculating vectors every time your application starts is pure waste, especially since the corpus itself hasn’t changed. Saving happens in two complementary files: vectors go into NumPy, a compact binary format that loads instantly, while metadata—text, source, position—goes into readable JSON that you can open to inspect a suspicious chunk. The two files are interdependent, linked by line order: never regenerate one without the other, or your search results will point to the wrong texts.
import json
def save_index(chunks, embeddings, filepath="rag_index"):
"""Save chunks and embeddings."""
# Save embeddings (NumPy)
np.save(f"{filepath}_embeddings.npy", embeddings)
# Save metadata (JSON)
metadata = []
for c in chunks:
metadata.append({
"text": c["text"],
"source": c.get("source", ""),
"chunk_index": c.get("chunk_index", 0),
})
with open(f"{filepath}_metadata.json", "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
print(f"Index saved: {len(chunks)} chunks, {embeddings.shape}")
def load_index(filepath="rag_index"):
"""Load chunks and embeddings from disk."""
embeddings = np.load(f"{filepath}_embeddings.npy")
with open(f"{filepath}_metadata.json", "r", encoding="utf-8") as f:
metadata = json.load(f)
print(f"Index loaded: {len(metadata)} chunks, {embeddings.shape}")
return metadata, embeddings
# Save
save_index(chunks, embeddings)
# Load (on next startup)
chunks_loaded, embeddings_loaded = load_index()Verify Before Continuing
A silently corrupted index shows up late, usually when a user reports an absurd answer. A few offline checks take thirty seconds and save you from that discovery. The verification function first displays basic metrics—number of chunks, dimension, average norm—then measures similarity between neighboring chunks: since they come from the same document, they should show significantly higher values than two random passages. The last test is the simplest and most revealing: a chunk compared to itself must yield exactly 1.0. Any other value signals an alignment issue or numerical conversion problem.
from sklearn.metrics.pairwise import cosine_similarity
def verify_embeddings(chunks, embeddings, n_tests=5):
"""Verify embedding quality with simple tests."""
print("=== Verifying embeddings ===")
print(f"Number of chunks: {len(chunks)}")
print(f"Dimensions: {embeddings.shape[1]}")
print(f"Average norm: {np.linalg.norm(embeddings, axis=1).mean():.4f}")
# Test: chunks from the same document should be closer
for i in range(min(n_tests, len(chunks) - 1)):
sim = cosine_similarity([embeddings[i]], [embeddings[i + 1]])[0][0]
print(f" Similarity chunk {i} <-> {i+1}: {sim:.4f}")
# Test: a chunk compared to itself = 1.0
self_sim = cosine_similarity([embeddings[0]], [embeddings[0]])[0][0]
print(f" Self-similarity (must be 1.0): {self_sim:.6f}")
verify_embeddings(chunks, embeddings)Strung together, the workflow boils down to four useful lines. You prepare chunks with the pipeline from the previous lesson, encode in robust mode, verify, and save. This is the script you’ll rerun every time the corpus updates.
# 1. Prepare chunks (previous lesson)
chunks = prepare_documents(raw_docs)
# 2. Encode
embeddings = encode_chunks_robust(chunks, batch_size=25)
# 3. Verify
verify_embeddings(chunks, embeddings)
# 4. Save
save_index(chunks, embeddings)
print("Index ready for search!")Key Takeaways
- Use batching to encode large corpora (batches of 25–50 chunks)
- Add retry logic with exponential backoff for robustness
- Save embeddings to disk (NumPy) to avoid recalculation
- Keep metadata (text, source, position) in a separate JSON file
- Verify embedding quality before using them in production