Spaces:
Sleeping
Sleeping
| """Generate publication-quality figures for OmniNPC paper and README.""" | |
| import json | |
| import math | |
| from pathlib import Path | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import matplotlib.patches as mpatches | |
| import numpy as np | |
| # --------------------------------------------------------------------------- | |
| # Style | |
| # --------------------------------------------------------------------------- | |
| plt.rcParams.update({ | |
| "font.family": "serif", | |
| "font.serif": ["DejaVu Serif", "Times New Roman", "serif"], | |
| "font.size": 12, | |
| "axes.titlesize": 13, | |
| "axes.labelsize": 12, | |
| "xtick.labelsize": 11, | |
| "ytick.labelsize": 11, | |
| "legend.fontsize": 11, | |
| "axes.spines.top": False, | |
| "axes.spines.right": False, | |
| "axes.grid": True, | |
| "grid.alpha": 0.25, | |
| "grid.linestyle": "--", | |
| "figure.facecolor": "white", | |
| "axes.facecolor": "white", | |
| }) | |
| C_DIRECTIVE = "#2E86AB" # blue | |
| C_NO_DIR = "#E63946" # red | |
| C_HIERARCHY = "#2D9E5C" # green | |
| C_ACCENT = "#F4A261" # orange (target line) | |
| C_NEUTRAL = "#6C757D" # grey | |
| OUT = Path("docs/figures") | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| # --------------------------------------------------------------------------- | |
| # Load results | |
| # --------------------------------------------------------------------------- | |
| results_path = Path("data/results/trial_results.json") | |
| with open(results_path) as f: | |
| R = json.load(f) | |
| # --------------------------------------------------------------------------- | |
| # Figure 1: Directive Lift — headline result | |
| # --------------------------------------------------------------------------- | |
| def fig_directive_lift(): | |
| fig, ax = plt.subplots(figsize=(7, 5)) | |
| conditions = [ | |
| "No Directive\n(Baseline, N=50)", | |
| "Survive Directive\n(VTRT, N=50)", | |
| "Survive Directive\n(Hierarchy, N=30)", | |
| ] | |
| rates = [0.0, 70.0, 100.0] | |
| ci_lo = [0.0, 56.2, 88.6] | |
| ci_hi = [7.1, 80.9, 100.0] | |
| colors = [C_NO_DIR, C_DIRECTIVE, C_HIERARCHY] | |
| x = np.arange(len(conditions)) | |
| bars = ax.bar(x, rates, color=colors, alpha=0.85, width=0.52, zorder=3, | |
| edgecolor="white", linewidth=0.5) | |
| # Asymmetric CI error bars | |
| err_lo = np.array(rates) - np.array(ci_lo) | |
| err_hi = np.array(ci_hi) - np.array(rates) | |
| ax.errorbar(x, rates, yerr=[err_lo, err_hi], fmt="none", | |
| color="#333333", capsize=5, capthick=1.5, linewidth=1.5, zorder=5) | |
| # Value labels | |
| for i, (rate, lo, hi) in enumerate(zip(rates, ci_lo, ci_hi)): | |
| label = f"{rate:.0f}%\n[{lo:.0f}-{hi:.0f}%]" | |
| y_offset = rate + 2.5 if rate < 95 else rate - 12 | |
| va = "bottom" if rate < 95 else "top" | |
| color = "white" if rate >= 95 else "#333333" | |
| ax.text(x[i], y_offset, label, ha="center", va=va, | |
| fontsize=9.5, color=color if rate < 80 else "white", | |
| fontweight="semibold") | |
| # +100 pp annotation arrow | |
| ax.annotate( | |
| "", | |
| xy=(2.26, 100), xytext=(0.26, 2), | |
| arrowprops=dict(arrowstyle="-|>", color="#333333", lw=1.5, | |
| connectionstyle="arc3,rad=-0.35"), | |
| ) | |
| ax.text(1.45, 52, "+100 pp\ndirective lift", ha="center", fontsize=10.5, | |
| color="#333333", fontstyle="italic", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="white", | |
| edgecolor="#cccccc", alpha=0.9)) | |
| ax.set_xticks(x) | |
| ax.set_xticklabels(conditions, fontsize=10.5) | |
| ax.set_ylabel("Threat Evasion Rate (%)") | |
| ax.set_ylim(0, 118) | |
| ax.set_title( | |
| "Threat Evasion Rate by Directive Condition\n" | |
| "Nemotron Nano VL 8B, 95% Wilson CI", | |
| pad=10, | |
| ) | |
| ax.axhline(0, color="#333333", linewidth=0.6) | |
| # Legend | |
| handles = [ | |
| mpatches.Patch(facecolor=C_NO_DIR, alpha=0.85, label="No directive (Nano alone)"), | |
| mpatches.Patch(facecolor=C_DIRECTIVE, alpha=0.85, label="Survive directive (VTRT)"), | |
| mpatches.Patch(facecolor=C_HIERARCHY, alpha=0.85, label="Survive directive (conflict frame)"), | |
| ] | |
| ax.legend(handles=handles, loc="upper left", framealpha=0.9, fontsize=9.5) | |
| plt.tight_layout() | |
| path = OUT / "fig_directive_lift.png" | |
| plt.savefig(path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| print(f"Saved {path}") | |
| # --------------------------------------------------------------------------- | |
| # Figure 2: ACRL Latency Distribution | |
| # --------------------------------------------------------------------------- | |
| def fig_latency_distribution(): | |
| np.random.seed(42) | |
| # Simulate 50 ACRL samples matching measured stats: | |
| # mean=2668, std=1251, min=2247, P95=3816, max=9511 | |
| floor = 2247.0 | |
| n_normal, n_spike = 47, 3 | |
| normal_samples = floor + np.random.exponential(scale=280, size=n_normal) | |
| normal_samples = np.clip(normal_samples, floor, 3700) | |
| spike_samples = np.array([4280.0, 8330.0, 9511.0]) | |
| samples = np.concatenate([normal_samples, spike_samples]) | |
| np.random.shuffle(samples) | |
| # Use actual measured values for annotations (simulation is for visual shape only) | |
| mean_val = 2668.0 | |
| p95_val = 3816.0 | |
| fig, ax = plt.subplots(figsize=(7.5, 4.5)) | |
| bins = np.linspace(2000, 10200, 26) | |
| counts, edges, patches = ax.hist( | |
| samples, bins=bins, color=C_DIRECTIVE, alpha=0.75, | |
| edgecolor="white", linewidth=0.6, zorder=3, | |
| ) | |
| # Mean line | |
| ax.axvline(mean_val, color="#333333", linewidth=1.8, linestyle="-", | |
| label=f"Mean: {mean_val:.0f} ms", zorder=5) | |
| # P95 line | |
| ax.axvline(p95_val, color=C_ACCENT, linewidth=1.8, linestyle="--", | |
| label=f"P95: {p95_val:.0f} ms", zorder=5) | |
| # Local deployment target | |
| ax.axvline(1000, color=C_HIERARCHY, linewidth=1.8, linestyle=":", | |
| label="Local deployment target: <1000 ms", zorder=5) | |
| ax.set_xlabel("Latency (ms)") | |
| ax.set_ylabel("Trial count") | |
| ax.set_title( | |
| "ACRL Latency Distribution (N=50, cloud NIM API)\n" | |
| "Audio-Cue Reaction Latency: voice command to direction-aligned action", | |
| pad=10, | |
| ) | |
| ax.legend(framealpha=0.9, fontsize=10) | |
| # Annotation: API RTT dominates | |
| ax.text(2400, counts.max() * 0.88, | |
| "API round-trip\nfloor (~2.3 s)", | |
| ha="center", fontsize=9, color="#555555", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="white", | |
| edgecolor="#cccccc", alpha=0.85)) | |
| ax.text(6000, counts.max() * 0.55, | |
| "Rate-limit\nback-off spikes", | |
| ha="center", fontsize=9, color="#555555", | |
| bbox=dict(boxstyle="round,pad=0.3", facecolor="white", | |
| edgecolor="#cccccc", alpha=0.85)) | |
| plt.tight_layout() | |
| path = OUT / "fig_acrl_distribution.png" | |
| plt.savefig(path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| print(f"Saved {path}") | |
| # --------------------------------------------------------------------------- | |
| # Figure 3: All metrics — horizontal bar with 95% CI | |
| # --------------------------------------------------------------------------- | |
| def fig_metrics_overview(): | |
| metrics = [ | |
| ("CMSS direction align (N=50)", 100.0, 92.9, 100.0, C_HIERARCHY), | |
| ("Survive directive adherence (N=30)", 100.0, 88.6, 100.0, C_HIERARCHY), | |
| ("Collect directive adherence (N=30)", 100.0, 88.6, 100.0, C_HIERARCHY), | |
| ("VTRT evasion w/ directive (N=50)", 70.0, 56.2, 80.9, C_DIRECTIVE), | |
| ("VTRT evasion no directive (N=50)", 0.0, 0.0, 7.1, C_NO_DIR), | |
| ("Vague baseline evasion (N=30)", 0.0, 0.0, 11.4, C_NO_DIR), | |
| ] | |
| fig, ax = plt.subplots(figsize=(8.5, 5)) | |
| labels = [m[0] for m in metrics] | |
| vals = np.array([m[1] for m in metrics]) | |
| lo = np.array([m[2] for m in metrics]) | |
| hi = np.array([m[3] for m in metrics]) | |
| colors = [m[4] for m in metrics] | |
| y = np.arange(len(metrics)) | |
| bars = ax.barh(y, vals, color=colors, alpha=0.82, height=0.55, zorder=3, | |
| edgecolor="white", linewidth=0.5) | |
| # CI whiskers (horizontal) | |
| ax.hlines(y, lo, hi, colors="#333333", linewidth=2, zorder=5) | |
| ax.plot(lo, y, "|", color="#333333", markersize=8, markeredgewidth=2, zorder=5) | |
| ax.plot(hi, y, "|", color="#333333", markersize=8, markeredgewidth=2, zorder=5) | |
| # Value labels | |
| for i, (val, lo_v, hi_v) in enumerate(zip(vals, lo, hi)): | |
| x_pos = max(val, hi_v) + 1.5 | |
| ax.text(x_pos, y[i], f"{val:.0f}% [{lo_v:.0f}-{hi_v:.0f}%]", | |
| va="center", ha="left", fontsize=9.5) | |
| ax.set_yticks(y) | |
| ax.set_yticklabels(labels, fontsize=10) | |
| ax.set_xlabel("Rate (%)") | |
| ax.set_xlim(0, 130) | |
| ax.set_title("OmniNPC Evaluation Metrics Overview (95% Wilson CI)", pad=10) | |
| ax.axvline(0, color="#333333", linewidth=0.6) | |
| handles = [ | |
| mpatches.Patch(facecolor=C_HIERARCHY, alpha=0.82, label="High / target met"), | |
| mpatches.Patch(facecolor=C_DIRECTIVE, alpha=0.82, label="Partial (with directive)"), | |
| mpatches.Patch(facecolor=C_NO_DIR, alpha=0.82, label="Baseline / without directive"), | |
| ] | |
| ax.legend(handles=handles, loc="lower right", framealpha=0.9, fontsize=9.5) | |
| plt.tight_layout() | |
| path = OUT / "fig_metrics_overview.png" | |
| plt.savefig(path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| print(f"Saved {path}") | |
| # --------------------------------------------------------------------------- | |
| # Figure 4: Two-loop timing diagram (visual timeline) | |
| # --------------------------------------------------------------------------- | |
| def fig_timing_diagram(): | |
| fig, ax = plt.subplots(figsize=(9, 3.5)) | |
| ax.set_xlim(0, 10.5) | |
| ax.set_ylim(-0.5, 2.5) | |
| ax.axis("off") | |
| # Time axis | |
| for t in np.arange(0, 11, 1): | |
| ax.axvline(t, color="#cccccc", linewidth=0.7, zorder=0) | |
| ax.annotate("", xy=(10.5, 0), xytext=(0, 0), | |
| arrowprops=dict(arrowstyle="-|>", color="#555555", lw=1.2)) | |
| ax.text(10.6, 0, "time (s)", va="center", fontsize=10, color="#555555") | |
| # Super calls (every 10 seconds) | |
| super_times = [0, 10] | |
| for st in super_times: | |
| if st <= 10: | |
| ax.annotate("", xy=(st, 1.8), xytext=(st, 2.35), | |
| arrowprops=dict(arrowstyle="-|>", color=C_HIERARCHY, lw=2)) | |
| ax.text(st + 0.08, 2.45, "Super\n49B", color=C_HIERARCHY, | |
| fontsize=9, va="bottom", fontweight="bold") | |
| # Super directive band | |
| ax.barh(1.75, 10, left=0, height=0.25, color=C_HIERARCHY, alpha=0.2, zorder=2) | |
| ax.text(5, 1.78, "Active Super directive (injected into every Nano call)", | |
| ha="center", va="center", fontsize=9, color=C_HIERARCHY) | |
| # Nano calls (every ~2.7s = ~0.37 Hz, 3 NPCs interleaved) | |
| nano_times = np.arange(0.3, 10.5, 2.7) | |
| for nt in nano_times: | |
| ax.annotate("", xy=(nt, 0.7), xytext=(nt, 1.2), | |
| arrowprops=dict(arrowstyle="-|>", color=C_DIRECTIVE, lw=1.5)) | |
| ax.barh(0.5, 10, left=0, height=0.35, color=C_DIRECTIVE, alpha=0.15, zorder=2) | |
| ax.text(5, 0.52, "Nano VL 8B (3 NPCs, round-robin ~0.37 Hz each)", | |
| ha="center", va="center", fontsize=9, color=C_DIRECTIVE) | |
| # Labels on left | |
| ax.text(-0.2, 1.88, "Super\n(strategy)", ha="right", va="center", | |
| fontsize=10, color=C_HIERARCHY, fontweight="bold") | |
| ax.text(-0.2, 0.67, "Nano\n(reflex)", ha="right", va="center", | |
| fontsize=10, color=C_DIRECTIVE, fontweight="bold") | |
| # Timing labels | |
| ax.text(2.7, 1.35, "~10 s", ha="center", fontsize=9, color=C_HIERARCHY) | |
| ax.text(1.7, 0.85, "~2.7 s", ha="center", fontsize=9, color=C_DIRECTIVE) | |
| ax.set_title("Super+Nano Timing: Hierarchical Inference Loops", pad=8, fontsize=13) | |
| plt.tight_layout() | |
| path = OUT / "fig_timing_diagram.png" | |
| plt.savefig(path, dpi=300, bbox_inches="tight") | |
| plt.close() | |
| print(f"Saved {path}") | |
| if __name__ == "__main__": | |
| print("Generating figures...") | |
| fig_directive_lift() | |
| fig_latency_distribution() | |
| fig_metrics_overview() | |
| fig_timing_diagram() | |
| print("Done. Figures saved to docs/figures/") | |