In this lesson we build a complete semantic search pipeline: documents go in, natural language queries come out with the most relevant chunks ranked by meaning. This is the foundation of every RAG system you will build.
from langchain.text_splitter import RecursiveCharacterTextSplitter
def chunk_document(text: str) -> list[str]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=512, # Characters per chunk
chunk_overlap=64, # Overlap to avoid cutting mid-sentence
separators=["
", "
", ". ", " "], # Try these in order
)
return splitter.split_text(text)
# A document with 5,000 characters → ~12 overlapping chunks
document = open("annual_report.txt").read()
chunks = chunk_document(document)
print(f"{len(chunks)} chunks created") # → ~12import chromadb
from openai import OpenAI
import hashlib
chroma = chromadb.PersistentClient(path="./search_db")
openai_client = OpenAI()
collection = chroma.get_or_create_collection(
name="documents",
metadata={"hnsw:space": "cosine"},
)
def ingest(chunks: list[str], source: str) -> None:
"""Embed and store chunks with source metadata."""
# Embed all chunks in one batched API call
response = openai_client.embeddings.create(
model="text-embedding-3-small",
input=chunks,
)
embeddings = [item.embedding for item in response.data]
# Deterministic IDs based on content — safe to re-ingest
ids = [hashlib.md5(c.encode()).hexdigest() for c in chunks]
collection.upsert(
ids=ids,
embeddings=embeddings,
documents=chunks,
metadatas=[{"source": source, "chunk_index": i} for i, _ in enumerate(chunks)],
)
print(f"Ingested {len(chunks)} chunks from '{source}'")def search(query: str, n_results: int = 5, source_filter: str | None = None) -> list[dict]:
"""Return top-n most semantically relevant chunks."""
query_embedding = openai_client.embeddings.create(
model="text-embedding-3-small",
input=query,
).data[0].embedding
where = {"source": source_filter} if source_filter else None
results = collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where,
include=["documents", "distances", "metadatas"],
)
return [
{
"text": doc,
"distance": dist,
"source": meta["source"],
"chunk_index": meta["chunk_index"],
}
for doc, dist, meta in zip(
results["documents"][0],
results["distances"][0],
results["metadatas"][0],
)
]
# Usage
hits = search("What was the revenue growth in Q3?")
for hit in hits:
print(f"[{hit['distance']:.3f}] {hit['text'][:80]}...")import glob
# 1. Ingest all .txt files in a directory
for filepath in glob.glob("./documents/*.txt"):
text = open(filepath).read()
chunks = chunk_document(text)
ingest(chunks, source=filepath)
print(f"Total indexed chunks: {collection.count()}")
# 2. Run a semantic search
results = search("How does the company plan to expand in Asia?", n_results=3)
# 3. Feed results to an LLM (this is RAG!)
context = "
".join(r["text"] for r in results)
print(f"
Context for LLM:
{context[:500]}...")source metadata field and filter one query to only search within a specific source document.You now have a complete semantic search pipeline. The capstone project in Lesson 8 puts this together with a PDF loader and LLM generation into a full document intelligence system.