Spaces:
Sleeping
Sleeping
File size: 4,118 Bytes
4e316d6 4d20233 4e316d6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """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)
|