"""Generate a standalone HTML visualization for an agent run."""
from __future__ import annotations
import html
import json
from pathlib import Path
from typing import Any
import markdown as _markdown_lib
from supply_agent.logging.parser import load_run_events, summarize_run
_MARKDOWN_EXTENSIONS = ["fenced_code", "tables", "sane_lists"]
def _render_markdown(text: Any, *, soft_breaks: bool = False, boxed: bool = True) -> str:
"""Render text as Markdown; ``boxed`` wraps it in a bordered, scrollable box."""
if not text:
return ""
extensions = [*_MARKDOWN_EXTENSIONS, "nl2br"] if soft_breaks else _MARKDOWN_EXTENSIONS
body = _markdown_lib.markdown(str(text), extensions=extensions)
if not boxed:
return f'
{body}
'
return f''
def _esc(text: Any) -> str:
if text is None:
return ""
return html.escape(str(text))
def _pretty(obj: Any) -> str:
if obj is None:
return ""
if isinstance(obj, str):
try:
obj = json.loads(obj)
except (json.JSONDecodeError, TypeError):
return obj
return json.dumps(obj, ensure_ascii=False, indent=2)
def _pre(text: Any, *, klass: str = "code") -> str:
if text is None:
content = ""
else:
content = _pretty(text)
return f'{_esc(content)}'
def _message_fingerprint(msg: dict[str, Any]) -> str:
return json.dumps(msg, ensure_ascii=False, sort_keys=True, default=str)
def _common_prefix_len(a: list[dict[str, Any]], b: list[dict[str, Any]]) -> int:
n = 0
for left, right in zip(a, b):
if _message_fingerprint(left) != _message_fingerprint(right):
break
n += 1
return n
def _split_history_and_latest(
messages: list[dict[str, Any]],
prev_messages: list[dict[str, Any]] | None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""
Split request messages into (history, latest).
``latest`` is only what is newly added since the previous LLM input;
``history`` is everything else (hidden by default in the UI).
"""
if not messages:
return [], []
if not prev_messages:
if len(messages) > 1 and messages[0].get("role") == "system":
return messages[:1], messages[1:]
return [], list(messages)
curr_has_sys = messages[0].get("role") == "system"
prev_has_sys = prev_messages[0].get("role") == "system"
curr_sys = messages[0] if curr_has_sys else None
prev_sys = prev_messages[0] if prev_has_sys else None
curr_rest = messages[1:] if curr_has_sys else messages
prev_rest = prev_messages[1:] if prev_has_sys else prev_messages
k = _common_prefix_len(curr_rest, prev_rest)
history_rest = curr_rest[:k]
latest = list(curr_rest[k:])
history: list[dict[str, Any]] = []
if curr_sys is not None:
history.append(curr_sys)
history.extend(history_rest)
# Surface an updated system prompt in the latest section
if (
curr_sys is not None
and prev_sys is not None
and _message_fingerprint(curr_sys) != _message_fingerprint(prev_sys)
):
latest = [curr_sys, *latest]
history = list(history_rest)
if not latest:
return messages[:-1], messages[-1:]
return history, latest
def _role_badge(role: str) -> str:
return f'{_esc(role)}'
def _render_messages(
messages: list[dict[str, Any]] | None,
*,
default_open: bool | None = None,
) -> str:
if not messages:
return '无消息
'
parts: list[str] = []
for msg in messages:
role = msg.get("role", "?")
body_parts: list[str] = []
if msg.get("content"):
body_parts.append(_pre(msg["content"], klass="code prose"))
if role in ("user", "tool"):
open_attr = ""
elif default_open is None:
open_attr = "open" if role != "system" else ""
else:
open_attr = "open" if default_open else ""
if role == "tool":
tool_name = msg.get("name") or "?"
summary_html = (
f'{_role_badge(role)} {_esc(tool_name)}'
)
else:
preview = _preview(msg.get("content"))
if not preview and msg.get("tool_calls"):
names = []
for tc in msg["tool_calls"]:
name, _, _ = _normalize_tool_call(tc)
names.append(name)
preview = "调用: " + ", ".join(names)
summary_html = (
f'{_role_badge(role)} {_esc(preview)}'
)
parts.append(
f"""
{summary_html}
{"".join(body_parts)}
"""
)
return '' + "".join(parts) + "
"
def _preview(text: Any, limit: int = 80) -> str:
if not text:
return ""
s = " ".join(str(text).split())
return s if len(s) <= limit else s[: limit - 1] + "…"
def _tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
if not tools:
return []
names: list[str] = []
for t in tools:
fn = t.get("function") or t
names.append(fn.get("name", "?"))
return names
def _extract_system_prompt_and_tools(
events: list[dict[str, Any]],
) -> tuple[str | None, list[str]]:
"""Pull the system prompt and tool list from the first llm_input event."""
for ev in events:
if ev.get("event") != "llm_input":
continue
data = ev.get("data") or {}
messages = data.get("messages") or []
sys_msg = next((m for m in messages if m.get("role") == "system"), None)
content = sys_msg.get("content") if sys_msg else None
return content, _tool_names(data.get("tools"))
return None, []
def _render_llm_input_section(
data: dict[str, Any],
*,
prev_messages: list[dict[str, Any]] | None = None,
) -> tuple[str, list[dict[str, Any]]]:
"""Render the LLM-input body; returns (html, messages) so callers can track history."""
messages = list(data.get("messages") or [])
_history, latest = _split_history_and_latest(messages, prev_messages)
latest_count = len(latest)
html_out = f"""
{_render_messages(latest, default_open=True)}
"""
return html_out, messages
def _render_reasoning(reasoning: Any) -> str:
if not reasoning:
return ""
return f"""
思考过程
{_render_markdown(reasoning, soft_breaks=True, boxed=False)}
"""
def _parse_tool_arguments(arguments: Any) -> Any:
if isinstance(arguments, str):
try:
return json.loads(arguments)
except (json.JSONDecodeError, TypeError):
return arguments
return arguments
def _normalize_tool_call(tc: dict[str, Any]) -> tuple[str, Any, str]:
"""Return (name, parsed_args, id) from flat or OpenAI function-wrapped tool_call."""
name = tc.get("name") or (tc.get("function") or {}).get("name") or "?"
raw_args = tc.get("arguments")
if raw_args is None and isinstance(tc.get("function"), dict):
raw_args = tc["function"].get("arguments")
return name, _parse_tool_arguments(raw_args), str(tc.get("id") or "")
def _render_output_tool_calls(tool_calls: list[dict[str, Any]]) -> str:
"""Show each tool the model decided to call, collapsed by default; expand for args."""
if not tool_calls:
return ""
parts: list[str] = []
for tc in tool_calls:
name, args, _ = _normalize_tool_call(tc)
parts.append(
f"""
tool {_esc(name)}
{_pre(args)}
"""
)
return '' + "".join(parts) + "
"
def _render_llm_output_section(data: dict[str, Any]) -> str:
tool_calls = data.get("tool_calls") or []
reasoning = data.get("reasoning")
content = data.get("content")
usage = data.get("usage")
usage_html = ""
if usage:
usage_html = f"""
prompt: {_esc(usage.get("prompt_tokens", "—"))}
completion: {_esc(usage.get("completion_tokens", "—"))}
total: {_esc(usage.get("total_tokens", "—"))}
cost: {_esc(usage.get("cost", "—"))}
"""
content_html = (
f"模型输出文本
{_render_markdown(content, soft_breaks=True)}"
if content
else ""
)
tags: list[str] = []
if reasoning:
tags.append('有思考')
if tool_calls:
tags.append(f'{len(tool_calls)} tool calls')
return f"""
{_render_reasoning(reasoning)}
{content_html}
{_render_output_tool_calls(tool_calls)}
{usage_html}
"""
def _render_step_card(
step_no: int,
iteration: Any,
title: str,
inner_html: str,
) -> str:
return f"""
{inner_html}
"""
def _render_skill(data: dict[str, Any], step_no: int, iteration: Any) -> str:
return f"""
"""
_HIDDEN_EVENT_TYPES = ("run_start", "run_end", "tool_call")
def _group_visible_events(
events: list[dict[str, Any]],
) -> list[list[dict[str, Any]]]:
"""
Group renderable events into steps.
A consecutive ``llm_input`` immediately followed by ``llm_output`` forms a
single step (one round-trip); everything else is its own step.
"""
visible = [ev for ev in events if ev.get("event") not in _HIDDEN_EVENT_TYPES]
groups: list[list[dict[str, Any]]] = []
i = 0
n = len(visible)
while i < n:
ev = visible[i]
if (
ev.get("event") == "llm_input"
and i + 1 < n
and visible[i + 1].get("event") == "llm_output"
):
groups.append([ev, visible[i + 1]])
i += 2
else:
groups.append([ev])
i += 1
return groups
def _event_iteration(ev: dict[str, Any]) -> Any:
data = ev.get("data") or {}
return ev.get("iteration", data.get("iteration", "—"))
def _render_timeline(events: list[dict[str, Any]]) -> str:
parts: list[str] = []
prev_llm_messages: list[dict[str, Any]] | None = None
for step_no, group in enumerate(_group_visible_events(events), start=1):
if len(group) == 2:
input_ev, output_ev = group
input_data = input_ev.get("data") or {}
output_data = output_ev.get("data") or {}
input_html, messages = _render_llm_input_section(
input_data, prev_messages=prev_llm_messages
)
prev_llm_messages = messages
output_html = _render_llm_output_section(output_data)
parts.append(
_render_step_card(
step_no,
_event_iteration(input_ev),
"LLM 输入 · 输出",
input_html + output_html,
)
)
continue
ev = group[0]
etype = ev.get("event")
data = ev.get("data") or {}
iteration = _event_iteration(ev)
if etype == "llm_input":
input_html, messages = _render_llm_input_section(
data, prev_messages=prev_llm_messages
)
prev_llm_messages = messages
parts.append(_render_step_card(step_no, iteration, "LLM 输入", input_html))
elif etype == "llm_output":
output_html = _render_llm_output_section(data)
parts.append(_render_step_card(step_no, iteration, "LLM 输出", output_html))
elif etype == "skill_loaded":
parts.append(_render_skill(data, step_no, iteration))
else:
parts.append(
f"""
{_pre(data)}
"""
)
return "\n".join(parts)
def _nav_items(events: list[dict[str, Any]]) -> str:
items: list[str] = []
labels = {
"llm_input": "LLM 输入",
"llm_output": "LLM 输出",
"skill_loaded": "技能",
}
for step_no, group in enumerate(_group_visible_events(events), start=1):
if len(group) == 2:
ev = group[0]
label = "LLM 输入 · 输出"
nav_class = "nav-step-pair"
detail = f" · iter {_event_iteration(ev)}"
else:
ev = group[0]
etype = ev.get("event")
label = labels.get(etype, etype or "?")
nav_class = f"nav-{_esc(etype or '')}"
detail = ""
if etype in ("llm_input", "llm_output"):
detail = f" · iter {_event_iteration(ev)}"
items.append(
f''
f'{step_no}{label}{detail}'
)
return "\n".join(items)
_CSS = """
:root {
--bg: #f4f6f9;
--bg-elev: #ffffff;
--bg-card: #ffffff;
--border: #d8dee8;
--text: #1a2332;
--muted: #6b7c93;
--accent: #2563eb;
--input: #2563eb;
--output: #059669;
--skill: #7c3aed;
--code-bg: #f8fafc;
--mono: "JetBrains Mono", "SF Mono", "Fira Code", ui-monospace, monospace;
--sans: "IBM Plex Sans", "Segoe UI", system-ui, sans-serif;
}
* { box-sizing: border-box; }
html { scroll-behavior: smooth; }
body {
margin: 0;
font-family: var(--sans);
background: var(--bg);
color: var(--text);
line-height: 1.55;
}
a { color: var(--accent); text-decoration: none; }
a:hover { text-decoration: underline; }
.layout {
display: grid;
grid-template-columns: 260px 1fr;
min-height: 100vh;
}
.sidebar {
position: sticky;
top: 0;
height: 100vh;
overflow: auto;
background: var(--bg-elev);
border-right: 1px solid var(--border);
padding: 1.25rem 1rem;
box-shadow: 1px 0 0 rgba(26, 35, 50, 0.02);
}
.sidebar h1 {
font-size: 0.95rem;
margin: 0 0 0.25rem;
letter-spacing: 0.02em;
color: var(--text);
}
.sidebar .run-id {
font-family: var(--mono);
font-size: 0.7rem;
color: var(--muted);
word-break: break-all;
margin-bottom: 1rem;
}
.nav-item {
display: flex;
align-items: center;
gap: 0.5rem;
padding: 0.4rem 0.55rem;
border-radius: 6px;
color: var(--text);
font-size: 0.8rem;
margin-bottom: 2px;
}
.nav-item:hover { background: #eef2f7; text-decoration: none; }
.nav-seq {
font-family: var(--mono);
font-size: 0.7rem;
color: var(--muted);
min-width: 1.4rem;
}
.nav-llm_input { border-left: 3px solid var(--input); }
.nav-llm_output { border-left: 3px solid var(--output); }
.nav-step-pair { border-left: 3px solid var(--input); }
.nav-skill_loaded { border-left: 3px solid var(--skill); }
.main { padding: 1.5rem 2rem 3rem; max-width: 1100px; }
.hero {
background: linear-gradient(145deg, #ffffff 0%, #f0f5ff 55%, #f3faf6 100%);
border: 1px solid var(--border);
border-radius: 12px;
padding: 1.5rem;
margin-bottom: 1.5rem;
box-shadow: 0 1px 2px rgba(26, 35, 50, 0.04);
}
.hero h2 { margin: 0 0 0.75rem; font-size: 1.35rem; }
.stats {
display: flex;
flex-wrap: wrap;
gap: 0.75rem;
margin: 1rem 0;
}
.stat {
background: #ffffff;
border: 1px solid var(--border);
border-radius: 8px;
padding: 0.55rem 0.85rem;
min-width: 110px;
box-shadow: 0 1px 1px rgba(26, 35, 50, 0.03);
}
.stat .label { font-size: 0.7rem; color: var(--muted); text-transform: uppercase; letter-spacing: 0.04em; }
.stat .value { font-family: var(--mono); font-size: 1rem; margin-top: 0.15rem; }
.user-prompt {
background: #eff6ff;
border: 1px solid #bfdbfe;
border-radius: 8px;
padding: 0.85rem 1rem;
white-space: pre-wrap;
}
.final {
margin-top: 1rem;
background: #ecfdf5;
border: 1px solid #a7f3d0;
border-radius: 8px;
padding: 0.85rem 1rem;
}
.final h3 { margin: 0 0 0.5rem; font-size: 0.85rem; color: var(--output); }
.timeline { display: flex; flex-direction: column; gap: 1rem; }
.card {
background: var(--bg-card);
border: 1px solid var(--border);
border-radius: 10px;
overflow: hidden;
box-shadow: 0 1px 2px rgba(26, 35, 50, 0.04);
}
.card-step { border-top: 3px solid var(--input); }
.card-skill { border-top: 3px solid var(--skill); }
.card-header {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 0.6rem;
padding: 0.75rem 1rem;
background: #f8fafc;
border-bottom: 1px solid var(--border);
}
.step-num {
font-family: var(--mono);
font-size: 0.75rem;
color: var(--muted);
background: #eef2f7;
padding: 0.15rem 0.45rem;
border-radius: 4px;
}
.card-title { font-weight: 600; font-size: 0.95rem; }
.card-tags { display: flex; flex-wrap: wrap; gap: 0.35rem; margin-left: auto; }
.tag {
font-size: 0.7rem;
font-family: var(--mono);
background: #ffffff;
border: 1px solid var(--border);
border-radius: 999px;
padding: 0.12rem 0.5rem;
color: var(--muted);
}
.card-body { padding: 1rem; }
.card-body h4 {
margin: 1rem 0 0.5rem;
font-size: 0.8rem;
color: var(--muted);
text-transform: uppercase;
letter-spacing: 0.05em;
}
.card-body h4:first-child { margin-top: 0; }
.card-body.step-body { padding: 0; }
.substep { padding: 1rem; }
.substep + .substep { border-top: 1px solid var(--border); }
.substep-header {
display: flex;
flex-wrap: wrap;
align-items: center;
gap: 0.5rem;
margin-bottom: 0.65rem;
}
.substep-title {
font-weight: 600;
font-size: 0.8rem;
color: var(--muted);
text-transform: uppercase;
letter-spacing: 0.05em;
}
.sublabel {
font-size: 0.72rem;
color: var(--muted);
text-transform: uppercase;
letter-spacing: 0.05em;
margin-bottom: 0.35rem;
}
.muted { color: var(--muted); font-size: 0.85rem; }
.code {
font-family: var(--mono);
font-size: 0.78rem;
background: var(--code-bg);
border: 1px solid var(--border);
border-radius: 6px;
padding: 0.75rem;
overflow: auto;
max-height: 420px;
white-space: pre-wrap;
word-break: break-word;
margin: 0;
color: #334155;
}
.prose { line-height: 1.6; }
.reasoning {
background: #fffbeb;
border: 1px solid #fde68a;
border-radius: 8px;
padding: 0.75rem;
margin-bottom: 1rem;
}
.msg-list { display: flex; flex-direction: column; gap: 0.4rem; }
.msg {
border: 1px solid var(--border);
border-radius: 6px;
background: #f8fafc;
}
.msg summary {
cursor: pointer;
padding: 0.45rem 0.65rem;
display: flex;
align-items: center;
gap: 0.5rem;
list-style: none;
}
.msg summary::-webkit-details-marker { display: none; }
.msg-body { padding: 0 0.65rem 0.65rem; }
.msg-preview { color: var(--muted); font-size: 0.78rem; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; flex: 1; }
.badge {
font-size: 0.65rem;
font-weight: 600;
text-transform: uppercase;
padding: 0.12rem 0.4rem;
border-radius: 4px;
font-family: var(--mono);
color: #fff;
}
.role-system { background: #475569; }
.role-user { background: #2563eb; }
.role-assistant { background: #059669; }
.role-tool { background: #d97706; }
.chip-row { display: flex; flex-wrap: wrap; gap: 0.35rem; margin-bottom: 0.5rem; }
.chip {
font-family: var(--mono);
font-size: 0.72rem;
background: #fff7ed;
border: 1px solid #fdba74;
color: #c2410c;
padding: 0.15rem 0.5rem;
border-radius: 999px;
}
.usage {
display: flex;
flex-wrap: wrap;
gap: 0.75rem;
margin-top: 0.75rem;
font-family: var(--mono);
font-size: 0.72rem;
color: var(--muted);
}
.sysprompt-block { margin: 0.85rem 0; }
.markdown-box {
background: var(--code-bg);
border: 1px solid var(--border);
border-radius: 6px;
padding: 0.75rem 1rem;
max-height: 420px;
overflow: auto;
}
.markdown-body { font-size: 0.88rem; line-height: 1.65; color: var(--text); }
.markdown-body > *:first-child { margin-top: 0; }
.markdown-body > *:last-child { margin-bottom: 0; }
.markdown-body p { margin: 0.5rem 0; }
.markdown-body h1,
.markdown-body h2,
.markdown-body h3,
.markdown-body h4,
.markdown-body h5,
.markdown-body h6 {
margin: 1rem 0 0.5rem;
line-height: 1.35;
color: var(--text);
}
.markdown-body h1 { font-size: 1.15rem; }
.markdown-body h2 { font-size: 1.05rem; }
.markdown-body h3 { font-size: 0.98rem; }
.markdown-body h4 { font-size: 0.9rem; }
.markdown-body ul,
.markdown-body ol { margin: 0.4rem 0; padding-left: 1.4rem; }
.markdown-body li { margin: 0.2rem 0; }
.markdown-body li > p { margin: 0.2rem 0; }
.markdown-body code {
background: #eef2f7;
border-radius: 4px;
padding: 0.1rem 0.35rem;
}
.markdown-body pre {
background: #0f172a;
color: #e2e8f0;
border-radius: 6px;
padding: 0.65rem 0.85rem;
overflow: auto;
margin: 0.5rem 0;
}
.markdown-body pre code { background: transparent; padding: 0; color: inherit; }
.markdown-body blockquote {
border-left: 3px solid var(--border);
margin: 0.5rem 0;
padding: 0.1rem 0.75rem;
color: var(--muted);
}
.markdown-body table { border-collapse: collapse; margin: 0.5rem 0; font-size: 0.82rem; }
.markdown-body th,
.markdown-body td { border: 1px solid var(--border); padding: 0.3rem 0.55rem; }
.markdown-body a { text-decoration: underline; }
.markdown-body strong { font-weight: 600; }
.markdown-body hr { border: none; border-top: 1px solid var(--border); margin: 0.75rem 0; }
code { font-family: var(--mono); font-size: 0.85em; }
@media (max-width: 900px) {
.layout { grid-template-columns: 1fr; }
.sidebar { position: relative; height: auto; max-height: 40vh; }
}
"""
def _render_system_prompt_block(system_prompt: str | None) -> str:
if not system_prompt:
return ""
return f"""
System Prompt
{_render_markdown(system_prompt)}
"""
def _render_available_tools_block(tool_names: list[str]) -> str:
if not tool_names:
return ""
chips = "".join(f'{_esc(n)}' for n in tool_names)
return f"""
可用工具({len(tool_names)})
{chips}
"""
def render_html(events: list[dict[str, Any]]) -> str:
"""Render a full standalone HTML page for the given events."""
meta = summarize_run(events)
skills = meta.get("skills_used") or []
skills_str = ", ".join(skills) if skills else "—"
system_prompt, tool_names = _extract_system_prompt_and_tools(events)
return f"""
Agent Run · {_esc(meta.get("run_id") or "unknown")}
运行概览
Agent
{_esc(meta.get("agent_name") or "—")}
Model
{_esc(meta.get("model") or "—")}
Iterations
{_esc(meta.get("iterations"))}
Tool Calls
{_esc(meta.get("tool_calls_made"))}
Events
{_esc(meta.get("event_count"))}
{_render_available_tools_block(tool_names)}
{_render_system_prompt_block(system_prompt)}
用户输入
{_esc(meta.get("user_input") or "—")}
{f'''
最终回答
{_render_markdown(meta.get("final_content"), soft_breaks=True, boxed=False)}
''' if meta.get("final_content") else ""}
执行时间线
按步骤展示:LLM 输入 → 思考/输出
{_render_timeline(events)}
"""
def generate_visualization(
run_path: Path | str,
output: Path | str | None = None,
) -> Path:
"""
Load a run log and write an HTML visualization.
Returns the output HTML path.
"""
run_path = Path(run_path)
events = load_run_events(run_path)
if not events:
raise ValueError(f"No events found in {run_path}")
html_doc = render_html(events)
if output is None:
# Prefer placing next to the source file
src = run_path
if src.suffix in {".log", ".jsonl"}:
out = src.with_suffix(".html")
else:
out = Path(str(src) + ".html")
else:
out = Path(output)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(html_doc, encoding="utf-8")
return out