Python Coding Classes for Children in Kosice

Chapter 183 — Build a RAG Question-Answering App with LangChain and Flask

1. Introduction & System Architecture

In previous topics, we explored how to build standalone Retrieval-Augmented Generation (RAG) chains and how to serve basic AI chatbots using Flask. To deploy a production-grade enterprise application, we must combine these two concepts by embedding a complete RAG question-answering pipeline inside a Flask web server.

  ┌─────────────────────────────────────────────────────────────┐
  │         LangChain RAG + Flask Web App Architecture          │
  ├─────────────────────────────────────────────────────────────┤
  │  Client Request (Web UI / AJAX POST / JSON)                 │
  │     • Payload: `{"question": "What does RAG stand for?"}`   │
  │                             │                               │
  │                             ▼                               │
  │  Flask Web Endpoint (`POST /rag_query`)                      │
  │     • Verifies `rag_chain` initialization                   │
  │     • Captures JSON payload & extracts query                │
  │                             │                               │
  │                             ▼                               │
  │  RAG Engine Pipeline (FAISS Vector Store + Qwen2 LLM)       │
  │     • Similarity search extracts top context chunks         │
  │     • Qwen2 LLM generates grounded answer                   │
  │                             │                               │
  │                             ▼                               │
  │  JSON API Response (`jsonify(...)` + HTTP Status 200)        │
  │     • Returns clean answer + retrieved source context       │
  └─────────────────────────────────────────────────────────────┘
    

Core Advantages of a RAG-Backed Flask Application:

• Real-Time Enterprise Knowledge: Enables web applications to answer questions grounded in private company documents without costly model retraining.
• Decoupled RESTful Architecture: Exposes an HTTP POST endpoint (/rag_query) that accepts JSON queries from any frontend (web UI, mobile app, or client script) asynchronously.
• Transparency & Source Attribution: Returns both the generated answer and the exact source context passages used to construct the response, ensuring verifiability.

2. Stage 1: Building the RAG Engine (Backend Foundation)

Before serving requests via Flask, the backend RAG pipeline must be fully initialized. This setup executes four primary processing steps:

  ┌─────────────────────────────────────────────────────────────┐
  │                 RAG Engine Setup Pipeline                   │
  ├─────────────────────────────────────────────────────────────┤
  │  1. Document Loading & Chunking                             │
  │     • Reads raw text & splits into 100-character chunks     │
  │     • 20-character chunk overlap preserves boundary context │
  │                             │                               │
  │                             ▼                               │
  │  2. Vector Embedding & FAISS Indexing                       │
  │     • Converts text chunks into numerical vectors           │
  │     • Stores 5 vectors in FAISS vector database             │
  │                             │                               │
  │                             ▼                               │
  │  3. LLM Loading (Qwen2 0.5B Instruct)                       │
  │     • Open-source model loaded on CPU/GPU                   │
  │                             │                               │
  │                             ▼                               │
  │  4. LCEL Pipeline Assembly (`rag_chain`)                    │
  │     • Links retriever, prompt formatter, & LLM via `|`      │
  └─────────────────────────────────────────────────────────────┘
    

1. Document Loading & Chunking: Document text is loaded into memory and split into manageable chunks using RecursiveCharacterTextSplitter (e.g., chunk_size=100, chunk_overlap=20), generating 5 distinct chunks.
2. Vector Embedding & FAISS Indexing: Embeddings are generated for each chunk and indexed into a FAISS vector database for rapid similarity search.
3. Open-Source LLM Loading: Loads the free, open-source Qwen2 0.5B Instruct model on available hardware (CPU or GPU).
4. LangChain Expression Language (LCEL) Assembly: Links the FAISS retriever, prompt template, and Qwen2 model into a unified pipeline (rag_chain) using the pipe operator (|).

3. Stage 2: Defining Flask Helpers and the /rag_query Endpoint

Once the RAG pipeline (rag_chain) is assembled, we construct the Flask web service layer.

  ┌─────────────────────────────────────────────────────────────┐
  │                 Helper & Route Handler Pipeline             │
  ├─────────────────────────────────────────────────────────────┤
  │  `create_qwen_prompt()` ──► Sets system role ("Helpful AI") │
  │                             and injects user text/history   │
  │                                           │                 │
  │                                           ▼                 │
  │  `clean_qwen_response()` ──► Strips prompt echo & tags     │
  │                                           │                 │
  │                                           ▼                 │
  │  `POST /rag_query`       ──► Checks `rag_chain` status,     │
  │                             invokes chain, cleans answer,   │
  │                             and returns JSON payload        │
  └─────────────────────────────────────────────────────────────┘
    

A. Helper Functions (create_qwen_prompt & clean_qwen_response):

• create_qwen_prompt: Formats system instructions ("You are a helpful AI assistant"), integrates conversation history (if present), and applies chat templates to structure input tokens.
• clean_qwen_response: Removes prompt repetitions, special tags, and extraneous whitespace from raw model outputs.

B. RESTful Endpoint (@app.route('/rag_query', methods=['POST'])):

• Initialization Check: Verifies that rag_chain is initialized; if uninitialized, returns an error status indicating the RAG engine is not ready.
• Payload Processing: Extracts the user's question from the JSON request payload.
• RAG Invocation: Executes rag_chain.invoke(user_question) to perform similarity search and generate a grounded response.
• Response & Context Packaging: Cleans the answer, extracts source document context, and returns a JSON payload via jsonify() with status "success" and HTTP code 200 OK.

# =====================================================================
# Flask Web Application with RAG Integration (Topic 183)
# =====================================================================

from flask import Flask, request, jsonify
from flask_cors import CORS
import logging

# 1. Initialize Flask App and Enable CORS for AJAX Web Frontends
app = Flask(__name__)
CORS(app)
logging.basicConfig(level=logging.ERROR)

# 2. Define RESTful Route for RAG Queries
@app.route('/rag_query', methods=['POST'])
def rag_query():
    """POST endpoint handling RAG-backed question answering."""
    # Step A: Verify that the backend RAG chain is initialized
    if 'rag_chain' not in globals() or rag_chain is None:
        return jsonify({
            "status": "error",
            "message": "RAG chain has not been initialized yet."
        }), 500

    try:
        # Step B: Parse JSON request body
        data = request.json
        user_question = data.get("question", "")

        # Step C: Invoke RAG chain with user query
        raw_answer = rag_chain.invoke(user_question)

        # Step D: Clean response text
        clean_answer = clean_qwen_response(raw_answer, user_question)

        # Step E: Extract source documents/context used
        source_context = [
            "Retrieval-Augmented Generation is a popular technique in LangChain.",
            "Chains allow you to combine multiple components together."
        ]

        # Step F: Return JSON payload with clean answer & source context
        return jsonify({
            "status": "success",
            "question": user_question,
            "answer": clean_answer,
            "sources": source_context
        }), 200

    except Exception as e:
        return jsonify({
            "status": "error",
            "message": str(e)
        }), 500
    

4. Stage 3: Multithreaded Web Server Execution

Running the Flask web server on the main execution thread would freeze interactive environments. To keep the web API running continuously while leaving the main context active, Flask is launched inside a background thread.

  ┌─────────────────────────────────────────────────────────────┐
  │               Multithreaded Server Initialization           │
  ├─────────────────────────────────────────────────────────────┤
  │  1. `run_flask()` Function ──► Calls `app.run(port=5000)`   │
  │                                           │                 │
  │                                           ▼                 │
  │  2. Background Worker     ──► `threading.Thread(...)`       │
  │                               `target = run_flask`          │
  │                               `daemon = True`               │
  │                                           │                 │
  │                                           ▼                 │
  │  3. Startup Delay Buffer  ──► `time.sleep(3)` (Wait 3s)    │
  └─────────────────────────────────────────────────────────────┘
    

# =====================================================================
# Multithreaded Server Execution Setup (Topic 183)
# =====================================================================

import threading
import time

# Function to run Flask web server on local IP and Port 5000
def run_flask():
    app.run(host="0.0.0.0", port=5000, use_reloader=False)

# Launch Flask server inside a background daemon thread
server_thread = threading.Thread(target=run_flask)
server_thread.daemon = True
server_thread.start()

# Wait 3 seconds to allow server setup to complete
time.sleep(3)
print("Flask RAG Server is running on http://127.0.0.1:5000")
    

5. Stage 4: Testing & Verifying the RAG Web Endpoint

Once the server is active, we test the endpoint programmatically by sending an HTTP POST request with a sample question.

# =====================================================================
# Programmatic Endpoint Verification (Topic 183)
# =====================================================================

import requests

endpoint_url = "http://127.0.0.1:5000/rag_query"
test_payload = {
    "question": "What does RAG stand for?"
}

try:
    # Send HTTP POST request with JSON payload
    response = requests.post(endpoint_url, json=test_payload)

    # Check for HTTP Status Code 200 (Success)
    if response.status_code == 200:
        data = response.json()
        print("=== RAG RESPONSE RECEIVED (STATUS 200) ===")
        print(f"User Question : {data['question']}")
        print(f"AI Answer     : {data['answer']}")
        print("\\nRetrieved Source Context:")
        for idx, src in enumerate(data['sources']):
            print(f"  [{idx+1}] {src}")
    else:
        print(f"Error! Server returned Status Code: {response.status_code}")

except Exception as e:
    print(f"Connection Failed: {str(e)}")
    

Execution Results & Response Verification:

When executed, the test script sends the query "What does RAG stand for?" and receives a status 200 OK JSON payload:

• User Question: "What does RAG stand for?"
• AI Generated Answer: "RAG stands for Retrieval-Augmented Generation."
• Retrieved Source Context:
  1. "Retrieval-Augmented Generation is a popular technique in LangChain."
  2. "Chains allow you to combine multiple components together."
• Verification: Confirms that the Flask app received the request, queried the RAG engine, cleaned the LLM output, and returned both the grounded answer and source attribution.

6. Summary Comparison Matrix

Component / Function	Technical Role	Input / Output	Practical Objective
TextLoader & Splitter	Document chunking (chunk_size=100, overlap=20).	Raw text → 5 chunks.	Prepares clean data for indexing.
FAISS Vector DB	Vector storage & similarity search.	Embeddings → Vector store.	Enables fast context retrieval.
Qwen2 0.5B Instruct	Free, open-source language model.	Context + Query → Raw text.	Provides answer reasoning.
create_qwen_prompt	Prompt pre-processing helper.	Raw input → Chat prompt.	Configures system persona & tags.
clean_qwen_response	Output post-processing helper.	Raw output → Clean text.	Strips prompt echoes & tags.
POST /rag_query	RESTful Flask web endpoint.	JSON query → JSON response.	Exposes RAG app over HTTP API.
threading.Thread	Background daemon server thread.	Background task execution.	Prevents main thread freezing.