📚 Paragraph RAG#

Paragraph RAG (Retrieval-Augmented Generation) combines the previous two demos (Chunking + Embeddings) with an LLM to answer questions strictly from a user-supplied source text (by default, the TechToday Return Policy). Instead of hoping the LLM knows your company's private rules, RAG retrieves the exact relevant snippets and injects them into the prompt before generation.

On each request, answer_rag() runs the complete 5-stage pipeline end-to-end: splits the paragraph into 200-character overlapping chunks, embeds them locally with all-MiniLM-L6-v2 into an ephemeral Chroma vector store, retrieves the top k ≤ 3 chunks closest to the user's question, deletes the temporary collection, and calls gpt-4o-mini at temperature=0 with a strict grounding prompt.

1. The Five-Stage RAG Architecture

1. Chunk: RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40) breaks the source text into focused passages.
2. Embed & Index: Chroma.from_texts(chunks, embedder) converts each chunk into a 384-d vector using local HuggingFaceEmbeddings and indexes them in memory.
3. Retrieve: db.similarity_search(question, k=min(3, len(chunks))) embeds the user's question with the same model and finds the top-k nearest chunk vectors.
4. Augment: The retrieved chunks are joined and slotted into the {context} placeholder of ChatPromptTemplate alongside {question}.
5. Generate: ChatOpenAI(temperature=0) synthesizes a concise, fact-based answer and the API returns both answer and sources.

2. Preventing Hallucinations via Negative Prompt Constraints

Without explicit instructions, an LLM asked a question not covered by the retrieved chunks will cheerfully fall back on its pretraining memory and guess. The prompt enforces a strict closed-world contract: "Answer the question using ONLY the context below. If the context doesn't contain the answer, say 'I couldn't find that in the provided paragraph.'"

3. Returning Provenance (sources) for Verifiability

By returning sources: relevant_chunks alongside the generated answer, the UI lets the user verify which exact sentences from the policy backed the model's claim — making retrieval failures (wrong chunks retrieved) easy to distinguish from generation failures (right chunks retrieved, wrong answer synthesized).

4. Ephemeral Collection Cleanup (db.delete_collection())

In an interactive demo where every visitor can paste a custom paragraph, keeping chunks in a shared default Chroma collection would cause later queries to retrieve leftover chunks from earlier requests. Calling db.delete_collection() immediately after retrieval guarantees complete request isolation.

src/python/03_rag.py implements a complete Retrieval-Augmented Generation (RAG) pipeline that splits a reference paragraph into overlapping chunks, indexes them in an in-memory ChromaDB vector store using local HuggingFace embeddings, retrieves the top matching passages for a question, and synthesizes a strictly grounded answer with OpenAI.

1. Core AI Concepts in the RAG Pipeline

• Retrieve → Augment → Generate: Rather than relying on an LLM's static parametric memory, RAG retrieves relevant passages from an external knowledge source, injects them into the prompt as context, and instructs the LLM to answer strictly from those passages.
• Hybrid Local Embedding + Cloud Synthesis: Chunking and vector indexing run locally on CPU using HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2"), while only the final grounded synthesis step invokes ChatOpenAI with temperature=0.
• Hallucination Refusal Guardrail: The prompt template explicitly instructs the model to answer using only the retrieved context and to reply "I couldn't find that in the provided paragraph." whenever the context lacks the answer.

2. Key Entities in src/python/03_rag.py

• PARAGRAPH constant: A default reference passage describing the TechToday Return Policy (30-day window, condition requirements, 5–7 business day refund timeline, and non-refundable shipping/clearance rules).
• RagRequest(BaseModel) & RagResponse(BaseModel): Pydantic schemas defining the inputs (paragraph: str = PARAGRAPH, question: str) and the output payload (paragraph: str, question: str, answer: str, and sources: list[str]).
• embedder = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2"): Module-level embedding model that maps text chunks and queries into 384-dimensional vectors.
• prompt = ChatPromptTemplate.from_template(...): Prompt template that binds {context} and {question} with strict grounding instructions.
• answer_rag(paragraph: str, question: str) -> dict:
  • Step ① — Chunk the paragraph: Splits paragraph into overlapping pieces via RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40).split_text(paragraph).
  • Step ② — Embed & store in Chroma: Calls db = Chroma.from_texts(chunks, embedder) to vectorize every chunk and store it in an in-memory Chroma collection.
  • Step ③ — Retrieve top-$k$ chunks: Computes k = min(3, len(chunks)) and runs db.similarity_search(question, k=k) to extract the most relevant chunk strings.
  • Step ④ — Join context & clean up collection: Joins relevant_chunks with "\n\n" into context and calls db.delete_collection() so the next query starts with a fresh in-memory vector store.
  • Step ⑤ — Generate grounded answer: Composes a LangChain Expression Language (LCEL) chain (prompt | model).invoke({"context": context, "question": question}) with ChatOpenAI(temperature=0) and returns paragraph, question, answer, and sources.

3. End-to-End RAG Execution Flow

1. answer_rag() splits the input paragraph into 200-character chunks with a 40-character overlap.
2. Chroma.from_texts() embeds all chunks with all-MiniLM-L6-v2 and indexes them in memory, then db.similarity_search() retrieves the top-$k$ chunks closest to question before deleting the temporary collection.
3. The retrieved chunks are formatted into prompt and passed to ChatOpenAI(temperature=0), producing a deterministic, fact-grounded answer alongside the exact source chunks used.

Paragraph RAG (Retrieval-Augmented Generation) Service.

"""Paragraph RAG (Retrieval-Augmented Generation) Service.

This service implements a RAG workflow for answering questions over a paragraph:
1. Splits the paragraph into manageable text chunks using RecursiveCharacterTextSplitter.
2. Indexes chunks and retrieves the top matching passages for the user question.
3. Formulates a grounded prompt with the retrieved context.
4. Generates a concise, fact-based answer using ChatOpenAI.
"""

import os

from dotenv import find_dotenv, load_dotenv
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pydantic import BaseModel

# Load API keys and model names from .env
load_dotenv(find_dotenv())

PARAGRAPH = (
    "TechToday Return Policy: Customers may return any eligible product within 30 days "
    "of the purchase date. All returned merchandise must be unworn, undamaged, and in its "
    "original packaging with all tags attached. Refunds are processed back to the original "
    "payment method within 5 to 7 business days after inspection. Standard shipping fees "
    "are non-refundable unless the return is due to our fulfillment error. International "
    "orders and clearance items marked as final sale cannot be returned or exchanged."
)


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class RagRequest(BaseModel):
    """Input payload for paragraph-based RAG query."""

    paragraph: str = PARAGRAPH
    question: str = "How long do I have to return something?"


class RagResponse(BaseModel):
    """Output payload containing grounded answer and retrieved sources."""

    paragraph: str
    question: str
    answer: str
    sources: list[str]


# ---------------------------------------------------------------------------
# RAG Pipeline: Chunk -> Embed & Store -> Retrieve -> Generate
# ---------------------------------------------------------------------------
# Load the local embedding model once (no API key needed)
embedder = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# RAG prompt: instructs the model to answer strictly from retrieved context
prompt = ChatPromptTemplate.from_template("""
Answer the question using ONLY the context below. If the context doesn't contain
the answer, say "I couldn't find that in the provided paragraph."
Be concise and quote facts directly.

Context:
{context}

Question: {question}
""")


def answer_rag(paragraph: str, question: str) -> dict:
    """Execute the complete paragraph RAG pipeline: chunk -> retrieve -> synthesize.

    Args:
        paragraph: The source paragraph to search.
        question: User query about the paragraph.

    Returns:
        Dictionary matching RagResponse schema.
    """
    # ① Chunk the paragraph into overlapping pieces
    splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40)
    chunks = splitter.split_text(paragraph)

    # ② Embed every chunk and store the vectors in an in-memory Chroma vector store
    db = Chroma.from_texts(chunks, embedder)

    # ③ Retrieve the top-k chunks most similar to the question
    k = min(3, len(chunks))
    relevant_chunks = [doc.page_content for doc in db.similarity_search(question, k=k)]

    # ④ Join the retrieved chunks into one context string
    context = "\n\n".join(relevant_chunks)
    # Drop the in-memory collection so the next request starts with an empty store
    db.delete_collection()

    # ⑤ Generate an answer grounded only in the retrieved context
    #   (temperature=0 for deterministic, factual responses)
    model = ChatOpenAI(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        base_url=os.getenv("OPENAI_BASE_URL") or None,
        temperature=0,
    )
    answer = (prompt | model).invoke({"context": context, "question": question}).content

    return {
        "paragraph": paragraph,
        "question": question,
        "answer": answer.strip(),
        "sources": relevant_chunks,
    }

Flask Blueprint server and route handler (rag_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)

Use these numbered takeaways to revise the end-to-end Retrieval-Augmented Generation (RAG) architecture:

1. The Three Pillars of RAG (Retrieve, Augment, Generate): Instead of relying on an LLM's static parametric memory, RAG (1) retrieves relevant document chunks from a vector store, (2) augments the prompt with those chunks as context, and (3) generates an answer grounded in that context.
2. Hybrid Local + Cloud Stack: This pipeline runs chunking and embedding locally on CPU with all-MiniLM-L6-v2 (free and fast) and only calls the cloud LLM (gpt-4o-mini) once for the final synthesis step.
3. Per-Request Ephemeral Collections (uuid.uuid4()): Naming each Chroma collection f"rag_{uuid.uuid4().hex[:8]}" guarantees that concurrent requests indexing different user-supplied paragraphs never overwrite or pollute each other's vectors.
4. Guaranteed Cleanup via try / finally: Calling db.delete_collection() inside a finally block ensures the temporary vector index is freed even if similarity search raises an exception.
5. Dynamic $k$ Selection (k=min(3, len(chunks))): Clamping k to len(chunks) prevents ChromaDB from requesting more neighbors than exist when the input passage is short.
6. Strict Grounding & Refusal Prompting: Explicitly instructing the model to use only the supplied context and output a deterministic refusal phrase when facts are missing is the primary defense against LLM hallucination.
7. Zero Temperature for Factual Synthesis: Using temperature=0 on ChatOpenAI keeps the generated answer faithful to the retrieved chunks rather than creatively extrapolating.
8. Source Provenance in the API Response: Returning retrieved_chunks alongside answer lets users verify which exact sentences from the passage supported the model's claim.