blocks. # That caused the earlier broken empty panels. # # This CSS uses var(--background-color) and var(--text-color) to # seamlessly support Streamlit's native Light and Dark modes. # ============================================================ st.markdown( """ """, unsafe_allow_html=True, ) # ============================================================ # STEP 7: SESSION STATE INITIALIZATION # ============================================================ def initialize_session_state() -> None: """ Initialize all session-state keys used by the app. Streamlit reruns the full script after interactions. Missing keys can cause deployment crashes, especially on Hugging Face Spaces. """ default_values = { "chat_history": [], "last_result": None, "last_rebuild_result": None, "api_test_result": None, "auto_rebuild_error": None, "knowledge_base_ready": False, "upload_status_message": None, "delete_status_message": None, "last_uploaded_files": [], "knowledge_base_signature": "", "recommended_questions": [], "context_recommendation_preview": None, "show_sources": True, "show_debug_json": False, "system_log": [], "source_search_text": "", "demo_mode": False, "demo_mode_previous": None, "custom_source_files": [], "active_source_signature": "", "active_data_folder": "", } for key, default_value in default_values.items(): if key not in st.session_state: st.session_state[key] = default_value initialize_session_state() # ============================================================ # STEP 8: SMALL UI HELPERS # ============================================================ def add_system_log(message: str) -> None: """ Add a short system log message to the UI. """ timestamp = time.strftime("%H:%M:%S") st.session_state["system_log"].append(f"{timestamp} | {message}") st.session_state["system_log"] = st.session_state["system_log"][-10:] def show_toast(message: str, icon: str = "✅") -> None: """ Show toast notification where supported. """ try: st.toast(message, icon=icon) except Exception: st.success(message) # ============================================================ # STEP 9: RAG PIPELINE LOADER # ============================================================ @st.cache_resource(show_spinner=False) def get_pipeline() -> RAGPipeline: """ Load configuration and initialize the RAG pipeline. Cached because the embedding model can take time to load. """ config = load_config() validate_config(config) create_required_folders(config) return RAGPipeline(config=config) def load_pipeline_safely() -> Optional[RAGPipeline]: """ Load pipeline safely and show useful setup errors. """ try: return get_pipeline() except Exception as error: st.error("KnowFlow AI could not start because configuration is incomplete.") st.code(str(error)) st.info( "Check Hugging Face Secrets or local .env values. " "Required values include CLOUD_API_KEY, CLOUD_API_BASE_URL, and CLOUD_CHAT_MODEL." ) return None # ============================================================ # STEP 10: KNOWLEDGE BASE FILE HELPERS # ============================================================ def get_knowledge_base_files(data_folder: Path) -> List[Path]: """ Return supported files from the knowledge base folder. """ if not data_folder.exists(): return [] return sorted( [ path for path in data_folder.rglob("*") if path.is_file() and path.suffix.lower() in SUPPORTED_SOURCE_EXTENSIONS ] ) def is_demo_mode_enabled() -> bool: """ Read the global Demo Mode switch from session state. """ return bool(st.session_state.get("demo_mode", False)) def get_routed_data_folder() -> Path: """ Return the active document repository for the current source mode. """ return DEMO_DATA_FOLDER if is_demo_mode_enabled() else RAW_DATA_FOLDER def apply_data_folder_routing(pipeline: RAGPipeline) -> None: """ Point the pipeline config at the active source directory. """ active_data_folder = get_routed_data_folder() active_data_folder.mkdir(parents=True, exist_ok=True) pipeline.config.data_folder = active_data_folder st.session_state["active_data_folder"] = str(active_data_folder) def register_custom_source_files(saved_paths: List[Path], data_folder: Path) -> None: """ Remember files uploaded by the user for non-demo source indexing. """ demo_source_paths = { (data_folder / file_name).resolve() for file_name in DEMO_SOURCE_FILE_NAMES } existing_paths = { str(Path(path).resolve()) for path in st.session_state.get("custom_source_files", []) } for file_path in saved_paths: resolved_path = file_path.resolve() if resolved_path in demo_source_paths: continue existing_paths.add(str(resolved_path)) st.session_state["custom_source_files"] = sorted(existing_paths) def unregister_custom_source_file(file_path: Path) -> None: """ Remove a deleted file from the active custom source registry. """ resolved_path = str(file_path.resolve()) st.session_state["custom_source_files"] = [ path for path in st.session_state.get("custom_source_files", []) if str(Path(path).resolve()) != resolved_path ] def get_custom_source_files(data_folder: Path) -> List[Path]: """ Return only user-uploaded source files that still exist. """ resolved_data_folder = data_folder.resolve() demo_source_paths = { (data_folder / file_name).resolve() for file_name in DEMO_SOURCE_FILE_NAMES } custom_files = [] for stored_path in st.session_state.get("custom_source_files", []): try: candidate_path = Path(stored_path).resolve() except Exception: continue if not candidate_path.exists() or not candidate_path.is_file(): continue if candidate_path.suffix.lower() not in SUPPORTED_SOURCE_EXTENSIONS: continue if candidate_path in demo_source_paths: continue if resolved_data_folder not in candidate_path.parents: continue custom_files.append(candidate_path) unique_files = sorted(dict.fromkeys(custom_files), key=lambda path: str(path).lower()) st.session_state["custom_source_files"] = [str(path) for path in unique_files] return unique_files def get_active_knowledge_base_files(data_folder: Path) -> List[Path]: """ Return the sources that should be visible and indexed right now. Demo Mode reads only from data/sample. When Demo Mode is off, the app uses only user-uploaded files from the raw working folder. """ if is_demo_mode_enabled(): return get_knowledge_base_files(data_folder) return get_custom_source_files(data_folder) def get_file_signature(files: List[Path]) -> str: """ Create a fingerprint from file name, size, and modification time. """ signature_parts = [] for file_path in files: try: stat = file_path.stat() signature_parts.append( f"{file_path.resolve()}|{stat.st_size}|{int(stat.st_mtime)}" ) except FileNotFoundError: continue raw_signature = "::".join(signature_parts) return hashlib.md5(raw_signature.encode("utf-8")).hexdigest() def get_total_file_size_kb(files: List[Path]) -> float: """ Calculate total knowledge-base file size in KB. """ total_bytes = 0 for file_path in files: try: total_bytes += file_path.stat().st_size except FileNotFoundError: continue return total_bytes / 1024 def save_uploaded_files(uploaded_files: List[Any], data_folder: Path) -> List[Path]: """ Save uploaded files into the knowledge-base folder. """ saved_paths = [] data_folder.mkdir(parents=True, exist_ok=True) for uploaded_file in uploaded_files: safe_file_name = uploaded_file.name.replace("/", "_").replace("\\", "_") if safe_file_name in DEMO_SOURCE_FILE_NAMES: safe_file_name = f"uploaded_{safe_file_name}" output_path = data_folder / safe_file_name output_path.write_bytes(uploaded_file.getbuffer()) saved_paths.append(output_path) return saved_paths def delete_knowledge_base_file(file_path: Path, data_folder: Path) -> None: """ Safely delete one source file from data/raw/. """ resolved_file = file_path.resolve() resolved_data_folder = data_folder.resolve() if resolved_data_folder not in resolved_file.parents: raise ValueError("Unsafe delete blocked. File is outside the knowledge base folder.") if resolved_file.exists() and resolved_file.is_file(): resolved_file.unlink() # ============================================================ # STEP 11: RECOMMENDED QUESTIONS # ============================================================ def load_base_demo_questions() -> List[str]: """ Load base suggested questions from config/sample_questions.json. """ sample_questions_path = PROJECT_ROOT / "config" / "sample_questions.json" fallback_questions = [ "What is the refund policy?", "How long does standard shipping take?", "What does the warranty cover?", "Does the company sell customer data?", "How many days of annual leave do employees get?", "How much training support can employees receive?", "What is the remote work policy?", "What should employees do if their laptop is damaged?", "What is the company policy about quantum teleportation?", ] if not sample_questions_path.exists(): return fallback_questions try: data = json.loads(sample_questions_path.read_text(encoding="utf-8")) return data.get("demo_questions", fallback_questions) except Exception: return fallback_questions def build_dynamic_recommended_questions(files: List[Path]) -> List[str]: """ Build suggested questions based on current source files. """ if not files: return [] questions = [] if is_demo_mode_enabled() or any( file_path.name in DEMO_SOURCE_FILE_NAMES for file_path in files ): questions.extend(load_base_demo_questions()) for file_path in files: clean_name = file_path.stem.replace("_", " ").replace("-", " ").strip() if clean_name: questions.append(f"Summarize {clean_name}.") questions.append(f"What are the key points in {clean_name}?") questions.append(f"What should I know from {clean_name}?") cleaned_questions = [] seen = set() for question in questions: normalized = question.lower().strip() if normalized not in seen: cleaned_questions.append(question) seen.add(normalized) return cleaned_questions[:10] def refresh_recommended_questions_if_needed( data_folder: Path, force: bool = False, files: Optional[List[Path]] = None, ) -> None: """ Refresh suggested questions when source files change. """ if files is None: files = get_active_knowledge_base_files(data_folder) current_signature = get_file_signature(files) needs_refresh = ( force or current_signature != st.session_state.get("knowledge_base_signature") or (files and not bool(st.session_state.get("recommended_questions"))) or (not files and bool(st.session_state.get("recommended_questions"))) ) if needs_refresh: st.session_state["knowledge_base_signature"] = current_signature st.session_state["recommended_questions"] = build_dynamic_recommended_questions(files) if files: add_system_log("Suggested questions refreshed.") else: add_system_log("Suggested questions cleared.") def create_context_recommendation_preview(data_folder: Path) -> Optional[str]: """ Placeholder for future random-chunk recommendation generation. Current behavior: - Randomly previews text from uploaded text-like files. Future behavior: - Pull random indexed chunks from ChromaDB. - Ask the LLM to generate new recommended questions. """ files = get_active_knowledge_base_files(data_folder) text_like_files = [ path for path in files if path.suffix.lower() in {".txt", ".md", ".csv"} ] if not text_like_files: return None selected_file = random.choice(text_like_files) try: text = selected_file.read_text(encoding="utf-8", errors="ignore").strip() except Exception: return None if not text: return None start = random.randint(0, max(0, len(text) - 500)) preview = text[start:start + 500].strip() return ( f"Random context preview from `{selected_file.name}`:\n\n" f"{preview}" ) # ============================================================ # STEP 12: KNOWLEDGE BASE REBUILD # ============================================================ def get_vector_count_safely(pipeline: RAGPipeline) -> int: """ Return vector count without letting Chroma errors break the UI. """ try: return pipeline.vector_store.count() except Exception: return 0 def load_clean_documents_from_files( files: List[Path], project_root: Path, ) -> List[Document]: """ Load and clean only the files selected by the current source mode. """ documents = [] for file_path in files: try: document = load_single_document( path=file_path, project_root=project_root, ) cleaned_text = clean_text(document.text) if not cleaned_text: continue documents.append( Document( source=document.source, text=cleaned_text, file_type=document.file_type, character_count=len(cleaned_text), ) ) except Exception as error: add_system_log(f"Skipped source {file_path.name}: {error}") return documents def rebuild_knowledge_base( pipeline: RAGPipeline, files: Optional[List[Path]] = None, ) -> None: """ Rebuild vector database from the currently active source files. """ active_files = files if files is not None else get_active_knowledge_base_files( pipeline.config.data_folder ) with st.spinner("Rebuilding knowledge base..."): documents = load_clean_documents_from_files( files=active_files, project_root=pipeline.config.project_root, ) chunks = build_chunks_from_documents( documents=documents, chunk_size=pipeline.config.chunk_size, chunk_overlap=pipeline.config.chunk_overlap, ) pipeline.vector_store.reset_collection() if chunks: embeddings = pipeline.embedding_model.embed_texts( [chunk.text for chunk in chunks] ) pipeline.vector_store.add_chunks( chunks=chunks, embeddings=embeddings, ) rebuild_result = { "documents_loaded": len(documents), "chunks_created": len(chunks), "vectors_stored": pipeline.vector_store.count(), "active_sources": [file_path.name for file_path in active_files], "demo_mode": is_demo_mode_enabled(), "collection_name": pipeline.config.collection_name, "embedding_model": pipeline.config.embedding_model_name, } st.session_state["last_rebuild_result"] = rebuild_result st.session_state["knowledge_base_ready"] = rebuild_result.get("vectors_stored", 0) > 0 st.session_state["active_source_signature"] = get_file_signature(active_files) st.session_state["auto_rebuild_error"] = None refresh_recommended_questions_if_needed( data_folder=pipeline.config.data_folder, force=True, files=active_files, ) add_system_log( f"Knowledge base rebuilt with {rebuild_result.get('vectors_stored', 0)} chunks." ) def purge_active_rag_state(pipeline: RAGPipeline) -> None: """ Clear index, chat history, and generated workspace recommendations. """ try: pipeline.vector_store.reset_collection() st.session_state["auto_rebuild_error"] = None except Exception as error: st.session_state["auto_rebuild_error"] = str(error) st.session_state["chat_history"] = [] st.session_state["last_result"] = None st.session_state["last_rebuild_result"] = None st.session_state["recommended_questions"] = [] st.session_state["context_recommendation_preview"] = None st.session_state["knowledge_base_signature"] = "" st.session_state["active_source_signature"] = "" st.session_state["knowledge_base_ready"] = False def synchronize_demo_mode_state(pipeline: RAGPipeline) -> None: """ Keep source visibility, indexing, and blank-state behavior in sync. """ demo_mode_enabled = is_demo_mode_enabled() previous_demo_mode = st.session_state.get("demo_mode_previous") active_files = get_active_knowledge_base_files(pipeline.config.data_folder) active_signature = get_file_signature(active_files) source_signature_changed = ( active_signature != st.session_state.get("active_source_signature", "") ) if previous_demo_mode is True and not demo_mode_enabled: purge_active_rag_state(pipeline) st.session_state["demo_mode_previous"] = demo_mode_enabled add_system_log("Demo Mode disabled. Workspace state cleared.") return if demo_mode_enabled: if not active_files: purge_active_rag_state(pipeline) st.session_state["auto_rebuild_error"] = ( f"No demo sources found in {DEMO_DATA_FOLDER.relative_to(PROJECT_ROOT)}." ) st.session_state["demo_mode_previous"] = demo_mode_enabled add_system_log("Demo Mode source file is missing.") return needs_rebuild = ( previous_demo_mode is not True or source_signature_changed or get_vector_count_safely(pipeline) == 0 ) if needs_rebuild: st.session_state["chat_history"] = [] st.session_state["last_result"] = None st.session_state["context_recommendation_preview"] = None rebuild_knowledge_base(pipeline, files=active_files) show_toast("Demo source prepared.", "✅") st.session_state["demo_mode_previous"] = demo_mode_enabled return if not active_files: if ( previous_demo_mode is None or get_vector_count_safely(pipeline) > 0 or st.session_state.get("chat_history") or st.session_state.get("recommended_questions") ): purge_active_rag_state(pipeline) add_system_log("Empty source mode initialized.") st.session_state["demo_mode_previous"] = demo_mode_enabled return if source_signature_changed: st.session_state["active_source_signature"] = active_signature vector_count = get_vector_count_safely(pipeline) if vector_count > 0: st.session_state["knowledge_base_ready"] = True refresh_recommended_questions_if_needed( data_folder=pipeline.config.data_folder, files=active_files, ) else: st.session_state["knowledge_base_ready"] = False st.session_state["recommended_questions"] = [] st.session_state["demo_mode_previous"] = demo_mode_enabled def build_debug_markdown(result: Dict[str, Any]) -> str: """ Build short debug information to show inside the chat when enabled. """ return f""" --- ### Debug Summary - Provider: `{result.get("provider")}` - Model: `{result.get("model")}` - Embedding model: `{result.get("embedding_model")}` - Prompt version: `{result.get("prompt_template_version")}` - Retrieved chunks: `{result.get("top_k")}` - Total time: `{result.get("total_elapsed_seconds")} seconds` """ def render_assistant_metadata(metadata: Dict[str, str]) -> None: """ Render assistant metadata using native Streamlit columns. """ metadata_items = [ ("Model", metadata.get("model", "N/A")), ("Retrieved chunks", metadata.get("retrieved_chunks", "0")), ("Processing time", metadata.get("processing_time", "N/A")), ("Saved output", metadata.get("saved_output", "N/A")), ] metadata_columns = st.columns([1.15, 1.0, 1.0, 1.7], gap="small") for column, (label, value) in zip(metadata_columns, metadata_items): with column: st.caption(label) st.markdown(f"`{value}`") def render_retrieved_references(retrieved_chunks: List[Dict[str, Any]]) -> None: """ Render retrieved chunks inside a native expandable section. """ with st.expander("📌 Show retrieved references", expanded=False): if not retrieved_chunks: st.caption("No retrieved references available.") return for index, item in enumerate(retrieved_chunks): source = item.get("source", "unknown") rank = item.get("rank", "N/A") chunk_index = item.get("chunk_index", "N/A") distance = item.get("distance", 0.0) text = str(item.get("text", "")).strip() preview = text[:700] if len(text) > 700: preview = f"{preview}..." st.markdown(f"**Source:** `{source}`") st.caption( f"Rank {rank} | Chunk {chunk_index} | Distance {float(distance):.4f}" ) st.markdown(preview) if index < len(retrieved_chunks) - 1: st.divider() # ============================================================ # STEP 14: SOURCES PANEL # ============================================================ def display_sources_panel(pipeline: RAGPipeline) -> None: """ Left panel: - Upload sources - Rebuild source index - View source files - Delete source files """ config = pipeline.config files = get_active_knowledge_base_files(config.data_folder) with st.container(border=True, key="sources_panel"): st.subheader("📚 Sources") st.markdown( "

Add files to your knowledge base. Answers are grounded in these sources.

", unsafe_allow_html=True, ) st.markdown( f"""

Files {len(files)}

Size {get_total_file_size_kb(files):.1f}KB

""", unsafe_allow_html=True, ) st.markdown("

", unsafe_allow_html=True) uploaded_files = st.file_uploader( "Upload sources", type=["txt", "md", "csv", "pdf"], accept_multiple_files=True, help="Supported files: TXT, Markdown, CSV, PDF", ) upload_col, rebuild_col = st.columns(2) with upload_col: if st.button("Add sources", use_container_width=True): if not uploaded_files: st.warning("Upload at least one file.") else: try: upload_folder = RAW_DATA_FOLDER saved_paths = save_uploaded_files( uploaded_files=uploaded_files, data_folder=upload_folder, ) register_custom_source_files( saved_paths=saved_paths, data_folder=upload_folder, ) st.session_state["last_uploaded_files"] = [ str(path.relative_to(PROJECT_ROOT)) for path in saved_paths ] rebuild_knowledge_base(pipeline) if is_demo_mode_enabled(): show_toast( ( f"Saved {len(saved_paths)} custom source(s). " "Demo Mode keeps the demo source indexed." ), "✅", ) else: show_toast(f"Added and indexed {len(saved_paths)} source(s).", "✅") add_system_log(f"Added {len(saved_paths)} source file(s).") st.rerun() except Exception as error: st.error("Upload failed.") st.code(str(error)) with rebuild_col: if st.button("Rebuild", use_container_width=True): try: rebuild_knowledge_base(pipeline) show_toast("Sources rebuilt.", "✅") st.rerun() except Exception as error: st.error("Rebuild failed.") st.code(str(error)) st.markdown("

", unsafe_allow_html=True) with st.container(height=320, key="sources_list_scroller"): st.markdown( f"""

Source library

{len(files)} total

""", unsafe_allow_html=True, ) st.session_state["source_search_text"] = st.text_input( "Search sources", value=st.session_state.get("source_search_text", ""), placeholder="Filter by file name...", label_visibility="collapsed", ) search_text = st.session_state.get("source_search_text", "").lower().strip() if search_text: files_to_show = [ file_path for file_path in files if search_text in file_path.name.lower() ] else: files_to_show = files if not files_to_show: if search_text: st.info("No matching sources found.") else: for index, file_path in enumerate(files_to_show): relative_path = file_path.relative_to(PROJECT_ROOT) file_size_kb = file_path.stat().st_size / 1024 modified_date = time.strftime( "%Y-%m-%d", time.localtime(file_path.stat().st_mtime), ) file_type = file_path.suffix.lower().replace(".", "") or "file" chip_class = ( file_type if file_type in {"pdf", "csv", "md", "txt"} else "txt" ) st.markdown( f"""

{file_path.name}

{relative_path}

{file_type.upper()} • {file_size_kb:.1f} KB • Updated {modified_date}

""", unsafe_allow_html=True, ) delete_clicked = False if is_demo_mode_enabled(): st.caption("Built-in demo source") else: delete_clicked = st.button( "Delete source", key=f"delete_source_{index}_{file_path.name}", ) if delete_clicked: try: delete_knowledge_base_file(file_path, config.data_folder) unregister_custom_source_file(file_path) rebuild_knowledge_base(pipeline) show_toast(f"Deleted {file_path.name}.", "🗑️") add_system_log(f"Deleted source: {file_path.name}") st.rerun() except Exception as error: st.error("Delete failed.") st.code(str(error)) if st.session_state.get("last_uploaded_files"): with st.expander("Last uploaded", expanded=False): for uploaded_path in st.session_state["last_uploaded_files"]: st.write(f"✅ `{uploaded_path}`") # ============================================================ # STEP 15: CHAT PANEL # ============================================================ def display_chat_history() -> None: """ Display chat messages using native Streamlit chat components. """ for message in st.session_state.get("chat_history", []): role = message.get("role", "assistant") with st.chat_message(role): if role == "user": st.markdown(message.get("content", "")) continue if message.get("message_type") == "assistant_result": st.markdown(message.get("answer", "")) with st.container(): render_assistant_metadata(message.get("metadata", {})) if message.get("references_enabled", True): render_retrieved_references(message.get("retrieved_chunks", [])) if message.get("debug_markdown"): with st.expander("Debug summary", expanded=False): st.markdown(message["debug_markdown"]) continue st.markdown(message.get("content", "")) def ask_question(pipeline: RAGPipeline, question: str) -> None: """ Ask a question through the RAG pipeline and save a structured assistant response for native chat rendering. """ question = question.strip() if not question: return active_files = get_active_knowledge_base_files(pipeline.config.data_folder) vector_count = get_vector_count_safely(pipeline) if active_files else 0 if vector_count == 0: st.session_state["chat_history"].append( { "role": "assistant", "content": ( "The knowledge base is not ready yet. " "Please add a source or rebuild the knowledge base first." ), } ) return st.session_state["chat_history"].append( { "role": "user", "content": question, } ) try: with st.spinner("Searching sources and generating answer..."): result = pipeline.ask(question, top_k=pipeline.config.top_k) saved_path = pipeline.save_result(result) except Exception as error: st.session_state["chat_history"].append( { "role": "assistant", "content": f"Something went wrong while generating the answer.\n\n`{error}`", } ) add_system_log(f"Answer failed: {question[:70]}") return saved_output_display = saved_path try: saved_output_display = str(Path(saved_path).resolve().relative_to(PROJECT_ROOT)) except Exception: saved_output_display = str(Path(saved_path).name) debug_markdown = "" if st.session_state.get("show_debug_json", False): debug_markdown = build_debug_markdown(result) st.session_state["chat_history"].append( { "role": "assistant", "message_type": "assistant_result", "answer": result.get("answer", ""), "metadata": { "model": result.get("model", "N/A"), "retrieved_chunks": str(len(result.get("retrieved_chunks", []))), "processing_time": f"{result.get('total_elapsed_seconds', 'N/A')}s", "saved_output": saved_output_display, }, "retrieved_chunks": result.get("retrieved_chunks", []), "references_enabled": bool(st.session_state.get("show_sources", True)), "debug_markdown": debug_markdown, } ) st.session_state["last_result"] = result add_system_log(f"Answered: {question[:70]}") def display_suggested_questions(pipeline: RAGPipeline) -> None: """ Display dynamic question recommendations inside the chat window. """ active_files = get_active_knowledge_base_files(pipeline.config.data_folder) if not active_files: st.session_state["recommended_questions"] = [] return refresh_recommended_questions_if_needed( data_folder=pipeline.config.data_folder, files=active_files, ) questions = st.session_state.get("recommended_questions", []) if not questions: return with st.container(height=210, key="suggested_questions_scroller"): st.markdown( """

Suggested questions

Start with one of these prompts or scroll for more.

""", unsafe_allow_html=True, ) cols = st.columns(2, gap="small") for index, question in enumerate(questions[:10]): with cols[index % 2]: if st.button( question, key=f"suggested_question_{index}", use_container_width=True, ): ask_question(pipeline, question) st.rerun() def display_question_input_form(vector_count: int) -> Optional[str]: """ Display the anchored chat input at the base of the workspace. """ return st.chat_input( "Ask anything from your uploaded sources...", disabled=vector_count == 0, key="workspace_chat_input", ) def display_demo_mode_awareness_note() -> None: """ Explain how the current source mode controls document routing. """ if is_demo_mode_enabled(): title = "Demo Mode is on" message = ( "The conversation is grounded exclusively in files from " "data/sample. Uploaded raw sources are saved separately and are " "not indexed until Demo Mode is turned off." ) else: title = "Demo Mode is off" message = ( "Sample content has been removed from the active workspace. " "The app now targets data/raw; upload custom files and rebuild " "to populate the source library, suggested prompts, and chat index." ) st.markdown( f"""

{title} {message}

""", unsafe_allow_html=True, ) def display_chat_panel(pipeline: RAGPipeline, vector_count: int) -> None: """ Central chat workspace with a bordered conversation area, containerized history, and an input anchored at the base. """ active_files = get_active_knowledge_base_files(pipeline.config.data_folder) with st.container(border=True, key="chat_workspace"): st.subheader("💬 Conversation Workspace") st.caption( "The full conversation stays inside this workspace while the input remains anchored at the bottom for a modern chat flow." ) display_demo_mode_awareness_note() history_height = 430 if st.session_state.get("chat_history") else 340 with st.container(height=history_height, key="chat_history"): if not st.session_state.get("chat_history"): if active_files: with st.container(key="chat_welcome_state"): st.markdown("#### Start chatting with your documents") st.caption( "Ask a question below or start with one of the suggested prompts. The conversation history will stay contained in this bordered workspace." ) if vector_count == 0: st.info("Add or rebuild sources before asking questions.") else: display_suggested_questions(pipeline) else: display_chat_history() with st.container(key="chat_input_shelf"): prompt = display_question_input_form(vector_count) if prompt and prompt.strip(): ask_question(pipeline, prompt) st.rerun() # ============================================================ # STEP 16: RIGHT GUIDE PANEL # ============================================================ def display_runtime_settings(pipeline: RAGPipeline) -> None: """ Runtime RAG settings. """ config = pipeline.config with st.expander("⚙️ Runtime settings", expanded=False): config.top_k = st.slider( "Top-K retrieved chunks", min_value=1, max_value=10, value=int(config.top_k), step=1, ) config.cloud_temperature = st.slider( "LLM temperature", min_value=0.0, max_value=1.0, value=float(config.cloud_temperature), step=0.05, ) config.cloud_max_completion_tokens = st.slider( "Max answer tokens", min_value=100, max_value=2000, value=int(config.cloud_max_completion_tokens), step=100, ) st.divider() config.chunk_size = st.slider( "Chunk size", min_value=300, max_value=2000, value=int(config.chunk_size), step=100, help="Used during rebuild.", ) config.chunk_overlap = st.slider( "Chunk overlap", min_value=0, max_value=400, value=int(config.chunk_overlap), step=20, help="Used during rebuild.", ) st.session_state["show_sources"] = st.toggle( "Attach references to answers", value=bool(st.session_state.get("show_sources", True)), help="References are hidden by default and visible only when clicked.", ) st.session_state["show_debug_json"] = st.toggle( "Show debug summary in chat", value=bool(st.session_state.get("show_debug_json", False)), ) def display_system_tools(pipeline: RAGPipeline) -> None: """ API test, rebuild summary, and logs. """ with st.expander("🛠 System tools", expanded=False): api_col, clear_col = st.columns(2) with api_col: if st.button("Test API", use_container_width=True): with st.spinner("Testing API..."): try: st.session_state["api_test_result"] = pipeline.llm_client.test_connection() add_system_log("API test successful.") show_toast("API connection works.", "✅") except Exception as error: st.session_state["api_test_result"] = str(error) add_system_log("API test failed.") show_toast("API connection failed.", "⚠️") with clear_col: if st.button("Clear chat", use_container_width=True): st.session_state["chat_history"] = [] st.session_state["last_result"] = None add_system_log("Chat cleared.") st.rerun() if st.session_state.get("api_test_result"): st.caption("API result") st.code(str(st.session_state["api_test_result"])[:600]) if st.session_state.get("last_rebuild_result"): st.caption("Last rebuild") st.json(st.session_state["last_rebuild_result"]) st.caption("System logs") if st.session_state.get("system_log"): for log_item in st.session_state["system_log"]: st.write(log_item) else: st.write("No logs yet.") def display_future_recommendation_panel(pipeline: RAGPipeline) -> None: """ Placeholder for future context-aware recommendations. """ st.markdown( """

Future recommendation feature
Later, this area can sample random chunks from uploaded sources and ask the LLM to generate context-based suggested questions.

""", unsafe_allow_html=True, ) if st.button("Preview random source context", use_container_width=True): st.session_state["context_recommendation_preview"] = ( create_context_recommendation_preview(pipeline.config.data_folder) ) add_system_log("Random context preview generated.") if st.session_state.get("context_recommendation_preview"): st.info(st.session_state["context_recommendation_preview"]) def display_guide_panel(pipeline: RAGPipeline, vector_count: int) -> None: """ Right panel: - Notebook guide - Settings - System tools - Future recommendation placeholder """ config = pipeline.config files = get_active_knowledge_base_files(config.data_folder) with st.container(border=True, key="guide_panel"): st.subheader("🧭 Settings") st.markdown( "

Manage runtime settings and review system state. Answer references are shown inside the chat.

", unsafe_allow_html=True, ) guide_col1, guide_col2 = st.columns(2) with guide_col1: st.metric("Sources", len(files)) with guide_col2: st.metric("Chunks", vector_count) st.info(f"Model: {config.cloud_chat_model}") st.caption(f"Embedding: {config.embedding_model_name}") with st.container(height=400, key="guide_tools_scroller"): display_runtime_settings(pipeline) display_system_tools(pipeline) # ============================================================ # STEP 17: TOP HEADER # ============================================================ def display_top_header(pipeline: RAGPipeline, vector_count: int) -> None: """ Display clean app header toolbar. """ demo_mode_enabled = is_demo_mode_enabled() mode_text = "Sample" if demo_mode_enabled else "Raw" status_text = "Sources ready" if vector_count > 0 else "Sources not indexed" status_dot_class = "top-header-status-dot-ready" if vector_count > 0 else "" with st.container(key="top_header"): brand_col, metrics_col, toggle_col = st.columns([2.3, 1.7, 1.0], gap="small") with brand_col: st.markdown( """ """, unsafe_allow_html=True, ) with metrics_col: st.markdown( f""" """, unsafe_allow_html=True, ) with toggle_col: with st.container(key="demo_mode_header_control"): st.toggle( "Demo Mode", key="demo_mode", help="Switch between the sample demo repository and the raw upload repository.", ) # ============================================================ # STEP 18: MAIN APP # ============================================================ def main() -> None: """ Main Streamlit app. """ initialize_session_state() pipeline = load_pipeline_safely() if pipeline is None: return apply_data_folder_routing(pipeline) config = pipeline.config synchronize_demo_mode_state(pipeline) active_files = get_active_knowledge_base_files(config.data_folder) vector_count = get_vector_count_safely(pipeline) if active_files else 0 display_top_header(pipeline, vector_count) with st.container(key="app_shell"): sources_col, chat_col, guide_col = st.columns( [0.9, 1.85, 1.0], gap="large", ) with sources_col: display_sources_panel(pipeline) with chat_col: display_chat_panel(pipeline, vector_count) with guide_col: display_guide_panel(pipeline, vector_count) # ============================================================ # STEP 19: ENTRY POINT # ============================================================ if __name__ == "__main__": main()