"""Professional Streamlit UI for the FastAPI Codebase Q&A RAG system. Run locally: streamlit run app.py Required environment variable: GROQ_API_KEY On Hugging Face Spaces, add GROQ_API_KEY under: Settings -> Variables and secrets """ import os import sys import time from pathlib import Path from textwrap import dedent from typing import Iterator from dotenv import load_dotenv load_dotenv(override=True) ROOT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(ROOT_DIR / "src")) import streamlit as st from ask import SYSTEM_PROMPT, build_prompt from retrieval import retrieve # --------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------- LLM_MODEL = os.environ.get("GROQ_MODEL", "openai/gpt-oss-120b") GITHUB_URL = "https://github.com/islam-mamedov/codebase-rag" SPACE_URL = ( "https://huggingface.co/spaces/" "islam-mamedov/fastapi-codebase-qa" ) EXAMPLES = [ { "title": "Custom status codes", "question": "How do I return a custom status code from an endpoint?", "icon": "↗", }, { "title": "Dependency injection", "question": "How does FastAPI's dependency injection system work?", "icon": "◫", }, { "title": "Find a class", "question": "Where is the APIRouter class defined?", "icon": "⌘", }, { "title": "Test refusal", "question": "How do I connect FastAPI to MongoDB?", "icon": "◇", }, ] MODE_LABELS = { "dense": "Dense retrieval", "dense_rw": "Dense + query rewriting", "hybrid": "Hybrid dense + BM25", } MODE_DESCRIPTIONS = { "dense": ( "Recommended. Best evaluation result with " "0.91 recall@5 and 0.71 MRR." ), "dense_rw": ( "Uses an LLM to expand the search query before retrieval. " "It matched dense recall but slightly reduced ranking quality." ), "hybrid": ( "Combines dense retrieval with BM25 keyword search. " "Included for comparison with the selected dense pipeline." ), } SOURCE_ICONS = { "code": "⌘", "doc": "▤", "issue": "◉", } # --------------------------------------------------------------------- # Page setup # --------------------------------------------------------------------- st.set_page_config( page_title="FastAPI Codebase Q&A", page_icon="⚡", layout="wide", initial_sidebar_state="expanded", ) def render_html(content: str) -> None: """Render trusted static HTML without Markdown parsing.""" st.html(dedent(content).strip()) # --------------------------------------------------------------------- # Styling # --------------------------------------------------------------------- render_html( """ """ ) # --------------------------------------------------------------------- # Index initialization # --------------------------------------------------------------------- @st.cache_resource( show_spinner=( "Preparing the search index. " "The first startup can take approximately three minutes." ) ) def ensure_index() -> bool: """Build the Chroma index when it does not already exist.""" import chromadb index_path = ROOT_DIR / "data" / "chroma" client = chromadb.PersistentClient(path=str(index_path)) try: collection = client.get_collection("chunks") if collection.count() > 0: return True except Exception: pass import index index.main() return True ensure_index() # --------------------------------------------------------------------- # Sidebar # --------------------------------------------------------------------- with st.sidebar: st.markdown("## ⚡ Codebase Q&A") st.caption( "Evaluation-driven retrieval over FastAPI source code, " "documentation and resolved issues." ) st.divider() st.markdown("### Retrieval") mode = st.selectbox( "Search strategy", options=["dense", "dense_rw", "hybrid"], format_func=lambda value: MODE_LABELS[value], help=( "Dense retrieval achieved the strongest overall " "evaluation result." ), ) st.caption(MODE_DESCRIPTIONS[mode]) st.divider() st.markdown("### Evaluation snapshot") st.caption("42-question hand-labelled benchmark") metric_col_1, metric_col_2 = st.columns(2) metric_col_1.metric("Recall@5", "0.91") metric_col_2.metric("MRR", "0.71") metric_col_3, metric_col_4 = st.columns(2) metric_col_3.metric("Correctness", "0.91") metric_col_4.metric("Refusals", "7 / 7") st.caption( "Every push runs unit tests and fails CI when " "recall@5 falls below 0.90." ) st.divider() st.markdown("### Project") st.link_button( "⌘ View source code", GITHUB_URL, use_container_width=True, ) st.link_button( "↗ Open live deployment", SPACE_URL, use_container_width=True, ) st.divider() if st.button( "Clear conversation", use_container_width=True, ): st.session_state.messages = [] st.rerun() # --------------------------------------------------------------------- # Hero # --------------------------------------------------------------------- render_html( """

Live evaluation-driven RAG system

Understand FastAPI through its actual codebase.

Ask technical questions and receive answers grounded in FastAPI source code, documentation and resolved GitHub issues. Each response includes retrieved evidence, while unsupported questions are refused instead of guessed.

1,352 indexed chunks AST-aware code retrieval Source-linked answers CI evaluation gate

""" ) # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- @st.cache_data(show_spinner=False) def cached_retrieve( question: str, retrieval_mode: str, ) -> list[dict]: """Retrieve the top five chunks for a question.""" return retrieve( question, k=5, mode=retrieval_mode, ) def stream_answer( question: str, hits: list[dict], ) -> Iterator[str]: """Stream a grounded answer from Groq.""" from groq import Groq api_key = os.environ.get("GROQ_API_KEY") if not api_key: st.error( "The GROQ_API_KEY environment variable is not configured." ) st.stop() client = Groq(api_key=api_key) try: stream = client.chat.completions.create( model=LLM_MODEL, messages=[ { "role": "system", "content": SYSTEM_PROMPT, }, { "role": "user", "content": build_prompt(question, hits), }, ], temperature=0.1, stream=True, ) for chunk in stream: delta = chunk.choices[0].delta.content if delta: yield delta except Exception as error: st.error( "The answer service encountered an error. " "Please try the question again." ) yield ( "\n\nThe answer could not be generated because the " "language-model request failed." ) print(f"[Groq error] {error}") def render_sources(hits: list[dict]) -> None: """Render retrieved evidence below an answer.""" render_html( f"""

Retrieved evidence {len(hits)} sources

""" ) for index_number, hit in enumerate(hits, start=1): metadata = hit.get("meta", {}) source_type = metadata.get("source_type", "doc") source_icon = SOURCE_ICONS.get(source_type, "▤") source_path = metadata.get("path", "Unknown source") source_symbol = metadata.get("symbol", "") source_url = metadata.get("url", "") source_score = hit.get("score") label = f"{source_icon} {index_number}. {source_path}" if source_symbol: label += f" · {source_symbol}" with st.expander(label): detail_col_1, detail_col_2 = st.columns([3, 1]) with detail_col_1: st.caption( f"Source type: {source_type.capitalize()}" ) with detail_col_2: if source_score is not None: st.caption( f"Similarity: {source_score:.3f}" ) if source_url: st.markdown( f"[Open original source on GitHub ↗]" f"({source_url})" ) code_language = ( "python" if source_type == "code" else "text" ) source_text = hit.get("text", "") st.code( source_text[:1600], language=code_language, wrap_lines=True, ) # --------------------------------------------------------------------- # Conversation state # --------------------------------------------------------------------- if "messages" not in st.session_state: st.session_state.messages = [] # --------------------------------------------------------------------- # Existing conversation # --------------------------------------------------------------------- for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"]) if message.get("timing"): st.caption(message["timing"]) if message.get("sources"): render_sources(message["sources"]) # --------------------------------------------------------------------- # Empty state # --------------------------------------------------------------------- pending_question = None if not st.session_state.messages: render_html( """

Start exploring

Ask a question or try an example

""" ) render_html( """

The assistant searches the indexed FastAPI corpus before answering. Questions about unrelated integrations should produce an honest refusal.

""" ) first_row = st.columns(2) second_row = st.columns(2) example_columns = [ first_row[0], first_row[1], second_row[0], second_row[1], ] for column, example in zip( example_columns, EXAMPLES, ): with column: button_label = ( f"{example['icon']} {example['title']}\n\n" f"{example['question']}" ) if st.button( button_label, key=f"example-{example['title']}", use_container_width=True, ): pending_question = example["question"] # --------------------------------------------------------------------- # Chat input # --------------------------------------------------------------------- typed_prompt = st.chat_input( "Ask about FastAPI's code, documentation or behaviour..." ) prompt = typed_prompt or pending_question # --------------------------------------------------------------------- # Generate response # --------------------------------------------------------------------- if prompt: st.session_state.messages.append( { "role": "user", "content": prompt, } ) with st.chat_message("user"): st.markdown(prompt) with st.chat_message("assistant"): retrieval_started = time.perf_counter() with st.spinner( "Searching 1,352 code, documentation and issue chunks..." ): hits = cached_retrieve( prompt, mode, ) retrieval_seconds = ( time.perf_counter() - retrieval_started ) generation_started = time.perf_counter() answer = st.write_stream( stream_answer( prompt, hits, ) ) generation_seconds = ( time.perf_counter() - generation_started ) timing = ( f"Retrieved in {retrieval_seconds:.2f}s" f" · Generated in {generation_seconds:.1f}s" f" · Strategy: {MODE_LABELS[mode]}" ) st.caption(timing) render_sources(hits) st.session_state.messages.append( { "role": "assistant", "content": answer, "sources": hits, "timing": timing, } ) # --------------------------------------------------------------------- # Footer # --------------------------------------------------------------------- render_html( """ """ )