",
unsafe_allow_html=True
)
# ββ Model selector with friendly names ββββββββββββββββββββββββββββββββ
st.markdown("#### Language Model")
if isinstance(AVAILABLE_MODELS, dict):
friendly_names = list(AVAILABLE_MODELS.keys())
model_ids = list(AVAILABLE_MODELS.values())
if st.session_state.selected_model not in model_ids:
st.session_state.selected_model = model_ids[0]
current_idx = model_ids.index(st.session_state.selected_model)
chosen_name = st.selectbox(
"Language model", friendly_names, index=current_idx,
help="Model used for answer generation.",
label_visibility="collapsed",
)
st.session_state.selected_model = AVAILABLE_MODELS[chosen_name]
st.markdown(
f"
{st.session_state.selected_model}
",
unsafe_allow_html=True,
)
else:
if st.session_state.selected_model not in AVAILABLE_MODELS:
st.session_state.selected_model = DEFAULT_MODEL
st.session_state.selected_model = st.selectbox(
"Language model", AVAILABLE_MODELS,
index=AVAILABLE_MODELS.index(st.session_state.selected_model),
help="Model used for answer generation.",
label_visibility="collapsed",
)
# ββ Vector DB (hidden in HF Spaces) ββββββββββββββββββββββββββββββββββ
if not IS_HF_SPACES:
st.markdown("#### Vector Database")
st.session_state.selected_vector_db = st.selectbox(
"Vector database", ["Chroma", "Pinecone"],
index=["Chroma", "Pinecone"].index(st.session_state.selected_vector_db),
help="Where embeddings are stored.",
label_visibility="collapsed",
)
# ββ File uploader βββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("#### Upload Documents")
if st.session_state.processing_status != "success":
if not st.session_state.uploaded_files:
st.markdown(
"""
""",
unsafe_allow_html=True,
)
uploaded = st.file_uploader(
"Upload documents",
type=["pdf", "docx", "pptx", "txt", "md"],
accept_multiple_files=True,
help="PDF, DOCX, PPTX, TXT, MD β max 25 MB each",
label_visibility="collapsed",
key=f"file_uploader_{st.session_state.clear_counter}",
)
if uploaded:
st.session_state.uploaded_files = uploaded
_rerun()
# ββ Show selected-file pills when files chosen but not yet processed ββ
if st.session_state.uploaded_files and st.session_state.processing_status != "success":
for f in st.session_state.uploaded_files:
ext = f.name.rsplit(".", 1)[-1].upper() if "." in f.name else "FILE"
size_kb = round(f.size / 1024, 1)
st.markdown(
f"
"
f"
"
f"
"
f""
f"
"
f"
"
f"
{f.name}
"
f"
"
f"{ext} Β· {size_kb} KB
"
f"
"
f"
Ready
"
f"
",
unsafe_allow_html=True,
)
# ββ Process button ββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("", unsafe_allow_html=True)
if st.button("βΆ Process & Index", type="primary", use_container_width=True):
if not st.session_state.uploaded_files:
st.warning("Upload at least one document first.")
else:
# Wipe everything doc-specific BEFORE processing so stale
# questions never survive into the new session even briefly
st.session_state.starter_questions = []
st.session_state.chat_history = []
st.session_state.vector_store = None
st.session_state.documents = []
st.session_state.pending_question = None
with st.spinner("Chunking and embeddingβ¦"):
try:
chunks = _load_documents(st.session_state.uploaded_files)
store = _create_vector_store(
chunks, st.session_state.selected_vector_db
)
st.session_state.vector_store = store
st.session_state.documents = chunks
st.session_state.processing_status = "success"
st.session_state.processing_message = (
f"β Indexed {len(chunks)} chunks from "
f"{len(st.session_state.uploaded_files)} file(s)."
)
st.session_state.processing_message_type = "success"
st.session_state.chat_history = []
st.session_state.pending_question = None
# Always wipe stale questions BEFORE generating
# new ones β if generation throws or returns [],
# the old doc's questions are already gone.
st.session_state.starter_questions = []
st.session_state["_saved_starter_questions"] = []
generator = ChatGenerator(model=st.session_state.selected_model)
questions = _generate_starter_questions(store, generator)
# Store in both the visible list and the saved mirror.
# The mirror lets _new_chat() restore chips without
# re-querying the LLM.
st.session_state.starter_questions = questions
st.session_state["_saved_starter_questions"] = list(questions)
except Exception as exc:
st.session_state.processing_status = "error"
st.session_state.processing_message = f"Processing failed: {exc}"
st.session_state.processing_message_type = "error"
# Ensure stale questions are never shown on error
st.session_state.starter_questions = []
st.session_state["_saved_starter_questions"] = []
# Clear all ββββββββββββββββββββββββββββββββββββββββββββββ
if st.button("Clear all", type="secondary", use_container_width=True):
_clear_all()
_rerun()
# ββ Indexed file cards ββββββββββββββββββββββββββββββββββββββββββββββββ
# Guard: skip during chip-click reruns to prevent double render.
# pending_question is set by chip click and consumed in _render_chat_interface.
if (
st.session_state.uploaded_files
and st.session_state.processing_status == "success"
and not st.session_state.get("pending_question")
):
st.markdown(
"
"
"Indexed Files
",
unsafe_allow_html=True,
)
chunk_counts = Counter(
doc.metadata.get("filename", "unknown")
for doc in st.session_state.documents
)
seen: set = set()
for file in st.session_state.uploaded_files:
if file.name in seen:
continue
seen.add(file.name)
n_chunks = chunk_counts.get(file.name, "β")
size_kb = round(file.size / 1024, 1)
ext = file.name.rsplit(".", 1)[-1].upper() if "." in file.name else "FILE"
st.markdown(
f"
""",
unsafe_allow_html=True,
)
# ββ Chat interface ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def _render_source_card(source: dict):
pct = source["match_pct"]
chunk_raw = source.get("chunk_id", "β")
# Chunk label is now secondary β rendered after filename in muted style
chunk_label = f"chunk {chunk_raw}" if chunk_raw not in ("β", "", None) else ""
st.markdown(
f"""
{source['filename']}
{f"{chunk_label}" if chunk_label else ""}
{source['text']}
{pct}%
match
""",
unsafe_allow_html=True,
)
# Helper β detect queries where low retrieval scores are structurally expected
_SUMMARY_PATTERNS = re.compile(
r"^\s*(summarize|summarise|give me a summary|overview|what is this (doc|document|about)|"
r"tldr|tl;dr|brief me|describe this)",
re.IGNORECASE,
)
def _is_retrieval_limited_query(query: str, pct: int) -> bool:
"""True when low confidence is caused by retrieval structure, not a model failure."""
if pct <= 40:
# Anything at or below 40% is retrieval-limited by definition β
# the model found weak matches regardless of query type.
return True
if _SUMMARY_PATTERNS.match(query.strip()):
return True
return False
def _render_confidence(trace, query: str = ""):
"""Capability 5 β always render confidence bar below the answer card."""
if not trace:
return
pct = round(trace.confidence * 100)
tier = trace.confidence_label # 'high' | 'medium' | 'low'
if tier == "low" and _is_retrieval_limited_query(query, pct):
tier_text = "Retrieval limited"
bar_class = "conf-bar-medium"
pct_class = "conf-pct-medium"
else:
tier_text = {
"high": "High confidence",
"medium": "Medium confidence",
"low": "Low confidence",
}[tier]
bar_class = f"conf-bar-{tier}"
pct_class = f"conf-pct-{tier}"
st.markdown(
f"
"
f"{tier_text}"
f"
"
f""
f"
"
f"{pct}%"
f"
",
unsafe_allow_html=True,
)
def _render_thumbs(turn_idx: int):
"""Capability 6 β thumbs feedback rendered as a clean HTML inline row."""
turn = st.session_state.chat_history[turn_idx]
existing = turn.get("feedback")
if existing:
icon = "π" if existing == "up" else "π"
st.markdown(
f"
"
f""
f"{icon} Feedback recorded
",
unsafe_allow_html=True,
)
return
col_up, col_dn, col_rest = st.columns([1, 1, 22])
with col_up:
if st.button("π", key=f"tb_up_{turn_idx}", help="Helpful"):
_record_feedback(turn_idx, thumbs_up=True)
with col_dn:
if st.button("π", key=f"tb_dn_{turn_idx}", help="Not helpful"):
_record_feedback(turn_idx, thumbs_up=False)
def _record_feedback(turn_idx: int, thumbs_up: bool):
"""
Capability 6 β write feedback into session state FeedbackStore and
annotate the chat_history turn so buttons become static.
"""
turn = st.session_state.chat_history[turn_idx]
turn["feedback"] = "up" if thumbs_up else "down"
# Lazily initialise FeedbackStore
if st.session_state.feedback_store is None:
st.session_state.feedback_store = FeedbackStore()
trace = turn.get("healing_trace")
chunk_ids = trace.retrieved_chunk_ids if trace else []
st.session_state.feedback_store.record(
query=turn["user"],
chunk_ids=chunk_ids,
thumbs_up=thumbs_up,
)
_rerun()
def _render_self_healing(trace, always_show: bool = False):
"""
Render self-healing badges + collapsible audit panel.
When always_show=True, renders even if no healing actions fired
(shows a minimal 'Pipeline ran cleanly' state for UI consistency).
"""
if not trace:
return
any_fired = trace.any_healing_fired
badge_text = trace.summary_badge() if trace else ""
# ββ Badge row ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
badges_html = (
"
"
""
"Faithfulness verified Β· no hallucination detected"
"
"
)
if badge_text:
badges_html += (
f"
"
f""
f"{badge_text}"
f"
"
)
st.markdown(f"
{badges_html}
", unsafe_allow_html=True)
# ββ Collapsible audit panel ββββββββββββββββββββββββββββββββββββββββββββ
steps = trace.steps if trace else []
# If no healing fired and we're in always_show mode, synthesise a clean-run step
if not steps:
if not always_show:
return
steps = ["Pipeline ran cleanly Β· no interventions required"]
uid = abs(hash(str(steps)))
st.markdown(
f"""""",
unsafe_allow_html=True,
)
rows_html = ""
for step in steps:
s = step.lower()
if "pipeline ran cleanly" in s:
dot_color, text_color = "#1D9E75", "#1D9E75"
elif any(w in s for w in ("complete", "success", "passed", "indexed")):
dot_color, text_color = "#1D9E75", "#1D9E75"
elif any(w in s for w in ("failed", "error", "invalid")):
dot_color, text_color = "#A32D2D", "#A32D2D"
elif any(w in s for w in ("filtered", "low-relevance", "inferential",
"re-answer", "threshold", "fallback", "faithfulness check")):
dot_color, text_color = "#BA7517", "#BA7517"
elif "confidence:" in s:
dot_color, text_color = "#7F77DD", "#afa9ec"
elif any(w in s for w in ("feedback:", "rerank", "rewrite skipped", "threshold adjusted")):
dot_color, text_color = "#5b8dd9", "#8ab4f8"
else:
dot_color, text_color = "#4e4d60", "#6b6980"
rows_html += (
f"
"
"Upload and process documents to start chatting.
",
unsafe_allow_html=True,
)
return
# ββ Consume pending_question FIRST before any rendering ββββββββββββββββββ
# This prevents the double-render: if a chip was clicked, we capture the
# question and clear state now, so the render below sees the final state.
pending = st.session_state.get("pending_question")
if pending:
st.session_state.pending_question = None
st.session_state.starter_questions = []
# Snapshot history BEFORE appending the new placeholder turn
history_for_context = list(st.session_state.chat_history)
st.session_state.chat_history.append({"user": pending, "assistant": "β¦"})
generator = ChatGenerator(model=st.session_state.selected_model)
generator._conversation_history = history_for_context # picked up by generate_answer
with st.spinner("Searching your documentsβ¦"):
try:
answer, relevant_docs, healing_trace = run_healing_pipeline(
query=pending,
vector_store=st.session_state.vector_store,
generator=generator,
k=6,
relevance_threshold=0.15 if _is_factual_query(pending) else 0.30,
feedback_store=st.session_state.feedback_store,
)
except RuntimeError as e:
err_msg = (
"β οΈ The demo's free API quota is temporarily exhausted. "
"Please try again later or run the app locally with Ollama."
if ("402" in str(e) or "Payment Required" in str(e) or "credits" in str(e).lower())
else f"β οΈ Something went wrong: {e}"
)
st.session_state.chat_history[-1] = {
"user": pending, "assistant": err_msg,
"sources": [], "healing_trace": None,
}
_rerun()
return
sources = []
for idx, (doc, score) in enumerate(relevant_docs, start=1):
meta = doc.metadata
sources.append({
"num": f"Source {idx}",
"filename": meta.get("filename") or meta.get("source") or f"Document {idx}",
"chunk_id": meta.get("chunk_id", "β"),
"match_pct": _score_to_pct(score),
"text": _excerpt_text(doc.page_content, max_chars=260),
"title": f"Source {idx}",
"match_label": f"{_score_to_pct(score)}% match",
})
st.session_state.chat_history[-1] = {
"user": pending, "assistant": answer,
"sources": sources, "healing_trace": healing_trace,
}
_rerun()
return
# ββ Suggested question cards (shown when chat is empty) βββββββββββββββββββ
if not st.session_state.chat_history and st.session_state.starter_questions:
questions_snapshot = list(st.session_state.starter_questions)
# Render the beautiful card UI via st.markdown (in parent frame, no iframe, no bg issues)
def _html_escape(s):
return s.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
cards_html = ""
for i, q in enumerate(questions_snapshot):
q_safe = _html_escape(q)
# data-idx used by the JS click handler below
cards_html += f"""
0{i+1}
{q_safe}
Ask β
"""
st.markdown(
f"""
β¦Suggested questions
{cards_html}
""",
unsafe_allow_html=True,
)
# Hidden native st.buttons β visually suppressed via CSS but Streamlit-clickable.
# The components.html script below wires each .sq-card click to its matching button.
btn_cols = st.columns(len(questions_snapshot))
for i, (col, q) in enumerate(zip(btn_cols, questions_snapshot)):
with col:
if st.button(q, key=f"sq_{i}", use_container_width=True):
st.session_state.pending_question = q
st.session_state.starter_questions = []
_rerun()
return
# Wire card clicks β hidden buttons via parent-frame JS.
# Buttons are found by matching their visible text to the question text.
# The same JS also hides the native buttons so only the styled cards show.
import json as _json
questions_json = _json.dumps(questions_snapshot)
components.html(
f"""
""",
height=0,
)
# ββ Render chat history βββββββββββββββββββββββββββββββββββββββββββββββββββ
for turn_idx, turn in enumerate(st.session_state.chat_history):
with st.chat_message("user"):
st.markdown(turn["user"])
with st.chat_message("assistant"):
n_src = len(turn.get("sources", []))
src_count_html = (
f"({n_src} source{'s' if n_src > 1 else ''})"
if n_src else ""
)
st.markdown(
f"