Wire the pieces into one repeatable pipeline: load, chunk, embed, store. Then handle the part everyone forgets — re-ingesting documents when they change.
Why: ingestion is a straight pipeline that turns raw documents into a searchable index — get it repeatable so re-running it is safe. When: run it once up front and again whenever sources change. Where: each stage is swappable (loader, chunker, embedder, store) without touching the others.
LOAD read files / URLs / DB rows, extract clean text
|
CHUNK split into retrievable pieces (+ metadata)
|
EMBED vector per chunk (same model as query time!)
|
STORE upsert vector + content + metadata into the DBWhy: seeing the four stages as one function makes the data flow obvious and testable. When: build the index into FAISS here for learning; the same shape targets pgvector or Pinecone by changing the store step. Where: keep the chunk texts and metadata parallel to the index so a hit maps back to a citable passage.
import faiss, numpy as np
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")
def build_index(docs: dict[str, str]): # {source: full_text}
texts, meta = [], []
for source, body in docs.items():
for i, c in enumerate(chunk(body)): # from the previous lesson
texts.append(c)
meta.append({"source": source, "chunk": i})
vecs = model.encode(texts, normalize_embeddings=True).astype("float32")
index = faiss.IndexFlatIP(vecs.shape[1])
index.add(vecs)
return index, texts, meta
index, texts, meta = build_index({"handbook.txt": open("handbook.txt").read()})Why: documents change — if you only ever append, the index fills with stale duplicates and retrieval returns outdated answers. When: on update, delete the old chunks for that source and insert the new ones (upsert), keyed by a stable id. Where: content-hash each chunk so you skip re-embedding text that did not actually change.
import hashlib
def chunk_id(source, i):
return f"{source}::{i}"
def content_hash(text):
return hashlib.sha256(text.encode()).hexdigest()
# Upsert strategy with a real store (pseudocode over your DB client):
# 1. Compute chunk_id + content_hash for each new chunk.
# 2. Skip chunks whose (id, hash) already exist -> no re-embedding.
# 3. Delete chunks for this source whose id is no longer present -> no stale data.
# 4. Embed + insert the rest.