Artificial Intelligence & MLOps

Building a Production-Grade Q-Learning Pathfinder Pipeline in Python

By EduArn Engineering Team • 6-Step Architecture Guide

Ever wondered how autonomous systems learn complex navigation tasks without human intervention? Most machine learning tutorials jump straight into heavy neural networks, skipping the foundational mechanics of tabular Reinforcement Learning (RL). Today, we break down a complete 6-file RL pipeline using Q-learning, binary model packaging, and a live web UI deployment.

In this walkthrough, we model a practical scenario: training an intelligent student agent to navigate a room layout, avoiding distraction penalty zones (social media) while prioritizing high-reward destinations (the study desk).

Architecture Overview & Project Pattern

To ensure code readability and modularity, we structure our workflow across six steps following industry standards:

• 00 Environment Setup: Virtual environment isolation and package dependencies.
• 01 Environment Configuration: Declarative state space definition saved via JSON.
• 02 Model Training: Bellman Equation implementation and Q-Table convergence.
• 03 Trajectory Evaluation: Deterministic path verification.
• 04 Model Serialization: Joblib binary pipeline packaging.
• 05 Web UI Deployment: Serving interactive real-time inference using Gradio.

Step 00: Environment Configuration

We start by setting up an isolated environment to prevent dependency conflicts across machine learning tools.

# 00 Environment Setup.txt
py -m venv reinforcement_env
.\reinforcement_env\Scripts\activate

python -m pip install --upgrade pip
pip install numpy pandas matplotlib gradio joblib

Step 01: Defining the Grid State Space

We configure the 4x4 room grid programmatically and export the parameters to a structured JSON file.

# 01 Create RL Grid Environment.py
import json

env_config = {
    "grid_size": 4,
    "start_state": 0,       # Bed (Start)
    "goal_state": 15,      # Study Desk (+10 Reward)
    "obstacles": [5, 12],  # Distractions (-10 Penalty)
    "actions": {
        "0": "LEFT",
        "1": "DOWN",
        "2": "RIGHT",
        "3": "UP"
    }
}

with open("grid_env_config.json", "w") as f:
    json.dump(env_config, f, indent=4)

print("Environment configuration saved to 'grid_env_config.json'")

Step 02: Core Q-Learning & Bellman Equation Updates

During training, the agent uses an Epsilon-Greedy strategy to balance exploration and exploitation. The Q-Table updates using the standard Bellman equation:

# 02 Train Q Learning Model.py
import json
import numpy as np
import pandas as pd

with open("grid_env_config.json", "r") as f:
    config = json.load(f)

GRID_SIZE = config["grid_size"]
NUM_STATES = GRID_SIZE * GRID_SIZE
NUM_ACTIONS = len(config["actions"])
START_STATE = config["start_state"]
GOAL_STATE = config["goal_state"]
OBSTACLES = config["obstacles"]

NUM_EPISODES = 1000
LEARNING_RATE = 0.8
GAMMA = 0.95
EPSILON = 1.0
EPSILON_DECAY = 0.995
MIN_EPSILON = 0.01

q_table = np.zeros((NUM_STATES, NUM_ACTIONS))

def step(state, action):
    row, col = divmod(state, GRID_SIZE)
    moves = [ (0, -1), (1, 0), (0, 1), (-1, 0) ]
    dr, dc = moves[action]
    
    new_row = max(0, min(GRID_SIZE - 1, row + dr))
    new_col = max(0, min(GRID_SIZE - 1, col + dc))
    next_state = new_row * GRID_SIZE + new_col

    if next_state == GOAL_STATE:
        return next_state, 10.0, True
    elif next_state in OBSTACLES:
        return next_state, -10.0, True
    else:
        return next_state, -1.0, False

for episode in range(NUM_EPISODES):
    state = START_STATE
    done = False

    while not done:
        if np.random.uniform(0, 1) < EPSILON:
            action = np.random.choice(NUM_ACTIONS)
        else:
            action = np.argmax(q_table[state])

        next_state, reward, done = step(state, action)

        best_future_q = np.max(q_table[next_state]) if not done else 0.0
        td_target = reward + GAMMA * best_future_q
        q_table[state, action] += LEARNING_RATE * (td_target - q_table[state, action])

        state = next_state

    EPSILON = max(MIN_EPSILON, EPSILON * EPSILON_DECAY)

df_qtable = pd.DataFrame(q_table, columns=["LEFT", "DOWN", "RIGHT", "UP"])
df_qtable.to_csv("trained_q_table.csv", index_label="State")
print("Q-Learning Model Trained and saved to 'trained_q_table.csv'")

Step 03: Trajectory Evaluation

We evaluate the policy deterministically using np.argmax(q_table[state]) to ensure optimal path selection.

# 03 Evaluate & Predict Optimal Trajectory.py
import json
import pandas as pd
import numpy as np

with open("grid_env_config.json", "r") as f:
    config = json.load(f)

df_qtable = pd.read_csv("trained_q_table.csv", index_col="State")
q_table = df_qtable.to_numpy()

GRID_SIZE = config["grid_size"]
GOAL_STATE = config["goal_state"]
OBSTACLES = config["obstacles"]
ACTIONS = ["LEFT", "DOWN", "RIGHT", "UP"]

def get_next_state(state, action_idx):
    row, col = divmod(state, GRID_SIZE)
    moves = [(0, -1), (1, 0), (0, 1), (-1, 0)]
    dr, dc = moves[action_idx]
    new_row = max(0, min(GRID_SIZE - 1, row + dr))
    new_col = max(0, min(GRID_SIZE - 1, col + dc))
    return new_row * GRID_SIZE + new_col

current_state = config["start_state"]
path = [current_state]
steps = 0

while current_state != GOAL_STATE and current_state not in OBSTACLES and steps < 15:
    action_idx = np.argmax(q_table[current_state])
    action_name = ACTIONS[action_idx]
    next_s = get_next_state(current_state, action_idx)
    
    print(f"State {current_state:2d} -> Action [{action_name:5s}] -> State {next_s:2d}")
    current_state = next_s
    path.append(current_state)
    steps += 1

print("Final Trajectory Path Taken:", path)

Step 04: Serialization with Joblib

The Q-table and environment properties are serialized into a single binary pipeline file (q_learning_pipeline_model.pkl).

# 04 Export Q Table Model to Joblib.py
import json
import joblib
import pandas as pd

with open("grid_env_config.json", "r") as f:
    config = json.load(f)

df_qtable = pd.read_csv("trained_q_table.csv", index_col="State")

rl_pipeline = {
    "config": config,
    "q_table": df_qtable.to_numpy()
}

joblib.dump(rl_pipeline, "q_learning_pipeline_model.pkl")
print("Saved binary pipeline artifact to 'q_learning_pipeline_model.pkl'")

Step 05: Interactive Gradio Application

Finally, we load the saved model and build a lightweight web interface using Gradio to visualize path planning dynamically.

# 05 Launch Interactive RL Demo.py
import gradio as gr
import joblib
import numpy as np

pipeline = joblib.load("q_learning_pipeline_model.pkl")
config = pipeline["config"]
q_table = pipeline["q_table"]

GRID_SIZE = config["grid_size"]
GOAL_STATE = config["goal_state"]
OBSTACLES = config["obstacles"]
ACTION_NAMES = ["LEFT", "DOWN", "RIGHT", "UP"]

def simulate_student_path(start_tile):
    current_state = int(start_tile)
    path = [current_state]
    actions_taken = []
    steps = 0
    
    while current_state != GOAL_STATE and current_state not in OBSTACLES and steps < 15:
        action_idx = np.argmax(q_table[current_state])
        action_name = ACTION_NAMES[action_idx]
        
        row, col = divmod(current_state, GRID_SIZE)
        moves = [(0, -1), (1, 0), (0, 1), (-1, 0)]
        dr, dc = moves[action_idx]
        new_row = max(0, min(GRID_SIZE - 1, row + dr))
        new_col = max(0, min(GRID_SIZE - 1, col + dc))
        
        current_state = new_row * GRID_SIZE + new_col
        path.append(current_state)
        actions_taken.append(action_name)
        steps += 1

    if current_state == GOAL_STATE:
        status = "Goal Reached! Optimal routine completed."
    elif current_state in OBSTACLES:
        status = "Trapped! Student hit an obstacle."
    else:
        status = "Max steps reached."

    return status, " -> ".join([f"State {s}" for s in path]), ", ".join(actions_taken)

demo = gr.Interface(
    fn=simulate_student_path,
    inputs=[gr.Slider(minimum=0, maximum=14, step=1, value=0, label="Select Starting State")],
    outputs=[gr.Textbox(label="Result"), gr.Textbox(label="Path"), gr.Textbox(label="Actions")],
    title="EduArn RL Pathfinder",
    theme="soft"
)

if __name__ == "__main__":
    demo.launch()

Master Machine Learning & Engineering

Ready to transition from basic script writing to building enterprise-grade AI applications and end-to-end MLOps pipelines?

Join EduArn Courses Today