"""Debug panel components — retrieved chunks display, pipeline info, agent trace, and guardrail flags.""" from __future__ import annotations import streamlit as st def render_chunks_panel(chunks: list[tuple[str, float]]): """Show retrieved chunks with scores in expandable sections.""" st.subheader("Retrieved Chunks") if chunks: for i, (text, score) in enumerate(chunks): with st.expander( "Chunk {} — score: {:.4f}".format(i + 1, score), expanded=(i == 0), ): st.text(text[:500] + ("..." if len(text) > 500 else "")) else: st.caption("No chunks retrieved yet. Ask a question after uploading documents.") def render_pipeline_info( device: str, n_chunks: int, n_files: int, mode: str, alpha: float, chunk_size: int, top_k: int, ): """Render a summary table of the current pipeline configuration.""" st.divider() st.subheader("Pipeline Info") st.markdown(""" | Setting | Value | |---|---| | LLM | Qwen3-0.6B | | Device | `{}` | | Embedding | BGE-small-en-v1.5 (384d) | | Retrieval | {} (α={:.1f}) | | Chunks indexed | {} | | Chunk size | {} chars | | Top-k | {} | """.format(device, mode, alpha, n_chunks, chunk_size, top_k)) def render_agent_trace(steps: list): """Show the agent's reasoning trace — thoughts, tool calls, and observations.""" st.subheader("Agent Trace") if not steps: st.caption("No agent trace yet. Enable agent mode and ask a question.") return for i, step in enumerate(steps): with st.expander( "Step {} — {}".format(i + 1, step.tool_call.tool_name if step.tool_call else "Final"), expanded=(i == len(steps) - 1), ): if step.thought: st.markdown("**Thought:** {}".format(step.thought)) if step.tool_call: args = ", ".join( "{}={}".format(k, v) for k, v in step.tool_call.arguments.items() ) st.info("Tool: {}({})".format(step.tool_call.tool_name, args)) if step.observation: st.code(step.observation[:800], language=None) if step.error: st.error(step.error) def render_agent_trace_events(steps: list[dict]): """Render a persisted agent trace (list of step dicts from the chat stream). Each dict may carry: step, thought, tool, observation, guardrail, answer. Unlike ``render_agent_trace`` (which takes AgentStep objects), this renders the lightweight, serializable form the Streamlit chat flow accumulates so the steps survive a rerun. """ st.subheader("Agent Trace") if not steps: st.caption("Ask a question to see the agent's reasoning steps.") return for d in steps: has_answer = "answer" in d and "tool" not in d label = "Final answer" if has_answer else d.get("tool", "Step").split("(")[0] with st.expander("Step {} — {}".format(d.get("step", "?"), label), expanded=False): if d.get("thought"): st.markdown("**Thought:** {}".format(d["thought"])) if d.get("tool"): st.info(d["tool"]) if d.get("observation"): st.code(d["observation"][:800], language=None) if d.get("guardrail"): st.warning(d["guardrail"]) if d.get("answer"): st.markdown("**Answer:** {}".format(d["answer"])) def render_guardrail_flags(flags: list[dict]): """Show guardrail flags with severity-colored badges.""" st.subheader("Guardrail Checks") if not flags: st.caption("No guardrail flags.") return for flag in flags: severity = flag.get("severity", "warn") description = flag.get("description", "") check_name = flag.get("check_name", "unknown") label = "[{}] {}".format(check_name.upper(), description) if severity == "block": st.error(label) else: st.warning(label)