"""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
from supply_agent.logging.parser import load_run_events, summarize_run
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:
content = _pretty(text) if not isinstance(text, str) else text
if content is None:
content = ""
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,
*,
index_offset: int = 0,
default_open: bool | None = None,
) -> str:
if not messages:
return '无消息
'
parts: list[str] = []
for i, msg in enumerate(messages):
role = msg.get("role", "?")
body_parts: list[str] = []
if msg.get("content"):
body_parts.append(_pre(msg["content"], klass="code prose"))
if msg.get("tool_calls"):
body_parts.append(
'tool_calls
'
+ _pre(msg["tool_calls"])
)
if msg.get("tool_call_id"):
body_parts.append(
f'tool_call_id: {_esc(msg["tool_call_id"])}
'
)
if msg.get("name"):
body_parts.append(
f'name: {_esc(msg["name"])}
'
)
if default_open is None:
open_attr = "open" if role != "system" else ""
else:
open_attr = "open" if default_open else ""
parts.append(
f"""
{_role_badge(role)} #{index_offset + i + 1}
{_esc(_preview(msg.get("content")))}
{"".join(body_parts) or '
(empty)
'}
"""
)
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 _render_tools_schema(tools: list[dict[str, Any]] | None) -> str:
if not tools:
return '本次请求未附带 tools
'
names = []
for t in tools:
fn = t.get("function") or t
names.append(fn.get("name", "?"))
chips = "".join(f'{_esc(n)}' for n in names)
return f"""
{chips}
查看完整 tools schema
{_pre(tools)}
"""
def _render_llm_input(
data: dict[str, Any],
seq: int,
iteration: Any,
*,
prev_messages: list[dict[str, Any]] | None = None,
) -> str:
messages = list(data.get("messages") or [])
history, latest = _split_history_and_latest(messages, prev_messages)
history_count = len(history)
latest_count = len(latest)
history_html = ""
if history:
history_html = f"""
查看历史输入({history_count} 条)
{_render_messages(history, default_open=False)}
"""
return f"""
本轮新增输入
{_render_messages(latest, index_offset=history_count, default_open=True)}
{history_html}
Available Tools
{_render_tools_schema(data.get("tools"))}
"""
def _render_reasoning(reasoning: Any) -> str:
if not reasoning:
return """
思考过程
本次回复未返回 reasoning 字段
"""
return f"""
思考过程
{_pre(reasoning, klass="code prose reasoning-text")}
"""
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 _render_planned_tool_calls(tool_calls: list[dict[str, Any]]) -> str:
"""Show each planned tool as name + args; keep raw JSON in a collapsible."""
if not tool_calls:
return ""
cards: list[str] = []
for i, tc in enumerate(tool_calls, 1):
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")
args = _parse_tool_arguments(raw_args)
tc_id = tc.get("id") or ""
cards.append(
f"""
"""
)
return f"""
模型决定调用的工具
{"".join(cards)}
查看原始 tool_calls JSON
{_pre(tool_calls)}
"""
def _render_llm_output(data: dict[str, Any], seq: int, iteration: Any) -> str:
tool_calls = data.get("tool_calls") or []
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", "—"))}
"""
tool_calls_html = _render_planned_tool_calls(tool_calls)
content = data.get("content")
content_html = (
f"模型输出文本
{_pre(content, klass='code prose')}"
if content
else '模型输出文本
(空,可能仅有 tool_calls)
'
)
raw_html = ""
if data.get("raw"):
raw_html = f"""
原始 API 响应 (raw)
{_pre(data["raw"])}
"""
return f"""
{_render_reasoning(data.get("reasoning"))}
{content_html}
{tool_calls_html}
{usage_html}
{raw_html}
"""
def _render_tool_call(data: dict[str, Any], seq: int, iteration: Any) -> str:
is_error = bool(data.get("is_error"))
status = "error" if is_error else "ok"
args = data.get("arguments_parsed", data.get("arguments"))
result = data.get("result_parsed", data.get("result"))
tool_id = data.get("tool_call_id") or ""
return f"""
输入 (arguments)
{_pre(args)}
→
输出 (result)
{_pre(result)}
"""
def _render_skill(data: dict[str, Any], seq: int, iteration: Any) -> str:
return f"""
"""
def _render_timeline(events: list[dict[str, Any]]) -> str:
parts: list[str] = []
prev_llm_messages: list[dict[str, Any]] | None = None
for ev in events:
etype = ev.get("event")
data = ev.get("data") or {}
seq = ev.get("seq", 0)
iteration = ev.get("iteration", data.get("iteration", "—"))
if etype == "run_start":
continue
if etype == "run_end":
continue
if etype == "llm_input":
messages = list(data.get("messages") or [])
parts.append(
_render_llm_input(
data,
seq,
iteration,
prev_messages=prev_llm_messages,
)
)
prev_llm_messages = messages
elif etype == "llm_output":
parts.append(_render_llm_output(data, seq, iteration))
elif etype == "tool_call":
parts.append(_render_tool_call(data, seq, iteration))
elif etype == "skill_loaded":
parts.append(_render_skill(data, seq, 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 输出",
"tool_call": "工具",
"skill_loaded": "技能",
}
for ev in events:
etype = ev.get("event")
if etype in ("run_start", "run_end"):
continue
data = ev.get("data") or {}
seq = ev.get("seq", 0)
label = labels.get(etype, etype or "?")
detail = ""
if etype == "tool_call":
detail = f" · {_esc(data.get('tool', ''))}"
elif etype in ("llm_input", "llm_output"):
detail = f" · iter {ev.get('iteration', data.get('iteration', ''))}"
items.append(
f''
f'{seq}{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;
--tool: #d97706;
--tool-ok: #059669;
--tool-err: #dc2626;
--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-tool_call { border-left: 3px solid var(--tool); }
.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-input { border-top: 3px solid var(--input); }
.card-output { border-top: 3px solid var(--output); }
.card-tool { border-top: 3px solid var(--tool); }
.card-tool.error { border-top-color: var(--tool-err); }
.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);
}
.tag-ok { color: var(--tool-ok); border-color: #86efac; background: #f0fdf4; }
.tag-error { color: var(--tool-err); border-color: #fecaca; background: #fef2f2; }
.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; }
.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;
}
.reasoning.empty { opacity: 0.75; }
.reasoning-text { max-height: 360px; border-color: #fcd34d; background: #fffef5; }
.tool-io {
display: grid;
grid-template-columns: 1fr auto 1fr;
gap: 0.75rem;
align-items: start;
}
.io-arrow {
color: var(--muted);
font-size: 1.4rem;
padding-top: 1.6rem;
}
.io-block { min-width: 0; }
.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-idx { font-family: var(--mono); font-size: 0.7rem; color: var(--muted); }
.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;
}
.planned-tool-list {
display: flex;
flex-direction: column;
gap: 0.65rem;
margin-bottom: 0.5rem;
}
.planned-tool {
border: 1px solid #fed7aa;
background: #fffbeb;
border-radius: 8px;
padding: 0.65rem 0.75rem;
}
.planned-tool-head {
display: flex;
align-items: center;
gap: 0.5rem;
margin-bottom: 0.45rem;
flex-wrap: wrap;
}
.planned-tool-idx {
font-family: var(--mono);
font-size: 0.7rem;
color: var(--muted);
}
.planned-tool-id {
font-size: 0.68rem;
color: var(--muted);
margin-left: auto;
}
.usage {
display: flex;
flex-wrap: wrap;
gap: 0.75rem;
margin-top: 0.75rem;
font-family: var(--mono);
font-size: 0.72rem;
color: var(--muted);
}
.raw-block { margin-top: 0.75rem; }
.history-block {
margin: 0.85rem 0 0.5rem;
border: 1px solid var(--border);
border-radius: 8px;
background: #f8fafc;
padding: 0.35rem 0.65rem 0.65rem;
}
.history-block > summary {
cursor: pointer;
font-size: 0.85rem;
color: var(--muted);
padding: 0.35rem 0;
font-weight: 500;
}
.history-body { margin-top: 0.35rem; }
.meta-line { font-size: 0.8rem; color: var(--muted); margin-bottom: 0.35rem; }
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; }
.tool-io { grid-template-columns: 1fr; }
.io-arrow { display: none; }
}
"""
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 "—"
return f"""
Agent Run · {_esc(meta.get("run_id") or "unknown")}
运行概览
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"))}
用户输入
{_esc(meta.get("user_input") or "—")}
{f'''
最终回答
{_esc(meta.get("final_content") or "")}
''' 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