"""AdaptiveRAG β under-the-hood pipeline visualizer.
Run: streamlit run app.py
"""
from __future__ import annotations
import json
import logging
import os
import tempfile
import time
from pathlib import Path
os.environ.setdefault("ANONYMIZED_TELEMETRY", "False")
# suppress harmless noise from Streamlit's torch inspector + ChromaDB posthog client
logging.getLogger("streamlit.watcher.local_sources_watcher").setLevel(logging.ERROR)
logging.getLogger("chromadb.telemetry.product.posthog").setLevel(logging.ERROR)
import numpy as np
import pandas as pd
import streamlit as st
from agent.critic import critique, refine_query
from agent.planner import plan
from agent.router import route
from agent.tools import image_retrieve_and_reason
from config import AGENT_CONFIG, EMBEDDING_CONFIG, HOSTED, LLM_CONFIG, PATHS, RETRIEVAL_CONFIG
from ingestion.embedder import embed_query
from ingestion.indexer import fetch_embeddings
from llm.client_factory import get_llm
from retrieval.dense import Hit, dense_search
from retrieval.hybrid import reciprocal_rank_fusion
from retrieval.reranker import rerank
from retrieval.sparse import sparse_search
st.set_page_config(page_title="AdaptiveRAG β Underhood", page_icon="π¬", layout="wide")
# ββ LLM backend check β shown before anything else ββββββββββββββββββ
if not os.environ.get("GROQ_API_KEY"):
# Running without Groq β check if Ollama is reachable locally
try:
import requests as _req
_req.get("http://localhost:11434/api/tags", timeout=2).raise_for_status()
_ollama_ok = True
except Exception:
_ollama_ok = False
if not _ollama_ok:
st.error(
"**No LLM backend found.**\n\n"
"- **Running on Hugging Face?** Add your `GROQ_API_KEY` secret in "
"Space Settings β Variables and secrets. Get a free key at "
"[console.groq.com](https://console.groq.com).\n"
"- **Running locally?** Start Ollama: `ollama serve`"
)
st.stop()
# βββββββββββββββββββββββββββββ styling ββββββββββββββββββββββββββββββ
st.markdown(
"""
""",
unsafe_allow_html=True,
)
# βββββββββββββββββββββββββββββ helpers ββββββββββββββββββββββββββββββ
@st.cache_resource
def _llm():
return get_llm()
def _load_manifest() -> dict:
p = PATHS["manifest_path"]
return json.loads(p.read_text()) if p.exists() else {}
def phase_header(num: int, title: str, subtitle: str = "") -> None:
st.markdown(
f"STEP {num}"
f"{title}
{subtitle}"
f"
",
unsafe_allow_html=True,
)
def hits_to_df(hits: list[Hit], score_label: str = "score") -> pd.DataFrame:
rows = []
for h in hits:
title = h.metadata.get("title") or h.metadata.get("source_path", "?")
short = title.split(" (")[0]
if len(short) > 38:
short = short[:35] + "β¦"
label = f"{short} Β· p{h.metadata.get('page_start')} Β· {h.chunk_id.split('::')[-1]}"
rows.append({"chunk": label, score_label: float(h.score), "chunk_id": h.chunk_id})
return pd.DataFrame(rows)
def render_hits(hits: list[Hit], badge_class: str, label: str, max_chars: int = 220) -> None:
if not hits:
st.caption(f"_(no {label.lower()} hits)_")
return
for i, h in enumerate(hits, start=1):
meta = h.metadata
snippet = h.text[:max_chars].replace("\n", " ")
if len(h.text) > max_chars:
snippet += "β¦"
st.markdown(
f""
f"
"
f"{label} #{i}"
f"score {h.score:.3f} Β· "
f"{meta.get('title','?')} Β· p.{meta.get('page_start')}β{meta.get('page_end')} Β· "
f"{h.chunk_id}"
f"
{snippet}
"
f""
f"βοΈ Health score: {health_score:.0f} / 100"
f""
f"{label}
",
unsafe_allow_html=True,
)
if not healing_trace:
st.success("β
Answer passed all checks on first attempt β no healing needed.")
return
for attempt in healing_trace:
num = attempt.get("attempt", "?")
healthy = attempt.get("healthy", False)
icon = "β
" if healthy else "π§"
issues = attempt.get("issues", [])
label_str = "Healthy" if healthy else f"Issues: {', '.join(issues)}"
with st.expander(f"{icon} Attempt {num} β {label_str}", expanded=not healthy):
if healthy:
st.success("All checks passed β answer accepted.")
continue
c1, c2, c3 = st.columns(3)
c1.metric("Hallucination", "β οΈ yes" if "hallucination" in issues else "β
no")
c2.metric("Low chunk quality", "β οΈ yes" if "low_chunk_quality" in issues else "β
no")
c3.metric("Knowledge gap", "β οΈ yes" if "knowledge_gap" in issues else "β
no")
for action in attempt.get("actions", []):
atype = action.get("type", "")
detail = action.get("detail", "")
icons = {
"hallucination_fix": "π",
"chunk_expansion": "π",
"web_search": "π",
"query_rewrite": "βοΈ",
"regenerate": "π",
}
prefix = icons.get(atype, "β’")
st.markdown(
f"{atype}"
f"{prefix} {detail}
",
unsafe_allow_html=True,
)
if atype == "hallucination_fix":
for s in action.get("flagged_sentences", []):
st.caption(f" β³ Flagged: \"{s[:120]}β¦\"")
def visual_pipeline(query: str, enable_healing: bool = True) -> None:
llm = _llm()
# ββ Step 1: embed the question ββββββββββββββββββββββββββββββββ
phase_header(1, "Question encoding",
"Convert text β 384-dim dense vector via sentence-transformers (MiniLM-L6).")
t0 = time.time()
qv = embed_query(query)
render_embedding_card(query, qv, time.time() - t0)
# ββ Step 2: Self-RAG router ββββββββββββββββββββββββββββββββ
phase_header(2, "Self-RAG router",
"Decide whether to RETRIEVE, ANSWER_DIRECTLY, or CLARIFY before touching the index.")
t0 = time.time()
decision = route(query, llm=llm)
dt = time.time() - t0
pill_map = {"RETRIEVE": "pill-blue", "ANSWER_DIRECTLY": "pill-green", "CLARIFY": "pill-amber"}
pill = pill_map.get(decision["action"], "pill-grey")
st.markdown(
f"{decision['action']}"
f"{decision.get('reason','')}"
f""
f"router latency: {dt*1000:.0f} ms",
unsafe_allow_html=True,
)
if decision["action"] == "ANSWER_DIRECTLY":
st.markdown("### Direct answer (no retrieval)")
ans = llm.generate(prompt=query,
system="You are a helpful research assistant. Be concise.",
temperature=0.2)
st.markdown(ans)
return
if decision["action"] == "CLARIFY":
st.markdown("### Clarifying question")
ans = llm.generate(
prompt=("The user asked: " + query +
"\n\nIt is too ambiguous to answer well. Ask one short clarifying question."),
system="You are a helpful research assistant.",
temperature=0.2,
)
st.markdown(ans)
return
# ββ Iterations of plan β retrieve β answer β critique ββββββββ
accumulated: list[Hit] = []
current_query = query
for it in range(AGENT_CONFIG["max_iterations"]):
st.markdown(f"---\n## π Iteration {it + 1}")
if current_query != query:
st.info(f"Refined query β **{current_query}**")
# ββ Step 3: plan βββββββββββββββββββββββββββββββββββββ
phase_header(3, "Planner", "LLM decomposes the question into focused sub-queries.")
prior = ""
if accumulated:
titles = sorted({h.metadata.get("title", "?") for h in accumulated})
prior = "Already gathered passages from: " + ", ".join(titles)
t0 = time.time()
steps = plan(current_query, prior_summary=prior, llm=llm)
dt = time.time() - t0
st.caption(f"Generated {len(steps)} sub-quer{'y' if len(steps)==1 else 'ies'} in {dt*1000:.0f} ms")
for i, s in enumerate(steps, start=1):
st.markdown(
f""
f"
sub-query {i}"
f"
{s['query']}"
f"
"
f"rationale: {s.get('rationale','β')}
"
f"
"
f"#{i}"
f"score {h.score:.3f} Β· {title} Β· "
f"p.{meta.get('page_start')}β{meta.get('page_end')}"
f"
{h.text[:300]}{'β¦' if len(h.text)>300 else ''}"
f"