🧩 Document Chunking#

Document Chunking is the foundational preprocessing stage of every Retrieval-Augmented Generation (RAG) system. Before a long document, manual, or book can be indexed in a vector database, it must be broken into smaller, self-contained passages called chunks so that embedding models can represent each idea accurately and only the relevant passages are sent to the LLM.

This demo uses LangChain's RecursiveCharacterTextSplitter with configurable chunk_size (default 250 characters) and chunk_overlap (default 40 characters). Rather than slicing text blindly every N characters — which cuts words and sentences in half — the recursive splitter walks a hierarchy of natural boundaries (paragraphs → lines → words → characters) and carries a sliding window of shared text across adjacent chunks.

1. Why Chunking Is Necessary for RAG

Embedding an entire 50-page document into a single vector averages hundreds of distinct topics into one blurry point in vector space, making retrieval imprecise. Conversely, feeding whole documents into an LLM prompt wastes tokens, increases latency, and triggers the "lost in the middle" phenomenon where models overlook facts buried inside huge contexts.

2. How RecursiveCharacterTextSplitter Preserves Meaning

The splitter attempts to divide text using an ordered list of separators: ["\n\n", "\n", " ", ""]:

1. First, it splits on double newlines ("\n\n") to keep whole paragraphs intact.
2. If a paragraph still exceeds chunk_size, it recursively splits that paragraph on single newlines ("\n").
3. If a line is still too long, it splits on spaces (" ") so words stay whole, falling back to individual characters only as a last resort.

3. The Chunk Size Trade-Off (Precision vs. Context)

• Too Small (e.g., <100 chars): Chunks become isolated sentence fragments missing their subject or antecedents (e.g., "It expires after 30 days" without saying what "It" is).
• Too Large (e.g., >2,000 chars): Multiple unrelated topics get merged into one chunk, diluting the embedding vector and adding noise to the LLM prompt.

4. Context Continuity via chunk_overlap

When an important thought straddles the boundary between Chunk i and Chunk i + 1, a hard cut would split the premise from its conclusion. Setting chunk_overlap=40 repeats the trailing ~40 characters of Chunk i at the start of Chunk i + 1, acting as semantic glue across split boundaries.

src/python/01_chunking.py implements the document chunking stage of a Retrieval-Augmented Generation (RAG) pipeline, splitting long documents into smaller, overlapping passages suitable for dense vector embedding and similarity search.

1. Core AI Concepts in Document Chunking

• Granular Retrieval Units: Embedding an entire multi-paragraph document into a single vector blurs distinct facts together. Splitting text into focused chunks ensures each vector represents a coherent passage.
• Recursive Separator Hierarchy: Rather than slicing blindly at fixed character offsets (which cuts words and sentences in half), RecursiveCharacterTextSplitter walks a hierarchy of natural boundaries—paragraphs ("\n\n"), lines ("\n"), and spaces (" ")—keeping semantically related sentences together whenever they fit within chunk_size.
• Sliding Window Context Glue (chunk_overlap): Repeating a configurable tail of characters at the start of the next chunk preserves context across boundaries so ideas spanning two adjacent chunks are not severed.

2. Key Entities in src/python/01_chunking.py

• TEXT_RAG constant: A default sample passage explaining RAG, recursive splitting, and chunk overlap.
• ChunkingRequest(BaseModel) & ChunkingResponse(BaseModel): Pydantic schemas defining the chunking parameters (text: str = TEXT_RAG, chunk_size: int = 250, chunk_overlap: int = 40) and the structured output (total_chunks: int, chunk_size: int, chunk_overlap: int, and chunks: list[str]).
• chunk_text(text: str = TEXT_RAG, chunk_size: int = 250, chunk_overlap: int = 40) -> list[str]:
  • Step ① — Configure the recursive splitter: Instantiates RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap) so chunks target chunk_size characters while preserving natural paragraph, line, and word boundaries.
  • Step ② — Split the document text: Calls splitter.split_text(text) and returns the ordered list of overlapping text chunks.

3. End-to-End Chunking Flow

1. chunk_text() receives the raw document string along with the target chunk_size and chunk_overlap.
2. RecursiveCharacterTextSplitter partitions the text along paragraph and sentence boundaries so each piece stays under chunk_size characters.
3. Adjacent chunks carry over up to chunk_overlap shared characters from the preceding chunk, producing a list of context-preserving passages ready for downstream embedding and vector indexing.

Text Chunking Service.

"""Text Chunking Service.

This service splits long documents into smaller, overlapping chunks suitable
for vector embeddings and retrieval (RAG).
"""

from langchain_text_splitters import RecursiveCharacterTextSplitter
from pydantic import BaseModel

TEXT_RAG = (
    "Retrieval-Augmented Generation (RAG) combines search retrieval with generative AI models. "
    "Before documents can be searched efficiently, they must be broken down into smaller pieces. "
    "This process is called chunking. "
    "RecursiveCharacterTextSplitter splits documents on paragraph breaks, line breaks, and spaces "
    "to keep semantically related sentences together. "
    "Chunk overlap ensures that context across boundaries is not lost between adjacent chunks. "
    "These chunks are then converted into numerical vectors using embedding models, allowing "
    "similarity search to locate the most relevant sections when answering a user question."
)


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class ChunkingRequest(BaseModel):
    """Input payload for text chunking."""

    text: str = TEXT_RAG
    chunk_size: int = 250
    chunk_overlap: int = 40


class ChunkingResponse(BaseModel):
    """Output payload containing generated chunks and configuration."""

    total_chunks: int
    chunk_size: int
    chunk_overlap: int
    chunks: list[str]


# ---------------------------------------------------------------------------
# Core Service Logic
# ---------------------------------------------------------------------------
def chunk_text(text: str = TEXT_RAG, chunk_size: int = 250, chunk_overlap: int = 40) -> list[str]:
    """Split text into overlapping chunks using RecursiveCharacterTextSplitter.

    Args:
        text: Input text document.
        chunk_size: Target characters per chunk.
        chunk_overlap: Shared characters between adjacent chunks.

    Returns:
        List of text chunks.
    """
    # ① Configure RecursiveCharacterTextSplitter with target chunk size and overlap.
    #   It tries to split on paragraphs (\n\n), then lines (\n), then spaces, so related
    #   sentences stay together while keeping each chunk under chunk_size.
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,        # aim for target chars per chunk
        chunk_overlap=chunk_overlap,  # adjacent chunks share chars (context glue)
    )
    # ② Split the document text into overlapping chunks preserving natural boundaries
    return splitter.split_text(text)

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

"""Flask server for 03-ai-rag-chroma."""

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_chunk = _load_service("01_chunking.py", "chunking")
mod_embed = _load_service("02_embedding.py", "embedding")
mod_rag = _load_service("03_rag.py", "rag")


@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/chunk", methods=["POST"])
def chunking_service():
    # ① Parse text, chunk_size, and chunk_overlap from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    text = (data.get("text") or "").strip()
    chunk_size = int(data.get("chunk_size") or 250)
    chunk_overlap = int(data.get("chunk_overlap") or 40)
    if not text:
        return jsonify({"error": "Text cannot be empty"}), 400
    try:
        # ② Split the text into overlapping chunks via 01_chunking.py and return JSON
        chunks = mod_chunk.chunk_text(text, chunk_size, chunk_overlap)
        return jsonify({
            "result": f"Split document into {len(chunks)} chunks (size {chunk_size}, overlap {chunk_overlap}).",
            "total_chunks": len(chunks),
            "chunk_size": chunk_size,
            "chunk_overlap": chunk_overlap,
            "chunks": chunks,
        })
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/embedding", methods=["POST"])
def embedding_service():
    # ① Parse both input strings and validate that neither is empty
    data = request.get_json(force=True, silent=True) or {}
    text1 = (data.get("text1") or "").strip()
    text2 = (data.get("text2") or "").strip()
    if not text1 or not text2:
        return jsonify({"error": "Both text1 and text2 are required"}), 400
    try:
        # ② Compute vector embeddings and cosine similarity via 02_embedding.py
        res = mod_embed.compute_similarity(text1, text2)
        res["result"] = f"Cosine Similarity: {res['similarity']:.4f} (Dimension: {res['dimensions']})"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/rag", methods=["POST"])
def rag_service():
    # ① Parse the source paragraph and user question from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    paragraph = (data.get("paragraph") or "").strip()
    question = (data.get("question") or data.get("message") or "").strip()
    if not paragraph or not question:
        return jsonify({"error": "Both paragraph and question are required"}), 400
    try:
        # ② Execute the Chroma RAG pipeline in 03_rag.py and return answer + sources
        res = mod_rag.answer_rag(paragraph, question)
        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)

Review these numbered points to solidify the core concepts behind document chunking in RAG pipelines:

1. Why Chunking is First in RAG: Embedding an entire multi-page document into a single vector dilutes specific facts and exceeds context limits; splitting documents into focused chunks lets vector search retrieve only the exact passages relevant to a query.
2. Recursive Separator Hierarchy: RecursiveCharacterTextSplitter attempts splits in descending semantic order ([" ", " ", " ", ""]), keeping paragraphs and sentences intact whenever they fit inside chunk_size instead of slicing mid-word.
3. Role of chunk_size: Controls the maximum size (measured by length_function=len in characters here) of each chunk—smaller chunks yield more surgical retrieval, while larger chunks preserve broader surrounding explanation.
4. Role of chunk_overlap: Repeats a sliding window of trailing characters at the beginning of the next chunk so ideas or sentences that cross a chunk boundary are not severed and lost during retrieval.
5. Constraint (chunk_overlap < chunk_size): Overlap must strictly be smaller than chunk size; otherwise the splitter window cannot advance forward through the document.
6. Zero External API Dependency: Text chunking is a purely local, deterministic string-processing step that runs in milliseconds on CPU without requiring an LLM or network call.
7. Impact on Downstream Embedding Quality: Cleanly chunked passages produce sharper, single-topic dense embeddings in the next stage of the RAG pipeline, directly improving cosine similarity recall in ChromaDB.