"""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'
{body}
' 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"""
LLM 输入 新增 {latest_count}
{_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"""
LLM 输出 {"".join(tags)}
{_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"""
Step {step_no}
{title}
iteration {iteration}
{inner_html}
""" def _render_skill(data: dict[str, Any], step_no: int, iteration: Any) -> str: return f"""
Step {step_no}
技能加载
iteration {iteration} {_esc(data.get("skill", ""))}
""" _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"""
Step {step_no}
{_esc(etype)}
{_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"))}
Skills
{_esc(skills_str)}
{_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