🔍 RAG Q&A

Concept

Retrieval-Augmented Generation (RAG) grounds LLM answers in a specific knowledge base rather than the model's training data. The pipeline: (1) chunk and embed the knowledge base into a Chroma vector store, (2) embed the question and find the most similar chunks, (3) inject those chunks into a prompt and ask GPT-4o mini to answer using only that context.

This prevents hallucination and lets the model answer questions about private documents.

Theory & Concepts

Grounding Generation in Fact

The RAG Q&A demo showcases how to prevent LLM hallucinations by injecting facts directly into the prompt context:

• Chunking and Embedding: You can't just pass massive documents to an LLM on every question. When you provide a knowledge base, the build_index() function uses LangChain to split the text into manageable 500-character chunks. These chunks are then converted into numerical vectors (embeddings) and stored in a Chroma database, which acts as a semantic search engine.
• Contextual Retrieval and Prompt Injection: When you ask a question, the system searches the Chroma database for the three most relevant chunks. The magic happens in the RAG_PROMPT: it combines those retrieved chunks with your question, and explicitly instructs GPT-4o mini to "Answer the question using ONLY the context below." By strictly limiting the model's knowledge universe to the provided chunks, the demo ensures factual, grounded answers rather than generic guesses.

Request flow

Code flow

Indexer

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma

from config import get_embedder


def build_index(
    docs: list[str],
    persist_directory: str = "./chroma_db",
    chunk_size: int = 500,
    chunk_overlap: int = 50,
) -> Chroma:
    """Chunk, embed, and persist documents into a Chroma vector store."""
    # Step 1: Instantiate RecursiveCharacterTextSplitter with 500 character chunks and 50 character overlap
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )

    # Step 2: Wrap each text segment in a LangChain Document object
    chunks = splitter.create_documents(docs)

    # Step 3: Load embedding model
    embedder = get_embedder()

    # Step 4: Embed chunks into dense vector representations and store in ChromaDB
    db = Chroma.from_documents(chunks, embedder, persist_directory=persist_directory)
    return db, len(chunks)

RAG query

from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate

from config import get_chat_model, get_embedder

# Instructs the model to answer ONLY from the provided context —
# prevents it from hallucinating or falling back on external pre-training data.
RAG_PROMPT = ChatPromptTemplate.from_template("""
Answer the question using ONLY the context below. If the context doesn't contain
the answer, say "I don't know." Be concise and quote facts directly.

Context:
{context}

Question: {question}
""")


def rag_answer(question: str, persist_directory: str = "./chroma_db", db=None) -> str:
    """Retrieve relevant chunks and generate a grounded answer."""
    # Step 1: Open existing vector store if not already provided in memory
    if db is None:
        embedder = get_embedder()
        db = Chroma(persist_directory=persist_directory, embedding_function=embedder)

    # Step 2: Retrieve top 3 nearest neighbor chunks based on cosine distance
    model = get_chat_model()
    chunks = db.similarity_search(question, k=3)

    # Step 3: Concatenate retrieved chunks into single context string
    context = "\n\n".join(c.page_content for c in chunks)

    # Step 4: Pipe context and question through RAG_PROMPT to the LLM
    chain = RAG_PROMPT | model
    return chain.invoke({"context": context, "question": question}).content

API route

@bp.route("/rag", methods=["POST"])
def rag_route():
    """Answer a question using a user-provided knowledge base.

    Request body (JSON): ``{ "knowledge_base": "<text>", "question": "<text>" }``
    Response (JSON):     ``{ "result": "<answer>" }``
    Error response:      ``{ "error": "<message>" }`` with HTTP 400 or 500
    """
    # Step 1: Parse request JSON payload
    data = request.get_json(force=True)
    knowledge = (data.get("knowledge_base") or "").strip()
    question = (data.get("question") or "").strip()

    # Step 2: Validate presence of both knowledge base and query text
    validation = validate_textarea(knowledge, "Knowledge base")
    if validation:
        return validation
    if not question:
        return jsonify({"error": "A question is required."}), 400

    # Step 3: Build ephemeral in-memory index from knowledge base lines and generate answer
    try:
        # Split pasted text into lines as distinct document passages
        docs = [line.strip() for line in knowledge.splitlines() if line.strip()]
        # persist_directory=None keeps the index in memory only — rebuilt fresh per request
        db, _ = build_index(docs, persist_directory=None)
        answer = rag_answer(question, db=db)
        return jsonify({"result": answer})
    except Exception as e:
        # Return HTTP 500 on unexpected errors
        return jsonify({"error": str(e)}), 500