Score Board
Total Score
{score.total:.1f}
Last Reward
{score.last_reward:+.1f}
Manhattan Distance
{distance}
Agent Position
{env.state}
Start Position
{env.start}
Status: {status}
Last Action: {score.last_action}
Rule Fired: {env.last_message}
Observation: [{obs[0]:.2f}, {obs[1]:.2f}, {obs[2]:.2f}, {obs[3]:.2f}]
"""
def reward_code_html() -> str:
return """
RL Training Board
Episode
{episode} / {total_episodes}
Exploration ε
{epsilon:.2f}
Training Successes
{successes}
Recent Success Rate
{recent_success_rate:.0%}
What is happening? {extra}
Q[s,a] ← Q[s,a] + α × (reward + γ × max(Q[next_state,*]) − Q[s,a])
This fast live demo uses tabular Q-learning against the same Gymnasium environment and reward function, so the learner improves by trial, reward, and correction.
"""
def new_game(grid_size: int):
env = GridWorldEnv(grid_size=int(grid_size), auto_generate=True, obstacle_density=0.20)
env.reset()
score = Score()
return env, score, grid_to_html(env), scoreboard_html(env, score), reward_code_html(), rl_log_html()
def move_agent(action: int, env: Optional[GridWorldEnv], score: Optional[Score], grid_size: int):
if env is None or score is None:
return new_game(grid_size)
if score.done:
env.last_message = "Episode already ended. Press Reset / Randomize Grid."
return env, score, grid_to_html(env), scoreboard_html(env, score), reward_code_html(), rl_log_html(extra="Episode ended. Reset to play again.")
_, reward, terminated, truncated, info = env.step(int(action))
score.last_reward = float(reward)
score.total += float(reward)
score.last_action = ACTION_LABELS[int(action)]
score.done = bool(terminated or truncated)
score.success = bool(info.get("success", False))
score.mode = "Human Play"
return env, score, grid_to_html(env), scoreboard_html(env, score), reward_code_html(), rl_log_html(extra="Human mode: use the arrow keys and watch the reward code fire.")
def clone_fixed_env(env: GridWorldEnv) -> GridWorldEnv:
fixed = GridWorldEnv(
grid=env.grid.copy(),
grid_size=env.grid_size,
auto_generate=False,
obstacle_density=env.obstacle_density,
)
fixed.reset()
return fixed
def valid_actions_for_grid(grid: np.ndarray, state: Position) -> List[int]:
size = grid.shape[0]
actions: List[int] = []
for action, (dr, dc) in enumerate(MOVES):
nr, nc = state[0] + dr, state[1] + dc
if 0 <= nr < size and 0 <= nc < size and grid[nr, nc] != "X":
actions.append(action)
return actions
def manhattan(pos1: Position, pos2: Position) -> int:
return abs(pos1[0] - pos2[0]) + abs(pos1[1] - pos2[1])
def choose_action(
q_table: np.ndarray,
grid: np.ndarray,
state: Position,
goal: Position,
epsilon: float,
rng: np.random.Generator,
episode: int,
total_episodes: int,
) -> int:
valid = valid_actions_for_grid(grid, state)
if not valid:
return 0
# Exploration: early episodes wander more; later exploration is biased slightly toward the goal.
if rng.random() < epsilon:
if episode > total_episodes * 0.25 and rng.random() < 0.45:
best_distance = min(manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), goal) for a in valid)
best = [a for a in valid if manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), goal) == best_distance]
return int(rng.choice(best))
return int(rng.choice(valid))
# Exploitation: choose the known best legal action; tie-break by Manhattan distance.
max_q = max(q_table[state[0], state[1], a] for a in valid)
best = [a for a in valid if q_table[state[0], state[1], a] == max_q]
if len(best) > 1:
best_distance = min(manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), goal) for a in best)
best = [a for a in best if manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), goal) == best_distance]
return int(rng.choice(best))
def initialize_q_table(grid: np.ndarray) -> np.ndarray:
size = grid.shape[0]
q_table = np.zeros((size, size, 4), dtype=np.float32)
# Mark illegal moves as very poor so the learned policy obeys the game board.
for r in range(size):
for c in range(size):
valid = set(valid_actions_for_grid(grid, (r, c)))
for action in range(4):
if action not in valid:
q_table[r, c, action] = -10.0
return q_table
def greedy_rollout(q_table: np.ndarray, grid: np.ndarray, max_steps: int) -> Tuple[bool, int]:
env = GridWorldEnv(grid=grid.copy(), grid_size=grid.shape[0], auto_generate=False, obstacle_density=0.20)
env.reset()
seen = set()
for step in range(max_steps):
state = env.state
valid = valid_actions_for_grid(grid, state)
if not valid:
return False, step
max_q = max(q_table[state[0], state[1], a] for a in valid)
best = [a for a in valid if q_table[state[0], state[1], a] == max_q]
best_distance = min(manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), env.goal) for a in best)
best = [a for a in best if manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), env.goal) == best_distance]
action = int(best[0])
loop_key = (state, action)
if loop_key in seen and step > grid.shape[0] * 2:
return False, step
seen.add(loop_key)
_, _, done, _, info = env.step(action)
if done:
return bool(info.get("success", False)), step + 1
return False, max_steps
def train_rl_solver(
env: Optional[GridWorldEnv],
score: Optional[Score],
grid_size: int,
training_episodes: int,
animation_delay: float,
):
"""
Train a Q-learning agent on the current fixed maze and animate the learning.
This is intentionally lightweight for Gradio: it uses the same Gymnasium env.step()
reward rules, but avoids long PPO training time so learners can watch the process live.
"""
if env is None:
env, score, grid_html, board_html, reward_html, log_html = new_game(grid_size)
yield env, score, grid_html, board_html, reward_html, log_html
assert env is not None
grid = env.grid.copy()
size = env.grid_size
max_steps = max(100, size * size)
total_episodes = int(training_episodes)
delay = max(0.0, float(animation_delay))
q_table = initialize_q_table(grid)
alpha = 0.65
gamma = 0.95
epsilon_start = 1.00
epsilon_end = 0.04
rng = np.random.default_rng()
success_window: deque[bool] = deque(maxlen=25)
total_successes = 0
best_steps: Optional[int] = None
# Animate early, middle, and late learning attempts.
preview_episodes = {
1,
2,
5,
max(10, total_episodes // 5),
max(20, total_episodes // 2),
total_episodes,
}
start_env = clone_fixed_env(env)
start_score = Score(mode="RL Training")
start_env.last_message = "RL training is starting on this exact maze."
yield (
start_env,
start_score,
grid_to_html(start_env),
scoreboard_html(start_env, start_score),
reward_code_html(),
rl_log_html("Starting", 0, total_episodes, epsilon_start, 0, 0.0, None, "The agent begins with high exploration and no useful policy."),
)
time.sleep(delay)
final_training_env = start_env
final_training_score = start_score
for episode in range(1, total_episodes + 1):
training_env = GridWorldEnv(grid=grid.copy(), grid_size=size, auto_generate=False, obstacle_density=env.obstacle_density)
training_env.reset()
episode_score = 0.0
episode_success = False
should_animate = episode in preview_episodes
# Linear decay keeps early exploration visible and late policy behavior cleaner.
progress = episode / max(1, total_episodes)
epsilon = max(epsilon_end, epsilon_start * (1.0 - progress))
for _step in range(max_steps):
old_state = training_env.state
action = choose_action(q_table, grid, old_state, training_env.goal, epsilon, rng, episode, total_episodes)
_, reward, terminated, truncated, info = training_env.step(action)
new_state = training_env.state
done = bool(terminated or truncated)
episode_score += float(reward)
next_valid = valid_actions_for_grid(grid, new_state)
next_best = max(q_table[new_state[0], new_state[1], a] for a in next_valid) if next_valid else 0.0
td_target = float(reward) if done else float(reward) + gamma * float(next_best)
td_error = td_target - float(q_table[old_state[0], old_state[1], action])
q_table[old_state[0], old_state[1], action] += alpha * td_error
if should_animate:
live_score = Score(
total=episode_score,
last_reward=float(reward),
last_action=ACTION_LABELS[action],
done=done,
success=bool(info.get("success", False)),
mode="RL Training",
)
extra = (
f"Animating training episode {episode}. The agent is using ε-greedy exploration, "
f"then updating its Q-table from the reward it receives."
)
yield (
training_env,
live_score,
grid_to_html(training_env),
scoreboard_html(training_env, live_score),
reward_code_html(),
rl_log_html("Learning", episode, total_episodes, epsilon, total_successes, (sum(success_window) / len(success_window)) if success_window else 0.0, best_steps, extra),
)
time.sleep(delay)
if done:
episode_success = bool(info.get("success", False))
break
success_window.append(episode_success)
if episode_success:
total_successes += 1
if best_steps is None or training_env.steps < best_steps:
best_steps = training_env.steps
final_training_env = training_env
final_training_score = Score(
total=episode_score,
last_reward=0.0,
last_action="Episode complete",
done=True,
success=episode_success,
mode="RL Training",
)
# Lightweight progress updates without animating every step.
if episode % max(10, total_episodes // 10) == 0 and not should_animate:
recent_rate = (sum(success_window) / len(success_window)) if success_window else 0.0
extra = f"Training checkpoint: recent success rate is {recent_rate:.0%}. The policy is becoming less random as ε decays."
yield (
final_training_env,
final_training_score,
grid_to_html(final_training_env),
scoreboard_html(final_training_env, final_training_score),
reward_code_html(),
rl_log_html("Training", episode, total_episodes, epsilon, total_successes, recent_rate, best_steps, extra),
)
# If the greedy policy does not solve yet, do a small hidden reinforcement top-up.
# This keeps the demo classroom-friendly while still using Q-learning updates.
solved, greedy_steps = greedy_rollout(q_table, grid, max_steps)
extra_rounds = 0
while not solved and extra_rounds < 3:
extra_rounds += 1
for _ in range(max(100, total_episodes // 2)):
training_env = GridWorldEnv(grid=grid.copy(), grid_size=size, auto_generate=False, obstacle_density=env.obstacle_density)
training_env.reset()
for _step in range(max_steps):
old_state = training_env.state
action = choose_action(q_table, grid, old_state, training_env.goal, 0.20, rng, total_episodes, total_episodes)
_, reward, terminated, truncated, _info = training_env.step(action)
new_state = training_env.state
done = bool(terminated or truncated)
next_valid = valid_actions_for_grid(grid, new_state)
next_best = max(q_table[new_state[0], new_state[1], a] for a in next_valid) if next_valid else 0.0
td_target = float(reward) if done else float(reward) + gamma * float(next_best)
q_table[old_state[0], old_state[1], action] += alpha * (td_target - q_table[old_state[0], old_state[1], action])
if done:
break
solved, greedy_steps = greedy_rollout(q_table, grid, max_steps)
yield (
final_training_env,
final_training_score,
grid_to_html(final_training_env),
scoreboard_html(final_training_env, final_training_score),
reward_code_html(),
rl_log_html("Extra RL Training", total_episodes, total_episodes, 0.20, total_successes, (sum(success_window) / len(success_window)) if success_window else 0.0, best_steps, "The first greedy rollout was not reliable yet, so the learner is getting extra reinforcement experience."),
)
# Final greedy policy rollout: no random exploration.
rollout_env = GridWorldEnv(grid=grid.copy(), grid_size=size, auto_generate=False, obstacle_density=env.obstacle_density)
rollout_env.reset()
rollout_score = Score(mode="Final Learned Policy")
yield (
rollout_env,
rollout_score,
grid_to_html(rollout_env),
scoreboard_html(rollout_env, rollout_score),
reward_code_html(),
rl_log_html("Final Rollout", total_episodes, total_episodes, 0.0, total_successes, (sum(success_window) / len(success_window)) if success_window else 0.0, best_steps, "Now exploration is off. The agent follows the learned policy greedily."),
)
time.sleep(delay)
visited_state_counts = {}
for _step in range(max_steps):
state = rollout_env.state
valid = valid_actions_for_grid(grid, state)
if not valid:
rollout_env.last_message = "No valid actions available."
break
max_q = max(q_table[state[0], state[1], a] for a in valid)
best = [a for a in valid if q_table[state[0], state[1], a] == max_q]
if len(best) > 1:
best_distance = min(manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), rollout_env.goal) for a in best)
best = [a for a in best if manhattan((state[0] + MOVES[a][0], state[1] + MOVES[a][1]), rollout_env.goal) == best_distance]
action = int(best[0])
_, reward, terminated, truncated, info = rollout_env.step(action)
rollout_score.total += float(reward)
rollout_score.last_reward = float(reward)
rollout_score.last_action = ACTION_LABELS[action]
rollout_score.done = bool(terminated or truncated)
rollout_score.success = bool(info.get("success", False))
rollout_score.mode = "Final Learned Policy"
recent_rate = (sum(success_window) / len(success_window)) if success_window else 0.0
extra = "The final learned policy is moving without random exploration."
yield (
rollout_env,
rollout_score,
grid_to_html(rollout_env),
scoreboard_html(rollout_env, rollout_score),
reward_code_html(),
rl_log_html("Final Rollout", total_episodes, total_episodes, 0.0, total_successes, recent_rate, best_steps, extra),
)
time.sleep(delay)
if rollout_score.done:
break
visited_state_counts[state] = visited_state_counts.get(state, 0) + 1
if visited_state_counts[state] > 4:
rollout_env.last_message = "The learned policy is looping. Press Train again or increase episodes."
rollout_score.done = True
rollout_score.success = False
yield (
rollout_env,
rollout_score,
grid_to_html(rollout_env),
scoreboard_html(rollout_env, rollout_score),
reward_code_html(),
rl_log_html("Needs More Training", total_episodes, total_episodes, 0.0, total_successes, recent_rate, best_steps, "The agent learned something, but this maze needs more episodes. Increase the episode slider and train again."),
)
break
def build_app() -> gr.Blocks:
with gr.Blocks(css=CSS, js=KEYBOARD_JS, title="Warehouse GridWorld Game + RL Solver") as demo:
gr.Markdown(
"""
# 🏗️ Warehouse GridWorld Game + Live RL Solver
Use the arrow keys to move the red agent from **S** to **G**. Obstacles are randomized at **20%** on every reset.
Press **Train RL Solver + Animate** to watch a reinforcement learner practice on the current maze and then perform a final learned run.
""",
elem_id="game-title",
)
env_state = gr.State(None)
score_state = gr.State(None)
with gr.Row():
with gr.Column(scale=2):
grid_display = gr.HTML(label="Warehouse Grid")
with gr.Row():
up_btn = gr.Button("↑ Up", elem_id="move-up")
with gr.Row():
left_btn = gr.Button("← Left", elem_id="move-left")
down_btn = gr.Button("↓ Down", elem_id="move-down")
right_btn = gr.Button("→ Right", elem_id="move-right")
with gr.Column(scale=1):
grid_size = gr.Slider(5, 15, value=10, step=1, label="Grid Size")
reset_btn = gr.Button("🔄 Reset / Randomize Grid", variant="primary")
train_episodes = gr.Slider(50, 2000, value=600, step=50, label="RL Training Episodes")
animation_delay = gr.Slider(0.00, 0.25, value=0.04, step=0.01, label="Animation Delay Seconds")
train_btn = gr.Button("🤖 Train RL Solver + Animate", variant="secondary")
score_display = gr.HTML(label="Score Board")
reward_display = gr.HTML(label="Reward Code")
rl_display = gr.HTML(label="RL Training Board")
outputs = [env_state, score_state, grid_display, score_display, reward_display, rl_display]
human_inputs = [env_state, score_state, grid_size]
train_inputs = [env_state, score_state, grid_size, train_episodes, animation_delay]
demo.load(fn=new_game, inputs=[grid_size], outputs=outputs)
reset_btn.click(fn=new_game, inputs=[grid_size], outputs=outputs)
up_btn.click(fn=lambda env, score, size: move_agent(0, env, score, size), inputs=human_inputs, outputs=outputs)
right_btn.click(fn=lambda env, score, size: move_agent(1, env, score, size), inputs=human_inputs, outputs=outputs)
down_btn.click(fn=lambda env, score, size: move_agent(2, env, score, size), inputs=human_inputs, outputs=outputs)
left_btn.click(fn=lambda env, score, size: move_agent(3, env, score, size), inputs=human_inputs, outputs=outputs)
train_btn.click(fn=train_rl_solver, inputs=train_inputs, outputs=outputs)
return demo
if __name__ == "__main__":
app = build_app()
app.queue()
app.launch()