"""Render pipeline trace events as HTML, for the Gradio Space.
The FastAPI build streams raw trace events to the browser and lets `web/static/app.js`
build the DOM. ZeroGPU — the free tier that can actually run Python — is Gradio-SDK-only,
and Gradio owns the page, so there is nowhere to hang that JS. This module does the same
job server-side: same event stream in, the same markup out, so `web/static/styles.css`
styles both without a single change.
`Renderer` is fed events one at a time and can emit its current HTML after each, which is
what makes the stepper animate as the pipeline runs rather than appearing all at once.
"""
from __future__ import annotations
from html import escape as _e
# Chips worth surfacing on a result card, in display order. `edition_year` and
# `cohort_qualifier` come first because they are the whole point of this corpus: they are
# what separates the near-identical twins that naive top-k RAG confuses.
_CHIP_FIELDS = ("page_type", "faculty", "program", "subject_code")
def _fmt(value: float, places: int = 3) -> str:
return f"{float(value):.{places}f}"
def _md_link_text(text: str) -> str:
"""Escape a Markdown link label. Calendar titles routinely carry the edition in square
brackets ("Academic Standing [2026/27]"), which would otherwise break the link."""
return text.replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]")
class Renderer:
"""Accumulates trace events and renders the three panels."""
def __init__(self, question: str) -> None:
self.question = question
self.steps: list[dict] = [] # {title, meta: [html], extras: [html]}
self.hops: list[dict] = [] # {query, columns, shortlist, finals}
self.docs: dict[int, dict] = {}
self.order: list[int] = []
self.answer_text = ""
self.route = ""
self.error = ""
# ---------------------------------------------------------------- ingest
def feed(self, event: dict) -> None:
kind = event.get("kind")
handler = getattr(self, f"_on_{kind}", None)
if handler:
handler(event)
def _step(self) -> dict:
if not self.steps:
self.steps.append({"title": "Running", "meta": [], "extras": []})
return self.steps[-1]
def _on_step(self, ev: dict) -> None:
self.steps.append({"title": _pretty_step(ev["title"]), "meta": [], "extras": []})
def _on_detail(self, ev: dict) -> None:
self._step()["meta"].append(
f"{_e(str(ev['label']))}: {_e(str(ev['value']))}"
)
def _on_route(self, ev: dict) -> None:
decision = ev["decision"]
tail = ("split into sub-questions and searched one at a time."
if decision == "complex" else "answered from a single search.")
self._step()["meta"].append(
f"Classified as {_e(decision)}, so it gets {tail}"
)
def _on_subquestions(self, ev: dict) -> None:
items = "".join(f"{_e(s)}" for s in ev["items"])
self._step()["extras"].append(f'{items}
')
def _on_llm_call(self, ev: dict) -> None:
self._step()["extras"].append(
''
f'prompt and reply · {_e(ev["model"])} · {ev["ms"]} ms
'
''
f'
System
{_e(ev["system"])}'
f'
User
{_e(ev["user"])}'
f'
Reply
{_e(ev["reply"])}'
"
"
)
def _on_merge(self, ev: dict) -> None:
self.order = ev["order"]
dropped = ev["dropped"]
self._step()["meta"].append(
f"Interleaved {len(ev['per_hop'])} rankings to give "
f"{len(ev['order'])} unique excerpts "
f"({dropped} duplicate{'' if dropped == 1 else 's'} dropped)."
)
def _on_retrieval_start(self, ev: dict) -> None:
self.hops.append({
"query": ev["query"], "columns": {}, "shortlist": set(), "finals": [],
})
def _on_candidates(self, ev: dict) -> None:
if self.hops:
self.hops[-1]["columns"][ev["stage"]] = ev["items"]
def _on_shortlist(self, ev: dict) -> None:
if self.hops:
self.hops[-1]["shortlist"] = set(ev["ids"])
def _on_retrieval_final(self, ev: dict) -> None:
if self.hops:
self.hops[-1]["finals"] = ev["items"]
for doc in ev["items"]:
self.docs[doc["id"]] = doc
if not self.order:
self.order = [d["id"] for d in ev["items"]]
# ---------------------------------------------------------------- render
def trace_html(self, running: bool = False) -> str:
if not self.steps:
return 'Ask a question and the steps will show up here.
'
rows = []
for i, step in enumerate(self.steps):
last = i == len(self.steps) - 1
state = "active" if (running and last) else "done"
meta = "".join(f'{m}
' for m in step["meta"])
rows.append(
f'{_e(step["title"])}
'
f'{meta}{"".join(step["extras"])}'
)
return f'{"".join(rows)}
'
def retrieval_html(self) -> str:
if not self.hops:
return 'The excerpts each search finds will show up here.
'
blocks = []
for i, hop in enumerate(self.hops, start=1):
multi = len(self.hops) > 1 or hop["query"].strip() != self.question.strip()
label = f"Sub-question {i}" if multi else "Single query"
survivors = {d["id"] for d in hop["finals"]}
columns = "".join(
_column(name, css, hop["columns"].get(key, []), hop["shortlist"], survivors)
for key, name, css in (
("dense", "Vector", "dense"),
("bm25", "Keyword", "bm25"),
("rrf", "Fused", "rrf"),
)
)
finals = ""
if hop["finals"]:
finals = (
''
f'
Final {len(hop["finals"])}
'
+ "".join(_doc_card(d) for d in hop["finals"]) + ""
)
blocks.append(
f'{_e(label)}: '
f'{_e(hop["query"])}
'
f'
{columns}
'
'
shortlisted '
'for reranking · kept '
"in the final results
"
f"{finals}
{_e(doc["section"])}
'
if doc.get("section") else "")
link = (f''
f"official page ↗") if doc.get("url") else ""
return (
''
f'{_e(doc["title"])}'
f'id {doc["id"]} · {_fmt(doc["score"])}
'
f'{section}
{"".join(chips)}
'
f'
{link}
'
f'
excerpt
'
f'{_e(doc.get("text", ""))}