Lesson 08d: Document Service Deep Dive#

"You can't search what you never indexed correctly."

Where This Lives#

backend/app/services/document_service.py (108 lines). This is the ingestion and retrieval engine behind every /agents/{id}/documents/* endpoint. It owns two external systems: the OpenAI Embeddings API and a local ChromaDB store.

Module-Level Setup#

import chromadb
from dotenv import load_dotenv
from openai import OpenAI
 
load_dotenv()
 
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
 
openai_client = OpenAI(api_key=OPENAI_API_KEY)
 
chroma_client = chromadb.PersistentClient(path="./chroma_db")
 
documents_collection = chroma_client.get_or_create_collection(
    name="agent_documents"
)

Four things happen the moment this module is imported (i.e., the moment main.py imports routers.documents, which imports this service):

1. .env is loaded — this is the third independent load_dotenv() call in the codebase (see Lesson 27: Configuration for why that's a smell, not a feature).
2. An OpenAI client is created once and reused for every embedding call.
3. chromadb.PersistentClient(path="./chroma_db") opens (or creates) a folder on disk relative to wherever the process was launched from — this is the backend/app/chroma_db/ directory mentioned in CLAUDE.md.
4. One single Chroma collection, agent_documents, is shared by every agent in the system. There is no per-agent collection. Isolation between agents is enforced entirely by a metadata filter at query time (where={"agent_id": ...}), not by physical separation. Keep this in mind — it resurfaces in Lesson 25: Security as a soft multi-tenancy boundary.

split_text_into_chunks() — The Chunking Algorithm#

def split_text_into_chunks(text: str, chunk_size: int = 800, overlap: int = 100) -> list[str]:
    cleaned_text = text.strip()
    if not cleaned_text:
        return []
 
    chunks = []
    start = 0
    while start < len(cleaned_text):
        end = start + chunk_size
        chunk = cleaned_text[start:end].strip()
        if chunk:
            chunks.append(chunk)
        start = end - overlap
        if start <= 0:
            start = end
    return chunks

This is a sliding window over raw characters — not words, not sentences, not tokens. Trace it by hand for a 2,000-character document:

start = end - overlap is what creates the 100-character overlap between consecutive chunks — this exists so a sentence that happens to fall exactly on a chunk boundary isn't cut in half and lost from both chunks' semantic meaning.

Why if start <= 0: start = end? It's a guard against an infinite loop. If overlap >= chunk_size, end - overlap could stay at or below the previous start, and the while loop would never advance. With the defaults (chunk_size=800, overlap=100) this branch is never actually reached — it only matters if someone calls this function with a misconfigured overlap.

What this algorithm does not do, worth knowing as a limitation, not a bug:

• No sentence- or paragraph-boundary awareness — a chunk can start or end mid-word.
• No token counting — "800 characters" is not "800 tokens" (English averages ~4 characters/token, so a chunk is roughly ~200 tokens).
• No overlap on the last chunk boundary — the tail end of the document only appears once.

create_embedding()#

def create_embedding(text: str) -> list[float]:
    if not OPENAI_API_KEY:
        raise ValueError("OPENAI_API_KEY is missing. Please add it to your .env file.")
    response = openai_client.embeddings.create(model=EMBEDDING_MODEL, input=text)
    return response.data[0].embedding

One HTTP call to OpenAI per invocation, returning a list[float]. With the default model, text-embedding-3-small, that list has 1536 dimensions (see Lesson 15: Embeddings — this corrects an earlier draft of this course that said 384).

add_chunk_to_vector_store()#

def add_chunk_to_vector_store(vector_id: str, chunk_text: str, metadata: dict) -> None:
    embedding = create_embedding(chunk_text)
    documents_collection.add(
        ids=[vector_id],
        documents=[chunk_text],
        embeddings=[embedding],
        metadatas=[metadata],
    )

Called once per chunk from routers/documents.py's upload endpoint. This means uploading a document with 10 chunks makes 10 separate, sequential OpenAI embedding calls inside a single HTTP request — no batching, no concurrency. See Lesson 24: Performance for the latency consequence and Lesson 18: Background Tasks for how you'd fix it.

The vector_id passed in is built by the caller as f"agent_{agent.id}_document_{document.id}_chunk_{index}" — a human-readable, collision-resistant primary key for Chroma that also gets persisted on the DocumentChunk.vector_id column in SQL. This is the thread that ties a relational row to its vector twin.

search_agent_documents()#

def search_agent_documents(agent_id: int, query: str, top_k: int = 3) -> list[dict]:
    query_embedding = create_embedding(query)
    results = documents_collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k,
        where={"agent_id": agent_id},
    )
    ...

Two things happen: the query text itself is embedded (same model, same function as ingestion — this symmetry matters, you cannot mix embedding models between ingestion and query and expect meaningful distances), and Chroma does an approximate nearest-neighbor search scoped to agent_id via the where filter. This is the multi-tenancy boundary mentioned above — it's a query-time filter, not a hard partition.

The result is reshaped from Chroma's parallel-list response (documents, metadatas, distances, each a list-of-lists) into a flat list[dict] via zip(). One naming subtlety worth flagging: the output field is called score, but the value assigned to it is distance straight from Chroma — for the default space, lower is better (closer = more similar), which is the opposite of what "score" usually implies (higher is better). A consumer of this API who assumes "score" means "confidence" would sort results backwards.

How the Router Uses This Service#

routers/documents.py never talks to Chroma or OpenAI directly — it only imports split_text_into_chunks, add_chunk_to_vector_store, and search_agent_documents from this file. That's the service-layer boundary in action: HTTP concerns (status codes, request parsing) stay in the router; embedding/vector-store concerns stay here. See Lesson 09b: Documents Router for the full endpoint walkthrough.

Common Mistakes#

• Assuming score is a similarity score. It's a distance. Sort ascending, not descending, if you want "most relevant first."
• Forgetting embeddings must match between ingestion and query. If OPENAI_EMBEDDING_MODEL changes after documents were already ingested, old vectors and new queries live in different vector spaces — searches will silently return garbage, not an error.
• Assuming per-agent isolation is bulletproof. It's a where clause, not a separate database. A bug that drops or mismatches the agent_id filter leaks every agent's documents to every other agent.

Exercises#

1. Trace split_text_into_chunks("A" * 2500) by hand (or in a REPL) and list the (start, end) pairs the loop produces.
2. search_agent_documents calls create_embedding once per search. What happens to cost and latency if the same query string is searched 100 times in a row? Propose a one-line fix using a cache (ties into Lesson 26: Caching).
3. Rename the score field to distance in schemas.DocumentSearchResult and update every place that reads it. What breaks, and why does this small rename matter for API consumers?

Key Takeaways#

🎯 Chunking here is character-based, not token- or sentence-aware — a deliberate simplicity trade-off. 🎯 Every chunk costs one OpenAI embedding call — ingestion latency scales linearly with document size. 🎯 Chroma isolation between agents is a where filter, not physical separation. 🎯 The score field is actually a distance — smaller means more similar.

Next Steps#

👉 Lesson 08e: RAG Graph Service (LangGraph) →

Module 3 · Services Deep Dive