"""
Streamlit web UI — Teachings Q&A
Two-column layout: chat left, sources right.
Saffron/espresso/cream palette.
Run:
venv/bin/streamlit run app.py
"""
import json
import os
import re
import sys
from pathlib import Path
sys.path.insert(0, ".")
import markdown as _md
import streamlit as st
import streamlit.components.v1 as _components
from answer import answer as get_answer
HISTORY_FILE = Path("data/chat_history.json")
def _password_gate() -> None:
"""Block access unless APP_PASSWORD env var is unset or correct password entered."""
required = os.environ.get("APP_PASSWORD", "").strip()
if not required:
return
if st.session_state.get("_auth"):
return
st.markdown("""
✿ Teachings Q&A
Enter the password to continue.
""", unsafe_allow_html=True)
pw = st.text_input("Password", type="password", label_visibility="collapsed",
placeholder="Password…")
if pw:
if pw == required:
st.session_state._auth = True
st.rerun()
else:
st.error("Incorrect password.")
st.stop()
def _load_history() -> list:
if HISTORY_FILE.exists():
try:
return json.loads(HISTORY_FILE.read_text())
except Exception:
return []
return []
def _save_history(messages: list) -> None:
HISTORY_FILE.write_text(json.dumps(messages, ensure_ascii=False, indent=2))
# ── page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title="Teachings Q&A",
page_icon="☸️",
layout="wide",
initial_sidebar_state="collapsed",
)
_password_gate()
# ── CSS ───────────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── FAQ data ─────────────────────────────────────────────────────────────────
import hashlib as _hashlib
import re as _re
from collections import defaultdict as _defaultdict
def _q_hash(question: str) -> str:
"""Stable topic-cache key — must match classify_faq_topics.py:q_hash."""
normalized = " ".join(question.lower().split())
return _hashlib.sha256(normalized.encode()).hexdigest()[:16]
_TOPIC_ORDER = [
"Meditation Practice",
"Craving & Desire",
"Anger & Difficult Emotions",
"Relationships & Others",
"Fear & Anxiety",
"Enlightenment & The Path",
"Kamma & Rebirth",
"Non-Self & The Mind",
"Daily Life & Practice",
"Other",
]
@st.cache_data
def _load_faq_data() -> list[tuple[str, list[tuple[str, str]]]]:
"""Load lecture_qa chunks grouped by LLM-assigned topic.
Falls back to chapter-based grouping if faq_topics.json not yet generated.
Returns list of (topic_label, [(q_full, a), ...])."""
chunks_path = Path("data/glp/lecture_chunks.jsonl")
topics_path = Path("data/glp/faq_topics.json")
if not chunks_path.exists():
return []
chunks = [json.loads(l) for l in chunks_path.open(encoding="utf-8") if l.strip()]
qa = [c for c in chunks if c.get("content_type") == "lecture_qa"]
def _parse_qa(text: str) -> tuple[str, str]:
if "\nA:" in text:
q_raw, a_raw = text.split("\nA:", 1)
return q_raw.replace("Q:", "", 1).strip(), a_raw.strip()
return text[:200], text
# ── topic-based grouping (LLM-classified, keyed by question hash) ─────────
if topics_path.exists():
topic_map: dict[str, str] = json.loads(topics_path.read_text())
groups: dict[str, list] = _defaultdict(list)
for c in qa:
q, a = _parse_qa(c["text"])
topic = topic_map.get(_q_hash(q), "Other")
groups[topic].append((q, a))
return [
(t, groups[t]) for t in _TOPIC_ORDER if t in groups
] + [(t, groups[t]) for t in sorted(groups) if t not in _TOPIC_ORDER]
# ── fallback: chapter-based grouping ─────────────────────────────────────
def _sort_key(sf: str) -> tuple:
m = _re.match(r"GLP-Ch(\d+)-", sf)
return (int(m.group(1)), sf) if m else (0, sf)
def _group_key(sf: str) -> str:
m = _re.match(r"(GLP-Ch\d+-[^_]+)_[A-Za-z0-9_-]{10,}\.txt$", sf)
return m.group(1) if m else _re.sub(r"_[A-Za-z0-9_-]{10,}\.txt$", "", sf)
def _display_label(gkey: str) -> str:
m = _re.match(r"GLP-Ch\d+-(.+)", gkey)
if m:
return m.group(1).replace("-", " ")
return _re.sub(r"^GLP-", "", gkey).replace("-", " ")
ch_groups: dict[str, list] = _defaultdict(list)
sort_keys: dict[str, tuple] = {}
for c in qa:
sf = Path(c["source_file"]).name
gk = _group_key(sf)
ch_groups[gk].append(_parse_qa(c["text"]))
if gk not in sort_keys:
sort_keys[gk] = _sort_key(sf)
return [
(_display_label(gk), ch_groups[gk])
for gk in sorted(ch_groups, key=lambda k: sort_keys[k])
]
@st.cache_data(show_spinner=False)
def _faq_semantic_search(query: str, top_n: int = 30) -> list[tuple[str, str, str]]:
"""Semantic (ANN) search over lecture_qa chunks via the shared ChromaDB
embedding index. Returns ranked [(question, answer, topic), ...]."""
import numpy as _np
from retrieval import _model, _collection
topics_path = Path("data/glp/faq_topics.json")
topic_map = json.loads(topics_path.read_text()) if topics_path.exists() else {}
embedding = _np.array(next(iter(_model().embed([query])))).tolist()
res = _collection().query(
query_embeddings=[embedding],
n_results=top_n,
where={"content_type": {"$eq": "lecture_qa"}},
include=["documents"],
)
out: list[tuple[str, str, str]] = []
for doc in res["documents"][0]:
if "\nA:" in doc:
q_raw, a_raw = doc.split("\nA:", 1)
q, a = q_raw.replace("Q:", "", 1).strip(), a_raw.strip()
else:
q, a = doc[:200], doc
topic = topic_map.get(_q_hash(q), "Other")
out.append((q, a, topic))
return out
# PREAMBLE_RE strips the rambling opener before the real question
_PREAMBLE_RE = _re.compile(
r"^(?:(?:okay|ok|well said|thank you|thanks|hi|hello|"
r"yes|yeah|yep|alright|right|sure|great|good|so|um+|uh+|"
r"ah+|er+|i see|i mean)[,\.]?\s*)+",
_re.IGNORECASE,
)
def _q_title(q: str, max_len: int = 115) -> str:
"""Return a clean, scannable question title."""
cleaned = _PREAMBLE_RE.sub("", q).strip()
if cleaned:
cleaned = cleaned[0].upper() + cleaned[1:]
else:
cleaned = q.strip()
if len(cleaned) > max_len:
cut = cleaned[:max_len].rsplit(" ", 1)[0]
return cut.rstrip(",.") + "…"
return cleaned
# ── top bar ───────────────────────────────────────────────────────────────────
st.markdown("""
✿
Teachings Q&A
David Roylance · 13 books · 6 presentations · 85 lectures
""", unsafe_allow_html=True)
# ── session state ─────────────────────────────────────────────────────────────
if "messages" not in st.session_state:
st.session_state.messages = _load_history()
if "pending_sources" not in st.session_state:
st.session_state.pending_sources = []
# ── tabs ─────────────────────────────────────────────────────────────────────
tab_qa, tab_faq = st.tabs(["Q&A", "From the Sessions"])
BADGE = {
"scripture": ("Scripture", "b-scripture"),
"commentary": ("Commentary", "b-commentary"),
"book_front_matter": ("Front matter","b-front"),
"slide_content": ("Slide", "b-slide"),
"lecture_commentary": ("Lecture", "b-lecture"),
"lecture_qa": ("Q&A", "b-qa"),
"guided_meditation": ("Meditation", "b-meditation"),
}
SUGGESTIONS = [
"What is craving?",
"How does one attain stream-entry?",
"What is the role of a teacher?",
"Explain the Five Precepts",
"What is the purpose of meditation?",
]
def linkify_citations(text: str, sources: list[dict]) -> str:
"""Replace [Source N] patterns with inline superscript citation chips."""
url_map = {s["index"]: s.get("source_url", "") for s in sources}
def _replace(m: re.Match) -> str:
nums = [int(n) for n in re.findall(r"\d+", m.group(0))]
parts = []
for n in nums:
url = url_map.get(n, "")
href = url if url else f"#src-{n}"
target = ' target="_blank"' if url else ""
parts.append(
f'{n}'
)
return "".join(parts)
return re.sub(r"\[(?:Source \d+)(?:,\s*(?:Source )?\d+)*\]", _replace, text)
TYPE_LABEL = {
"scripture": "Scripture",
"commentary": "Commentary",
"book_front_matter": "Front Matter",
"slide_content": "Slide",
"lecture_commentary": "Lecture",
"lecture_qa": "Q&A",
"guided_meditation": "Meditation",
}
def _card_lines(s: dict) -> tuple[str, str]:
"""Return (line1, line2) for a source card's structured metadata."""
ct = s["content_type"]
label = s["source_label"].replace("&", "&").replace("<", "<").replace(">", ">")
vol_num = s.get("vol_num") or 0
vol_title = (s.get("vol_title") or "").replace("&", "&").replace("<", "<").replace(">", ">")
ch_num = s.get("ch_num") or 0
ch_title = (s.get("ch_title") or "").replace("&", "&").replace("<", "<").replace(">", ">")
page = s.get("page") or 0
refs = " · ".join(s["sutta_refs"]) if s["sutta_refs"] else ""
if ct in ("scripture", "commentary", "book_front_matter"):
line1 = vol_title or label
parts2 = []
if vol_num:
parts2.append(f"Vol {vol_num}")
if ch_num:
parts2.append(f"Ch {ch_num}")
if ch_title:
parts2.append(ch_title[:50])
if page:
parts2.append(f"p. {page}")
if refs:
parts2.append(refs)
line2 = " · ".join(parts2)
elif ct == "slide_content":
line1 = label
line2 = f"p. {page}" if page else ""
else:
# lectures
line1 = label
line2 = refs if refs else ""
return line1, line2
def render_source_cards(sources: list[dict], show_excerpts: bool) -> str:
html = 'Sources
'
for s in sources:
ct = s["content_type"]
_, badge_cls = BADGE.get(ct, (ct, "b-front"))
type_lbl = TYPE_LABEL.get(ct, ct)
url = s.get("source_url", "")
idx = s["index"]
score = s["score"]
line1, line2 = _card_lines(s)
line1_html = (
f'
{line1}' if url else line1
)
excerpt = ""
if show_excerpts:
txt = s["text"][:280].replace("&", "&").replace("<", "<").replace(">", ">").replace("\n", " ")
if len(s["text"]) > 280:
txt += "…"
excerpt = f'
{txt}
'
html += f"""
[{idx}]
{type_lbl}
{score}
{line1_html}
{f'
{line2}
' if line2 else ''}
{excerpt}
"""
html += "
'
for msg in messages:
if msg["role"] == "user":
q = msg["content"].replace("&", "&").replace("<", "<").replace(">", ">")
html += f'''
▍
{q}
'''
else:
sources = msg.get("sources") or []
content = msg["content"]
usage = msg.get("usage") or {}
cached_badge = '
cached' if msg.get("cached") else ""
token_line = ""
if usage:
token_line = (
f'
'
f'{usage.get("input","")} in · {usage.get("output","")} out'
f'{cached_badge}
'
)
DECLINE_TRIGGER = "The source passages here don't cover that."
if content.strip().startswith(DECLINE_TRIGGER):
safe = content.replace("&", "&").replace("<", "<").replace(">", ">")
html += f'
'
else:
answer_with_chips = linkify_citations(content, sources)
answer_html = _md.markdown(answer_with_chips, extensions=["nl2br"])
html += f'
{answer_html}
'
html += f'{token_line}
'
html += '
✿
Ask a question about the Teachings.
Answers drawn from David Roylance's books,
retreat presentations, and class transcripts.
What is craving?
How does one attain stream-entry?
What is the role of a teacher?
Explain the Five Precepts
""", unsafe_allow_html=True)
else:
st.markdown(render_qa_thread(st.session_state.messages), unsafe_allow_html=True)
if question := st.chat_input("Ask a question…"):
st.session_state.messages.append({"role": "user", "content": question})
with st.spinner(""):
try:
result = get_answer(
question,
n_results=n_results,
content_type=content_type,
volume=volume,
)
st.session_state.messages.append({
"role": "assistant",
"content": result["answer"],
"sources": result["sources"],
"usage": result["usage"],
"cached": result.get("cached", False),
})
st.session_state.pending_sources = result["sources"]
_save_history(st.session_state.messages)
except Exception as e:
st.error(f"Error: {e}")
st.rerun()
# ── sources column ────────────────────────────────────────────────────────
with src_col:
latest_sources = None
for msg in reversed(st.session_state.messages):
if msg["role"] == "assistant" and msg.get("sources"):
latest_sources = msg["sources"]
break
if latest_sources:
st.markdown(
render_source_cards(latest_sources, show_excerpts),
unsafe_allow_html=True,
)
else:
st.markdown("""
📎
Sources from the retrieved passages
will appear here
""", unsafe_allow_html=True)
# ═══════════════════════════════════════════════════════════════════════════════
# TAB 2 — FROM THE SESSIONS (FAQ)
# ═══════════════════════════════════════════════════════════════════════════════
with tab_faq:
st.markdown("", unsafe_allow_html=True)
_faq_col, _ = st.columns([5, 2], gap="large")
with _faq_col:
_faq_data = _load_faq_data()
# ── topic selector (with counts) ──────────────────────────────────────
_topic_labels = ["All topics"] + [
f"{label} ({len(pairs)})" for label, pairs in _faq_data
]
_label_to_topic = {
f"{label} ({len(pairs)})": label for label, pairs in _faq_data
}
_picked = st.radio(
"Topic", _topic_labels, horizontal=True,
label_visibility="collapsed", key="faq_topic",
)
_selected_topic = _label_to_topic.get(_picked) # None = all
_faq_search = st.text_input(
"Search questions",
placeholder="Search by meaning (e.g. 'dealing with a difficult spouse')…",
label_visibility="collapsed",
key="faq_search",
)
st.markdown("", unsafe_allow_html=True)
def _render_qa(q: str, a: str) -> None:
title = _q_title(q)
with st.expander(title):
if q.strip().lower() != title.lower().rstrip("…"):
st.markdown(f"*{q}*")
st.markdown("---")
paras = [p.strip() for p in a.split("\n\n") if p.strip()]
for p in (paras or [a]):
st.markdown(p)
_query = _faq_search.strip()
if _query:
# ── semantic (ANN) search ─────────────────────────────────────────
results = _faq_semantic_search(_query)
if _selected_topic:
results = [(q, a, t) for q, a, t in results if t == _selected_topic]
if results:
scope = f" in {_selected_topic}" if _selected_topic else ""
st.markdown(
f'',
unsafe_allow_html=True,
)
for _q, _a, _t in results:
_render_qa(_q, _a)
else:
st.markdown(
'No questions match that search.
',
unsafe_allow_html=True,
)
else:
# ── browse by topic ───────────────────────────────────────────────
for _group_label, _qa_pairs in _faq_data:
if _selected_topic and _group_label != _selected_topic:
continue
if not _selected_topic:
st.markdown(
f'',
unsafe_allow_html=True,
)
for _q, _a in _qa_pairs:
_render_qa(_q, _a)
# ── §4 citation binding (JS via parent frame) ─────────────────────────────────
_components.html("""
""", height=0)