Spaces:
Sleeping
Sleeping
| """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) | |