')
cur = " iv-msg--current" if current else ""
return (f'
{AGENT_AVATAR}
'
f'
{safe}
')
def render_chat(messages: list[dict]) -> str:
rows = []
visible = [m for m in messages if m.get("role") in ("user", "assistant")]
last = len(visible) - 1
for i, m in enumerate(visible):
if not m.get("content", "").strip():
continue
role = "agent" if m["role"] == "assistant" else "user"
rows.append(_bubble(role, m["content"], current=(role == "agent" and i == last)))
return f'
{"".join(rows)}
'
def advance_to(session: SessionState, target) -> None:
if target is None:
return
if isinstance(target, str):
try:
target = State[target]
except KeyError:
return
while session.state < target:
session.advance()
# --------------------------------------------------------------------------
# Controls (deterministic per question)
# --------------------------------------------------------------------------
def _next_index(session: SessionState, idx: int) -> int:
"""Next applicable question index, skipping ones that don't apply."""
j = idx + 1
while j < REVIEW_INDEX and QUESTIONS[j].skip_if_in_origin and in_origin(session):
j += 1
return j
# The two multi-select questions get their own pre-mounted controls so their
# choices are set once (in the person's language) and never swapped at reveal —
# Gradio lags when it has to populate a CheckboxGroup's choices in the same beat
# it is shown, which made the options appear only after an extra "Continue".
_Q_GROUNDS = next(q for q in QUESTIONS if q.field == "persecution_types")
_Q_DOCS = next(q for q in QUESTIONS if q.field == "documents_available")
def _active_control(session: SessionState, idx: int):
"""Return (which, choices) for the active control at ``idx``.
which ∈ {"radio","grounds","docs","country","text"}."""
lang = session.language
if idx == CORRECT_INDEX:
return "text", None
if idx >= REVIEW_INDEX:
return "radio", [t(lang, "review_yes"), t(lang, "review_no")]
q = QUESTIONS[idx]
if q.control == "yesno":
return "radio", option_labels(lang, q)
if q.control == "choice":
return ("grounds" if q.field == "persecution_types" else "docs"), option_labels(lang, q)
if q.control == "country":
return "country", None
return "text", None
# Order of the five control updates everywhere: (radio, grounds, docs, country, text).
# radio/grounds/docs are choice controls: ALWAYS mounted-visible, driven by choices
# (empty = collapsed by CSS). country/text are toggled with `visible`.
def control_updates(session: SessionState, idx: int):
"""Updates that make exactly the active control appear. Choice controls get
their choices (empty when inactive — never a `visible` toggle); country/text
toggle visibility."""
which, choices = _active_control(session, idx)
lang = session.language
return (
gr.update(choices=choices if which == "radio" else [], value=None),
gr.update(choices=option_labels(lang, _Q_GROUNDS) if which == "grounds" else [], value=[]),
gr.update(choices=option_labels(lang, _Q_DOCS) if which == "docs" else [], value=[]),
gr.update(visible=which == "country", value=None),
gr.update(visible=which == "text", value=""),
)
# Clears every control: choice controls emptied (collapse via CSS), country/text
# hidden. Used while the conversation advances, before the active control is set
# in the next round-trip.
def _clear_controls():
return (
gr.update(choices=[], value=None),
gr.update(choices=[], value=[]),
gr.update(choices=[], value=[]),
gr.update(visible=False),
gr.update(visible=False),
)
def _store(session: SessionState, q: Question, radio_v, grounds_v, docs_v, country_v, text_v):
lang = session.language
if q.control == "country":
raw = country_name(country_v) if country_v else (text_v or "").strip()
if not raw:
return None
setattr(session.interview, q.field, raw)
return raw
if q.control == "yesno":
if not radio_v:
return None
is_yes = radio_v == t(lang, "opt_yes")
setattr(session.interview, q.field, is_yes)
return radio_v
if q.control == "choice":
multi_v = grounds_v if q.field == "persecution_types" else docs_v
if not multi_v:
return None
# map translated labels back to canonical English labels
rev = {t(lang, oid): t("English", oid) for oid in q.options}
canon = [rev.get(v, v) for v in multi_v]
setattr(session.interview, q.field, canon)
return ", ".join(multi_v)
# text
raw = (text_v or "").strip()
if not raw:
return None
if q.kind == "list_text":
setattr(session.interview, q.field, [p.strip() for p in raw.replace(";", ",").split(",") if p.strip()])
else:
setattr(session.interview, q.field, raw)
return raw
def _facts_recap(session: SessionState) -> str:
iv = session.interview
bits = []
for label, val in [
("•", iv.current_country), ("•", iv.origin_country),
("•", ", ".join(iv.persecution_types) if iv.persecution_types else None),
("•", iv.displacement_duration),
("•", ", ".join(iv.documents_available) if iv.documents_available else None),
("•", ", ".join(iv.languages_spoken) if iv.languages_spoken else None),
("•", ", ".join(iv.destination_preferences) if iv.destination_preferences else None),
]:
if val:
bits.append(f"{label} {val}")
return "\n".join(bits)
def _agent_message_for(session: SessionState, idx: int, *, welcome: bool = False) -> str:
if idx == CORRECT_INDEX:
return t(session.language, "q_correct")
text = question_text(session.language, QUESTIONS[idx], session)
if welcome:
return f"{t(session.language, 'welcome')}\n{text}"
return text
def _labeled_facts(session: SessionState) -> str:
iv = session.interview
fields = [
("Country of origin", iv.origin_country),
("Current country", iv.current_country),
("What happened", iv.free_text_history),
("Reason", ", ".join(iv.persecution_types) if iv.persecution_types else None),
("Immediate danger", None if iv.immediate_danger is None else ("yes" if iv.immediate_danger else "no")),
("Time since leaving", iv.displacement_duration),
("Documents", ", ".join(iv.documents_available) if iv.documents_available else None),
("Languages", ", ".join(iv.languages_spoken) if iv.languages_spoken else None),
("Preferred destination", ", ".join(iv.destination_preferences) if iv.destination_preferences else None),
]
return "\n".join(f"{k}: {v}" for k, v in fields if v)
def _bold_labels(text: str) -> str:
"""Wrap the 'Label:' prefix of each line in ** so it renders bold."""
out = []
for line in text.split("\n"):
if ":" in line and not line.lstrip().startswith("**"):
label, _, rest = line.partition(":")
if 0 < len(label) <= 40:
out.append(f"**{label.strip()}:**{rest}")
continue
out.append(line)
return "\n".join(out)
def _labeled_fallback(session: SessionState) -> str:
lang = session.language
facts = _bold_labels(_labeled_facts(session))
return f"{t(lang, 'review_intro')}\n{facts}\n\n{t(lang, 'review_confirm')}"
async def _draft_review(session: SessionState, loop) -> str:
"""LLM-written labeled review summary in the person's language (bold labels)."""
lang = session.language or "English"
system_prompt = (
load_prompt("system")
+ f"\n\n# Right now\nIn {lang}, briefly summarise back what the person told you so they "
"can confirm. State each item on its own line, clearly labelled (country of origin, "
"current country, what happened, immediate danger, time since leaving, documents, "
"languages, preferred destination). Use ONLY the facts given — do not invent or omit. "
"End by asking, in one short sentence, whether it is correct."
)
acc = ""
try:
async for ev in loop.run("Facts:\n" + _labeled_facts(session), session=None,
system_prompt=system_prompt, thinking_level="off"):
if isinstance(ev, TextDeltaEvent):
acc += ev.delta
except Exception:
acc = ""
return _bold_labels(acc.strip()) if acc.strip() else _labeled_fallback(session)
_FIELD_KEYS = {
"origin_country", "current_country", "free_text_history", "immediate_danger",
"displacement_duration", "documents_available", "languages_spoken", "destination_preferences",
}
async def _apply_correction(session: SessionState, loop, correction: str) -> None:
"""Agentic correction: the LLM maps the person's free-text fix to fields."""
system_prompt = (
"You update a structured interview record from a person's free-text correction. "
"Output ONLY lines of the form field=value, using these field names exactly: "
+ ", ".join(sorted(_FIELD_KEYS)) + ". "
"immediate_danger must be yes or no. Only output the fields that should change. "
"No commentary."
)
prompt = (
"Current record:\n" + _labeled_facts(session)
+ f"\n\nThe person says: {correction}\n\nWhat should change?"
)
acc = ""
try:
async for ev in loop.run(prompt, session=None, system_prompt=system_prompt, thinking_level="off"):
if isinstance(ev, TextDeltaEvent):
acc += ev.delta
except Exception:
acc = ""
for line in acc.splitlines():
if "=" not in line:
continue
key, _, val = line.partition("=")
key, val = key.strip(), val.strip()
if key not in _FIELD_KEYS or not val:
continue
if key == "immediate_danger":
setattr(session.interview, key, val.lower().startswith("y"))
elif key in ("documents_available", "languages_spoken", "destination_preferences"):
setattr(session.interview, key, [p.strip() for p in val.replace(";", ",").split(",") if p.strip()])
else:
setattr(session.interview, key, val)
# --------------------------------------------------------------------------
# UI assembly
# --------------------------------------------------------------------------
class InterviewUI:
def __init__(self, **kw):
self.__dict__.update(kw)
def build(visible: bool = True, session_st=None, loop_st=None, slot_idx_st=None) -> InterviewUI:
session_st = session_st or gr.State(None)
loop_st = loop_st or gr.State(None)
slot_idx_st = slot_idx_st or gr.State(0)
with gr.Column(visible=visible, elem_classes=["screen-wrap"]) as column:
with gr.Column(elem_id="iv-screen"):
rail = gr.HTML(render_rail(State.SITUATION))
chat = gr.HTML(render_chat([]))
with gr.Column(elem_id="iv-responder"):
lbl = gr.HTML('
Your answer
') # localised per turn
# Choice controls are mounted VISIBLE with empty choices and driven
# by their choices alone (CSS collapses them while empty). Toggling
# a CheckboxGroup's `visible` doesn't build its option DOM, so it
# would paint empty until the next click — this avoids that entirely.
radio = gr.Radio(choices=[], label="", show_label=False, visible=True, elem_id="iv-choice")
multi_grounds = gr.CheckboxGroup(choices=[], label="", show_label=False,
visible=True, elem_id="iv-multi")
multi_docs = gr.CheckboxGroup(choices=[], label="", show_label=False,
visible=True, elem_id="iv-multi-docs")
country = gr.Dropdown(choices=country_choices(), label="", show_label=False, visible=False,
allow_custom_value=True, filterable=True, elem_id="iv-country")
text = gr.Textbox(label="", show_label=False, lines=3, visible=False,
placeholder="…", elem_id="iv-text")
cont = gr.Button("Continue →", elem_id="iv-continue")
stream_outputs = [chat, rail, radio, multi_grounds, multi_docs, country, text,
session_st, loop_st, slot_idx_st, lbl, cont]
def _chrome_updates(session):
lang = session.language if session else "English"
return (
gr.update(value=f'
{html.escape(_chrome(lang, "your_answer"))}
'),
gr.update(value=_chrome(lang, "continue")),
)
async def _present(session, loop, idx):
"""Append the agent's message (scripted question, LLM review, or the
correction prompt) and show the right control for the step."""
target = State.REVIEW if idx >= REVIEW_INDEX else QUESTIONS[idx].phase
advance_to(session, target)
if idx == CORRECT_INDEX:
msg = _agent_message_for(session, idx)
elif idx >= REVIEW_INDEX:
msg = await _draft_review(session, loop)
else:
msg = _agent_message_for(session, idx, welcome=(idx == 0))
session.messages = list(session.messages) + [{"role": "assistant", "content": msg}]
return (render_chat(session.messages), render_rail(session.state, session.language),
*control_updates(session, idx), session, idx)
_KEEP5 = (gr.update(), gr.update(), gr.update(), gr.update(), gr.update())
# Each interview turn is two round-trips. Round 1 (the click handler) advances
# the conversation and CLEARS the controls. Round 2 (a chained
# ``.then(reveal_controls)``) sets the active control on its own — a separate
# render cycle with nothing else changing — so a CheckboxGroup's choices paint
# reliably (setting choices on an already-mounted-visible group always works;
# toggling its `visible` does not).
async def _advance(o, loop):
"""One frame: advance chat + rail, clear all controls (the active control
is set in the next round-trip). ``o`` is ``_present``'s tuple (…, session, idx)."""
session, idx = o[7], o[8]
yield (o[0], o[1], *_clear_controls(), session, loop, idx, *_chrome_updates(session))
async def start(session, loop, slot_idx=0):
if session is None:
session = SessionState(); session.transition_to(State.INTAKE)
loop = loop or create_loop()
o = await _present(session, loop, 0)
async for frame in _advance(o, loop):
yield frame
async def on_continue(radio_v, grounds_v, docs_v, country_v, text_v, session, loop, idx):
if session is None:
session = SessionState(); session.transition_to(State.INTAKE)
loop = loop or create_loop()
o = await _present(session, loop, 0)
async for frame in _advance(o, loop):
yield frame
return
loop = loop or create_loop()
lang = session.language
# Correction step: the person typed what to change; the agent applies it.
if idx == CORRECT_INDEX:
correction = (text_v or "").strip()
if not correction:
yield (gr.update(), gr.update(), *_KEEP5, session, loop, idx, gr.update(), gr.update())
return
session.messages = list(session.messages) + [{"role": "user", "content": correction}]
await _apply_correction(session, loop, correction)
o = await _present(session, loop, REVIEW_INDEX)
async for frame in _advance(o, loop):
yield frame
return
# Review step
if idx >= REVIEW_INDEX:
if not radio_v:
yield (gr.update(), gr.update(), *_KEEP5, session, loop, idx, gr.update(), gr.update())
return
session.messages = list(session.messages) + [{"role": "user", "content": radio_v}]
if radio_v == t(lang, "review_yes"):
advance_to(session, State.ASSESSMENT) # triggers assessment via .then()
yield (render_chat(session.messages), render_rail(session.state, session.language),
*_clear_controls(), session, loop, idx, gr.update(), gr.update())
return
o = await _present(session, loop, CORRECT_INDEX) # ask what to change
async for frame in _advance(o, loop):
yield frame
return
q = QUESTIONS[idx]
display = _store(session, q, radio_v, grounds_v, docs_v, country_v, text_v)
if display is None:
yield (gr.update(), gr.update(), *_KEEP5, session, loop, idx, gr.update(), gr.update())
return
session.messages = list(session.messages) + [{"role": "user", "content": display}]
o = await _present(session, loop, _next_index(session, idx))
async for frame in _advance(o, loop):
yield frame
control_comps = [radio, multi_grounds, multi_docs, country, text]
def reveal_controls(session, idx):
"""Second round-trip: set the active control on its own (no chat change
competing for the paint). Once the interview has handed off to the
assessment, clear everything (controls collapse)."""
if session is None or session.state >= State.ASSESSMENT:
return _clear_controls()
return control_updates(session, idx)
# show_progress="hidden": the choice controls are always mounted-visible, so
# Gradio's generic per-component "processing…" overlay would paint a spinner on
# each of them (it showed twice). The conversation itself signals progress.
continue_event = cont.click(
on_continue,
inputs=[radio, multi_grounds, multi_docs, country, text, session_st, loop_st, slot_idx_st],
outputs=stream_outputs,
show_progress="hidden",
).then(
reveal_controls,
inputs=[session_st, slot_idx_st],
outputs=control_comps,
show_progress="hidden",
)
return InterviewUI(
column=column, session=session_st, loop=loop_st, slot_idx=slot_idx_st,
start_fn=start, start_inputs=[session_st, loop_st, slot_idx_st],
stream_outputs=stream_outputs, continue_btn=cont, continue_event=continue_event,
reveal_fn=reveal_controls, reveal_inputs=[session_st, slot_idx_st], reveal_outputs=control_comps,
)
__all__ = ["build", "InterviewUI", "INTERVIEW_CSS", "render_chat", "render_rail",
"advance_to", "control_updates"]