🗄️ ChromaDB Notes Q&A#

ScalerGPT (ChromaDB Document RAG) upgrades the single-paragraph RAG demo into a persistent, multi-document knowledge base over a folder of Markdown engineering notes (docs/docker-basics.md, docs/compose-and-volumes.md, and docs/rag-notes.md).

Instead of rebuilding and deleting a vector store on every request, _get_collection() maintains a named "notes" collection in ChromaDB using OpenAI's text-embedding-3-small embedding function. On the first query, it reads all .md and .txt files from disk, splits them into paragraphs, attaches {"source": filename} metadata to each chunk, and performs an idempotent collection.upsert(). Subsequent questions immediately query the indexed collection, retrieve the top 3 chunks with their source filenames, and synthesize a grounded answer with gpt-4o-mini.

1. Hybrid Chroma Client Architecture (Container vs. Ephemeral)

In multi-container Docker deployments (where CHROMA_HOST is set), the service connects over HTTP via chromadb.HttpClient(host=chroma_host, port=chroma_port) to a dedicated ChromaDB container. When running standalone without a separate vector DB container, it transparently falls back to chromadb.EphemeralClient() in-process — allowing the exact same RAG code to run in both environments.

2. Idempotent Ingestion with Deterministic Chunk IDs

Calling collection.add() repeatedly with random UUIDs duplicates documents and pollutes top-k search results with identical chunks. ScalerGPT assigns each paragraph a deterministic composite key — f"{path.name}::{i}" (e.g., docker-basics.md::0) — and calls collection.upsert(), which inserts new chunks or updates existing ones in place, and skips disk ingestion altogether once collection.count() > 0.

3. Source Attribution via Vector Metadata

Every vector stored in ChromaDB carries a parallel metadata dictionary: metadatas.append({"source": path.name}). When collection.query() returns the top-3 nearest chunks, it returns both hits["documents"] and hits["metadatas"], allowing the UI to display which exact Markdown file each retrieved passage came from.

4. Filtering Noise During Paragraph Splitting

Markdown files often contain short headings or blank lines that make poor standalone vectors. The comprehension [c.strip() for c in text.split("\n\n") if len(c.strip()) > 40] discards tiny fragments under 40 characters so every indexed vector carries substantive explanatory content.

src/python/02_chromadb.py implements a Retrieval-Augmented Generation (RAG) workflow (ScalerGPT) over markdown and text study notes stored in docs/, indexing paragraph chunks into a ChromaDB collection with OpenAI embeddings and generating grounded answers with source metadata.

1. Core AI Concepts in Vector Database RAG

• Collection-Bound Embedding Function: Attaching OpenAIEmbeddingFunction(model_name="text-embedding-3-small") directly to the Chroma collection ensures that both document ingestion (collection.upsert()) and query-time lookup (collection.query()) automatically embed text into the exact same vector space.
• Chunking with Source Metadata: Splitting study notes on paragraph breaks ("\n\n") and tagging each chunk with {"source": path.name} and a composite ID (f"{path.name}::{i}") preserves document provenance alongside every stored vector.
• Client-Server & Ephemeral Chroma Modes: ChromaDB supports connecting to a standalone vector database server via chromadb.HttpClient when CHROMA_HOST is configured, or running an in-memory chromadb.EphemeralClient using the identical collection API.

2. Key Entities in src/python/02_chromadb.py

• DOCS_DIR & _chroma_client: Point to the docs/ directory of study notes and cache the active ChromaDB client instance.
• ChromaDbRequest, ChromaDbSource, & ChromaDbResponse: Pydantic schemas defining the user's question, each retrieved source chunk (source: str, content: str), and the complete response (question: str, answer: str, docs_indexed: int, and sources: list[ChromaDbSource]).
• _get_collection():
  • Step ① — Initialize ChromaDB client: Connects via chromadb.HttpClient(host=chroma_host, port=chroma_port) when CHROMA_HOST is set, or initializes chromadb.EphemeralClient() otherwise.
  • Step ② — Bind OpenAI embedding function: Configures embedding_functions.OpenAIEmbeddingFunction(api_key=api_key, model_name="text-embedding-3-small") and calls _chroma_client.get_or_create_collection(name="notes", embedding_function=openai_ef).
  • Step ③ — Ingest markdown/text notes on first query: When collection.count() == 0, reads all *.txt and *.md files in DOCS_DIR, splits each file on "\n\n" into paragraph chunks longer than 40 characters, and calls collection.upsert(ids=ids, documents=documents, metadatas=metadatas).
• ask_chromadb_notes(question: str) -> dict:
  • Step ④ — RETRIEVE: Calls collection.query(query_texts=[question], n_results=3) to fetch the top-3 semantically matching chunks (retrieved_docs) and their metadata dicts (retrieved_metas), joining the chunks with "\n\n---\n\n" into context.
  • Step ⑤ — AUGMENT: Builds the ScalerGPT system prompt instructing the model to answer using only the retrieved CONTEXT and to say it doesn't know if the answer is absent.
  • Step ⑥ — GENERATE: Calls llm.chat.completions.create(model=model, messages=[...]) with gpt-4o-mini and pairs the grounded answer with the total indexed chunk count (collection.count()) and structured sources list.

3. End-to-End ChromaDB RAG Flow

1. ask_chromadb_notes(question) obtains the "notes" collection from _get_collection(), which lazily chunks, embeds, and upserts all .md and .txt files from docs/ if the collection is empty.
2. collection.query() embeds question with text-embedding-3-small and returns the top-3 closest note chunks along with their {"source": filename} metadata.
3. The retrieved chunks are injected into the ScalerGPT system prompt and sent to OpenAI Chat Completions, returning the grounded answer and source attributions.

ScalerGPT — Document Ingestion & RAG over Markdown Notes with ChromaDB.

"""ScalerGPT — Document Ingestion & RAG over Markdown Notes with ChromaDB.

This service demonstrates a complete Retrieval-Augmented Generation (RAG) workflow over markdown notes:
1. Connects to ChromaDB (`chromadb.HttpClient` when `CHROMA_HOST` is configured, or an in-memory
   `chromadb.EphemeralClient` inside the Flask server).
2. Reads `.md` and `.txt` notes from `docs/`, splits them into paragraph chunks, embeds them with OpenAI's
   `text-embedding-3-small`, and upserts them into the `"notes"` Chroma collection.
3. Executes the 3-stage RAG pipeline (Retrieve -> Augment -> Generate): queries top-3 relevant chunks from
   ChromaDB, injects them into a grounded system prompt, and generates a factual answer with `gpt-4o-mini`.
"""

import os
from pathlib import Path

import chromadb
from chromadb.utils import embedding_functions
from dotenv import find_dotenv, load_dotenv
from openai import OpenAI
from pydantic import BaseModel

load_dotenv(find_dotenv())

DOCS_DIR = Path(__file__).with_name("docs")
_chroma_client = None


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class ChromaDbRequest(BaseModel):
    """Input payload for ScalerGPT ChromaDB RAG query."""

    question: str = "Why should I copy requirements.txt before the rest of my code?"


class ChromaDbSource(BaseModel):
    """A retrieved document chunk with its source filename."""

    source: str
    content: str


class ChromaDbResponse(BaseModel):
    """Output payload containing grounded answer, indexed chunk count, and retrieved sources."""

    question: str
    answer: str
    docs_indexed: int
    sources: list[ChromaDbSource]


# ---------------------------------------------------------------------------
# ChromaDB Connection & Markdown Note Ingestion
# ---------------------------------------------------------------------------
def _get_collection():
    """Return the 'notes' Chroma collection, ingesting markdown files from docs/ if empty."""
    global _chroma_client
    api_key = os.getenv("OPENAI_API_KEY", "").strip()
    if not api_key:
        raise RuntimeError("OPENAI_API_KEY is not configured in environment.")

    # ① Connect to Chroma server if CHROMA_HOST is set; otherwise use in-memory EphemeralClient
    if _chroma_client is None:
        chroma_host = os.getenv("CHROMA_HOST", "").strip()
        chroma_port = int(os.getenv("CHROMA_PORT", "8000"))
        if chroma_host:
            _chroma_client = chromadb.HttpClient(host=chroma_host, port=chroma_port)
        else:
            _chroma_client = chromadb.EphemeralClient()

    # ② Configure OpenAI text-embedding-3-small for the collection
    openai_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=api_key,
        model_name="text-embedding-3-small",
    )
    collection = _chroma_client.get_or_create_collection(
        name="notes",
        embedding_function=openai_ef,
    )

    # ③ Ingest markdown/text notes from docs/ on first query
    if collection.count() == 0:
        files = sorted(list(DOCS_DIR.glob("*.txt")) + list(DOCS_DIR.glob("*.md")))
        ids, documents, metadatas = [], [], []
        for path in files:
            text = path.read_text(encoding="utf-8").strip()
            chunks = [c.strip() for c in text.split("\n\n") if len(c.strip()) > 40]
            for i, chunk in enumerate(chunks):
                ids.append(f"{path.name}::{i}")
                documents.append(chunk)
                metadatas.append({"source": path.name})
        if documents:
            collection.upsert(ids=ids, documents=documents, metadatas=metadatas)

    return collection


# ---------------------------------------------------------------------------
# Core Service Logic (Retrieve -> Augment -> Generate)
# ---------------------------------------------------------------------------
def ask_chromadb_notes(question: str) -> dict:
    """Query indexed markdown notes in ChromaDB and synthesize a grounded answer."""
    collection = _get_collection()

    # ④ RETRIEVE: Find the top-3 most relevant chunks from ChromaDB
    hits = collection.query(query_texts=[question], n_results=3)
    retrieved_docs = (hits.get("documents") or [[]])[0]
    retrieved_metas = (hits.get("metadatas") or [[]])[0]
    context = "\n\n---\n\n".join(retrieved_docs)

    # ⑤ AUGMENT: Build the strictly grounded ScalerGPT system prompt
    system_prompt = (
        "You are ScalerGPT, a helpful teaching assistant. "
        "Answer the user's question using ONLY the context below. "
        "If the context does not contain the answer, say you don't know.\n\n"
        f"CONTEXT:\n{context}"
    )

    # ⑥ GENERATE: Synthesize the final answer with OpenAI
    llm = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL") or None,
        timeout=20.0,
    )
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
    resp = llm.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question},
        ],
    )

    sources = [
        {
            "source": (meta.get("source", "notes") if isinstance(meta, dict) else "notes"),
            "content": doc,
        }
        for doc, meta in zip(retrieved_docs, retrieved_metas)
    ]

    return {
        "question": question,
        "answer": (resp.choices[0].message.content or "").strip(),
        "docs_indexed": collection.count(),
        "sources": sources,
    }

Flask Blueprint server and route handler (chromadb_route) in src/python/app.py.

"""Flask server for 04-ai-model-training-chromadb."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_ml = _load_service("01_model_training.py", "model_training")
mod_chroma = _load_service("02_chromadb.py", "chromadb_rag")
mod_redis = _load_service("03_redis.py", "redis_agent")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/model-training", methods=["POST"])
def model_training_service():
    # ① Parse delivery order features from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    try:
        distance_km = float(data.get("distance_km", 4.5))
        prep_time_min = float(data.get("prep_time_min", 15.0))
        rider_available = int(data.get("rider_available", 1))
        is_raining = int(data.get("is_raining", 1))
        if distance_km <= 0 or prep_time_min <= 0:
            return jsonify({"error": "Distance and prep time must be positive."}), 400
        # ② Run RandomForestRegressor inference in 01_model_training.py and return ETA
        res = mod_ml.predict_delivery_eta(distance_km, prep_time_min, rider_available, is_raining)
        res["result"] = f"Predicted Arrival: {res['eta_minutes']} min 🍔"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/chromadb", methods=["POST"])
def chromadb_service():
    # ① Parse student question from JSON payload and reject empty queries
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Query ChromaDB notes collection via 02_chromadb.py and return answer + sources
        res = mod_chroma.ask_chromadb_notes(question)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/redis", methods=["POST"])
def redis_agent_service():
    # ① Parse session_id, message, and optional reset flag from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    session_id = (data.get("session_id") or "demo").strip()
    message = (data.get("message") or "").strip()
    reset_session = bool(data.get("reset_session", False))
    if not session_id or (not reset_session and not message):
        return jsonify({"error": "Session ID and message are required"}), 400
    try:
        # ② Run DeskBuddy agent turn (or session reset) backed by Redis in 03_redis.py
        res = mod_redis.run_redis_agent(session_id, message, reset_session)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Use these numbered takeaways to revise persistent ChromaDB collections, graceful service fallback, and cited RAG:

1. Client-Server vs. Embedded Vector Store: ChromaDB supports both a standalone server mode (chromadb.HttpClient talking to a Docker container over HTTP) and an in-process mode (chromadb.EphemeralClient), using the exact same Python collection API.
2. Graceful Infrastructure Fallback: Probing _chroma_client.heartbeat() inside a try/except block allows the app to use the dedicated Chroma container when running under Docker Compose and automatically fall back to in-memory storage during standalone local development.
3. Collection-Bound Embedding Functions: Passing OpenAIEmbeddingFunction(model_name="text-embedding-3-small") into get_or_create_collection() delegates vectorization to the collection itself—calling upsert(documents=...) or query(query_texts=...) automatically embeds texts with the same model.
4. Idempotent Upsert & Lazy Seeding: Guarding ingestion with if collection.count() == 0: and using collection.upsert(documents=docs, ids=ids) ensures study notes are embedded only once when the collection is empty and never duplicated if re-run with the same filename IDs.
5. Source Attribution via Document IDs: Using each file's name (e.g., docker_notes.md) as its Chroma id and prefixing each context block with [Source: {sid}] enables the LLM to cite exact filenames in its answer.
6. Embedding Dimension Consistency: Every query against a collection must use the exact same embedding model (text-embedding-3-small, 1536 dimensions) that was used to index the documents; binding the embedding function to the collection enforces this invariant automatically.
7. Observability in the UI Response: Returning both the backend string (http://... vs. in-memory (ephemeral)) and the sources preview list lets the user see where vectors are stored and which files answered their question.