"""
main.py — Streamlit UI for the Board Game RAG system.
Run (from project root):
streamlit run main.py
"""
import base64
import csv
import json
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts"))
import streamlit as st
from dotenv import load_dotenv
load_dotenv()
# ── Helpers ───────────────────────────────────────────────────────────────────
def img_to_base64(path: str) -> str:
with open(path, "rb") as f:
return base64.b64encode(f.read()).decode()
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title="Board Game Rules AI",
page_icon="🎲",
layout="wide",
initial_sidebar_state="collapsed",
)
# ── Load images ───────────────────────────────────────────────────────────────
CATAN_IMG = img_to_base64("games/Catan_game.png")
MONOPOLY_IMG = img_to_base64("games/Monopoly_game.png")
GAME_IMAGES = {"catan": CATAN_IMG, "monopoly": MONOPOLY_IMG}
# ── CSS ───────────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Game config ───────────────────────────────────────────────────────────────
GAMES = {
"catan": {
"name": "Catan",
"description": "Trade, build, and settle the island",
"rulebook": "data/raw/catan_rulebook.txt",
},
"monopoly": {
"name": "Monopoly",
"description": "Buy, sell, and scheme your way to victory",
"rulebook": "data/raw/monopoly_rulebook.txt",
},
}
EXAMPLES: dict[str, list[str]] = {
"catan": [
"What resources do you need to build a settlement?",
"What happens when you roll a 7?",
"Can players trade when it's not their turn?",
"Can I play a knight card before rolling?",
"How does Longest Road work?",
"What are development cards?",
],
"monopoly": [
"How much money does each player start with?",
"What happens when you pass GO?",
"How do you get out of jail?",
"What does Free Parking do?",
"Can you collect rent while in jail?",
"Can players lend money to each other?",
],
}
RETRIEVERS = {
"dense": {"name": "Dense Embeddings", "label": "Deep Learning"},
"tfidf": {"name": "TF-IDF", "label": "Classical ML"},
"random": {"name": "Random Baseline", "label": "Naive Baseline"},
}
# ── Cached loaders ────────────────────────────────────────────────────────────
@st.cache_resource(show_spinner="Loading model and index...")
def load_pipeline(game: str):
from model import retrieve, generate, _load
_load(game)
return retrieve, generate
@st.cache_data(show_spinner=False)
def load_eval_results(game: str, retriever: str):
path = (
f"data/outputs/{game}_results.csv"
if retriever == "dense"
else f"data/outputs/{game}_{retriever}_results.csv"
)
if not os.path.exists(path):
return None
with open(path) as f:
return list(csv.DictReader(f))
@st.cache_data(show_spinner=False)
def load_chunks(game: str):
path = f"models/{game}/chunks.json"
if not os.path.exists(path):
return []
with open(path) as f:
return json.load(f)
@st.cache_data(show_spinner=False)
def load_rulebook(game: str):
path = GAMES[game]["rulebook"]
if not os.path.exists(path):
return "Rulebook not found."
with open(path) as f:
return f.read()
def compute_eval_metrics(results):
"""Compute summary metrics from eval results."""
if not results:
return None
correctness = [r for r in results if r["category"] == "correctness"]
stress = [r for r in results if r["category"] == "stress"]
stress_c = [r for r in results if r.get("sub_category") == "C"]
def _avg(group, field):
vals = [float(r[field]) for r in group if r.get(field, "") != ""]
return sum(vals) / len(vals) if vals else 0
hall = (sum(1 for r in stress_c if r.get("hallucination_check") == "POTENTIAL_HALLUCINATION")
/ len(stress_c) * 100) if stress_c else 0
return {
"precision": _avg(results, "precision_at_k"),
"correctness": _avg(correctness, "correctness_score"),
"stress": _avg(stress, "correctness_score"),
"hallucination": hall,
}
# ── Session state ─────────────────────────────────────────────────────────────
if "page" not in st.session_state:
st.session_state.page = "home"
if "selected_game" not in st.session_state:
st.session_state.selected_game = None
if "selected_retriever" not in st.session_state:
st.session_state.selected_retriever = "dense"
if "question" not in st.session_state:
st.session_state.question = ""
if "last_answer" not in st.session_state:
st.session_state.last_answer = None
if "last_retrieved" not in st.session_state:
st.session_state.last_retrieved = None
if "rerank_enabled" not in st.session_state:
st.session_state.rerank_enabled = False
# ══════════════════════════════════════════════════════════════════════════════
# PAGE: HOME
# ══════════════════════════════════════════════════════════════════════════════
if st.session_state.page == "home":
# ── Hero ──
st.markdown("""
Know the Rules,
Win the Game
Ask any question about your favorite board games.
Powered by retrieval-augmented generation.
""", unsafe_allow_html=True)
# ── Games section ──
st.markdown("")
st.markdown("")
st.markdown('Choose Your Game
', unsafe_allow_html=True)
st.markdown(f"""
Catan
Trade, build, and settle the island
Monopoly
Buy, sell, and scheme your way to victory
""", unsafe_allow_html=True)
_, bcol1, _, bcol2, _ = st.columns([1.5, 1, 0.5, 1, 1.5])
with bcol1:
if st.button("Play Catan", use_container_width=True, type="secondary"):
st.session_state.page = "game"
st.session_state.selected_game = "catan"
st.session_state.question = ""
st.session_state.last_answer = None
st.session_state.last_retrieved = None
st.rerun()
with bcol2:
if st.button("Play Monopoly", use_container_width=True, type="secondary"):
st.session_state.page = "game"
st.session_state.selected_game = "monopoly"
st.session_state.question = ""
st.session_state.last_answer = None
st.session_state.last_retrieved = None
st.rerun()
# ── Rulebooks section ──
st.markdown('
', unsafe_allow_html=True)
st.markdown('Rulebooks
', unsafe_allow_html=True)
rb_col1, rb_col2 = st.columns(2, gap="medium")
game_details = {
"catan": {"sections": "Game Rules & Almanac", "designer": "Klaus Teuber", "year": "2020"},
"monopoly": {"sections": "21 Rule Sections", "designer": "Parker Brothers", "year": "2007"},
}
for col, (gk, gi) in zip([rb_col1, rb_col2], GAMES.items()):
with col:
chunks = load_chunks(gk)
chunk_count = len(chunks) if chunks else "—"
details = game_details[gk]
st.markdown(f"""
{gi['name']}
{details['designer']} · {details['year']}
{gi['description']}. Full official rulebook parsed into
{chunk_count} semantic chunks for intelligent retrieval.
{chunk_count} Chunks
{details['sections']}
""", unsafe_allow_html=True)
# ── Evaluation section ──
st.markdown('
', unsafe_allow_html=True)
st.markdown('Evaluation Overview
', unsafe_allow_html=True)
# Best result (dense) for each game
ev_cols = st.columns(2, gap="large")
for col, (gk, gi) in zip(ev_cols, GAMES.items()):
with col:
results = load_eval_results(gk, "dense")
metrics = compute_eval_metrics(results)
if metrics:
st.markdown(f"""
{gi['name']} — Dense Retrieval
{metrics['precision']:.0%}
Retrieval P@3
{metrics['correctness']:.0%}
Correctness
{metrics['hallucination']:.0f}%
Hallucination
""", unsafe_allow_html=True)
else:
st.markdown(f"""
{gi['name']}
—
No evaluation data
""", unsafe_allow_html=True)
# ── Footer ──
st.markdown("""
""", unsafe_allow_html=True)
# ══════════════════════════════════════════════════════════════════════════════
# PAGE: GAME
# ══════════════════════════════════════════════════════════════════════════════
elif st.session_state.page == "game":
game = st.session_state.selected_game
game_info = GAMES[game]
game_img = GAME_IMAGES[game]
# ── Compact hero with game info ──
st.markdown(f"""
![{game_info['name']}](data:image/png;base64,{game_img})
{game_info['name']} Rules Assistant
{game_info['description']}
""", unsafe_allow_html=True)
# Back button
st.markdown("")
if st.button("← Back to Games", type="secondary"):
st.session_state.page = "home"
st.session_state.selected_game = None
st.session_state.question = ""
st.session_state.last_answer = None
st.session_state.last_retrieved = None
st.rerun()
st.markdown("")
# ── Tabs ──
tab_ask, tab_rules, tab_eval = st.tabs(["Ask a Question", "Rulebook", "Evaluation"])
# ── TAB 1: Ask ──
with tab_ask:
st.markdown('Retrieval Strategy
',
unsafe_allow_html=True)
ret_cols = st.columns(3, gap="medium")
for j, (rk, ri) in enumerate(RETRIEVERS.items()):
with ret_cols[j]:
is_active = st.session_state.selected_retriever == rk
st.markdown(f"""
{ri['label']}
{ri['name']}
""", unsafe_allow_html=True)
if st.button(
"Selected" if is_active else "Use",
key=f"ret_{rk}",
use_container_width=True,
type="primary" if is_active else "secondary",
):
st.session_state.selected_retriever = rk
st.rerun()
retriever = st.session_state.selected_retriever
# Reranker toggle (only for non-random retrievers)
if retriever != "random":
rerank_on = st.toggle(
"Cross-Encoder Reranker",
value=st.session_state.rerank_enabled,
help="Retrieve 10 candidates, then re-rank to top 3 using a cross-encoder model for higher precision.",
)
if rerank_on != st.session_state.rerank_enabled:
st.session_state.rerank_enabled = rerank_on
else:
st.session_state.rerank_enabled = False
st.markdown('
', unsafe_allow_html=True)
# Examples
st.markdown('Try an example
', unsafe_allow_html=True)
ex_cols = st.columns(3, gap="small")
clicked = None
for i, ex in enumerate(EXAMPLES[game]):
with ex_cols[i % 3]:
if st.button(ex, key=f"ex_{i}", use_container_width=True):
clicked = ex
if clicked:
st.session_state.question = clicked
st.markdown("")
question = st.text_area(
"question",
value=st.session_state.question,
height=90,
placeholder=f"Ask anything about {game_info['name']} rules...",
label_visibility="collapsed",
)
_, btn_col, _ = st.columns([2, 1, 2])
with btn_col:
ask = st.button("Ask", type="primary", use_container_width=True)
if ask and question.strip():
st.session_state.question = question
retrieve_fn, generate_fn = load_pipeline(game)
with st.spinner("Searching rulebook..."):
retrieved = retrieve_fn(question, game=game, k=3, retriever=retriever, rerank=st.session_state.rerank_enabled)
with st.spinner("Generating answer..."):
answer = generate_fn(question, retrieved, game=game)
st.session_state.last_answer = answer
st.session_state.last_retrieved = retrieved
elif ask:
st.warning("Please enter a question.")
# Display answer
if st.session_state.last_answer:
answer = st.session_state.last_answer
retrieved = st.session_state.last_retrieved
st.markdown('
', unsafe_allow_html=True)
st.markdown('Answer
', unsafe_allow_html=True)
st.markdown(f'{answer}
', unsafe_allow_html=True)
ABSTAIN = [
"don't specify", "doesn't specify", "not specify",
"rules don't", "rules do not", "not addressed",
"not mentioned", "not explicit", "not state",
"silent", "i don't know", "cannot answer",
"unclear", "no rule", "not covered",
]
if not any(p in answer.lower() for p in ABSTAIN) and retrieved:
st.markdown(
'Retrieved Sections
',
unsafe_allow_html=True,
)
chunks_data = load_chunks(game)
for chunk in retrieved:
score = chunk["retrieval_score"]
rerank_score = chunk.get("rerank_score")
title = chunk["title"]
idx = chunk["chunk_idx"]
text = chunks_data[idx]["text"] if idx < len(chunks_data) else chunk.get("text", "")
if rerank_score is not None:
score_str = f"Rerank: {rerank_score:.3f} · Retrieval: {score:.3f}"
elif score > 0:
score_str = f"Similarity: {score:.3f}"
else:
score_str = "Random selection"
st.markdown(f"""
{title}
{score_str}
{text[:600]}{'...' if len(text) > 600 else ''}
""", unsafe_allow_html=True)
# ── TAB 2: Rulebook ──
with tab_rules:
st.markdown(
f''
f'{game_info["name"]} Official Rulebook
',
unsafe_allow_html=True,
)
rulebook = load_rulebook(game)
st.markdown(f'{rulebook}
', unsafe_allow_html=True)
chunks_data = load_chunks(game)
if chunks_data:
st.markdown(
f''
f'Indexed Chunks ({len(chunks_data)})
',
unsafe_allow_html=True,
)
for i, c in enumerate(chunks_data):
with st.expander(f"{i+1}. {c['title']}"):
st.markdown(
f"{c['text']}
",
unsafe_allow_html=True,
)
# ── TAB 3: Evaluation ──
with tab_eval:
st.markdown('Performance
',
unsafe_allow_html=True)
m_cols = st.columns(3, gap="medium")
for j, (rk, ri) in enumerate(RETRIEVERS.items()):
results = load_eval_results(game, rk)
metrics = compute_eval_metrics(results)
with m_cols[j]:
st.markdown(f"""
{ri['label']}
{ri['name']}
""", unsafe_allow_html=True)
if metrics:
for val, label in [
(f"{metrics['precision']:.0%}", "Retrieval P@3"),
(f"{metrics['correctness']:.0%}", "Correctness"),
(f"{metrics['stress']:.0%}", "Stress Test"),
(f"{metrics['hallucination']:.0f}%", "Hallucination"),
]:
st.markdown(f"""
""", unsafe_allow_html=True)
else:
st.markdown("""
""", unsafe_allow_html=True)
st.markdown('
', unsafe_allow_html=True)
st.markdown('Detailed Results
',
unsafe_allow_html=True)
detail_retriever = st.selectbox(
"Retriever",
options=["dense", "tfidf", "random"],
format_func=lambda r: RETRIEVERS[r]["name"],
key="eval_detail",
)
detail_results = load_eval_results(game, detail_retriever)
if detail_results:
for r in detail_results:
score = r.get("correctness_score", "")
if score == "1":
icon, color = "✓", "#48964b"
elif score == "0.5":
icon, color = "●", "#c49a2a"
elif score == "0":
icon, color = "✗", "#c44a3f"
else:
icon, color = "○", "#9a9590"
cat = r.get("sub_category", r["category"]).upper()
prec = r.get("precision_at_k", "")
st.markdown(f"""
{icon}
{cat}
{r['question'][:85]}{'...' if len(r['question']) > 85 else ''}
P@3: {prec}
""", unsafe_allow_html=True)
else:
st.info("No evaluation results available. Run the evaluation script first.")
# ── Footer ──
st.markdown("""
""", unsafe_allow_html=True)
