"""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"""
Step {seq}
LLM 输入
iteration {iteration} {_esc(data.get("model", ""))} temp {_esc(data.get("temperature", ""))} 新增 {latest_count} 共 {len(messages)} msgs

本轮新增输入

{_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"""
{_esc(name)} #{i} {f'{_esc(tc_id)}' if tc_id else ""}
参数
{_pre(args)}
""" ) 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"""
Step {seq}
LLM 输出
iteration {iteration} {'有思考' if data.get('reasoning') or data.get('has_reasoning') else '无思考'} {len(tool_calls)} tool calls
{_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"""
Step {seq}
工具调用 · {_esc(data.get("tool", "?"))}
iteration {iteration} {"ERROR" if is_error else "OK"} {f'{_esc(tool_id)}' if tool_id else ""}
输入 (arguments)
{_pre(args)}
输出 (result)
{_pre(result)}
""" def _render_skill(data: dict[str, Any], seq: int, iteration: Any) -> str: return f"""
Step {seq}
技能加载
iteration {iteration} {_esc(data.get("skill", ""))}
""" 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"""
Step {seq}
{_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 输出", "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"))}
Skills
{_esc(skills_str)}
用户输入
{_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