app.py

from __future__ import annotations

import tempfile
from pathlib import Path

import streamlit as st

from src.rag_cag_engine import (
    answer_adaptive,
    answer_with_cag,
    answer_with_rag,
    build_chunks,
    choose_route,
    estimate_query_complexity,
)


st.set_page_config(page_title="Adaptive RAG-CAG QA", page_icon="R", layout="wide")

st.title("Adaptive RAG-CAG Question Answering")
st.caption("A research demo based on CAG vs RAG for limited domain knowledge bases.")

with st.sidebar:
    st.header("Documents")
    uploaded_files = st.file_uploader(
        "Upload TXT, MD, or PDF files",
        type=["txt", "md", "pdf"],
        accept_multiple_files=True,
    )
    mode = st.radio("Answering mode", ["Adaptive", "RAG", "CAG", "Compare"], index=0)
    top_k = st.slider("RAG retrieved chunks", 1, 8, 4)


@st.cache_data(show_spinner=False)
def load_uploaded_documents(file_payloads):
    temp_paths = []
    with tempfile.TemporaryDirectory() as temp_dir:
        base = Path(temp_dir)
        for name, data in file_payloads:
            path = base / name
            path.write_bytes(data)
            temp_paths.append(path)
        return build_chunks(temp_paths)


if uploaded_files:
    payloads = [(file.name, file.getvalue()) for file in uploaded_files]
    try:
        chunks = load_uploaded_documents(payloads)
    except RuntimeError as exc:
        st.error(str(exc))
        st.stop()
else:
    sample_path = Path(__file__).parent / "sample_docs" / "rag_cag_notes.txt"
    chunks = build_chunks([sample_path])
    st.info("No file uploaded yet, so the app is using the included sample document.")

query = st.text_input("Ask a question from the documents", "When should CAG be preferred over RAG?")

left, middle, right = st.columns(3)
route, reason = choose_route(query, chunks)
left.metric("Chunks", len(chunks))
middle.metric("Query complexity", estimate_query_complexity(query).title())
right.metric("Adaptive route", route)
st.caption(reason)

if st.button("Run QA", type="primary"):
    if not query.strip():
        st.warning("Please enter a question.")
        st.stop()

    if mode == "RAG":
        result = answer_with_rag(query, chunks, top_k=top_k)
        results = [result]
    elif mode == "CAG":
        results = [answer_with_cag(query, chunks)]
    elif mode == "Adaptive":
        results = [answer_adaptive(query, chunks)]
    else:
        results = [answer_with_rag(query, chunks, top_k=top_k), answer_with_cag(query, chunks)]

    for result in results:
        st.subheader(result.mode)
        c1, c2 = st.columns(2)
        c1.metric("Latency", f"{result.latency_ms:.2f} ms")
        c2.metric("Confidence", f"{result.confidence:.3f}")
        st.write(result.answer)
        with st.expander("Grounded source snippets"):
            for chunk in result.sources:
                st.markdown(f"**{chunk.source} - chunk {chunk.index + 1}**")
                st.write(chunk.text[:900] + ("..." if len(chunk.text) > 900 else ""))