智能软件工程 · 案例实战：Agent 度量仪表盘——从"能跑"到"可运营"

为什么讲这个案例：它把「指标定义 → 采集埋点 → 看板呈现 → 门禁行动」串成一条完整链路，且直接复用 pr03 的 Agent loop，不需要新写业务就能做实证。它也是全系列的"数据枢纽"：上游吃 pr03 / pr05 / pr06 / pr07 产生的轨迹，下游喂 pr09（失败编码）、pr14（统计推断）、pr15（版本回归门禁）。

阅读路径：§1 立"三层"框架（为什么单层度量必然自欺），§2 交付本篇真正的核心——指标定义文档表，§3 用一个反例说明"定义先于采集"，§4–§5 是可运行的 trace 中间件与看板，§6 给 p95 与置信区间的数学直觉，§7–§8 讲仪表盘落地与"从度量到行动"，§9–§10 是诚实标注与接口。

1. 度量的三层：模型层 / 工具层 / 工作流层

很多团队只盯着"任务成没成"这一个数——这是最粗也最没用的一层。一个 Agent 跑失败，可能是模型吞吐塌了、某个工具连续 500、也可能是计划步骤绕进了死循环；三者的修法完全不同。所以度量必须分三层下钻，每层有自己的自变量，逐层归因：

1.1 模型层：这个"大脑"能不能扛住

关心底座本身在给定硬件/量化配置下的表现：基准保留率（量化后相对全精度掉了多少分）、tokens/s（稳态解码吞吐）、峰值显存。这一层的口径与列，由 pr11 / pr12 建立的度量四元组提供——本篇不重造，只是把它们并进统一 traces.csv（§4）。模型层劣化往往不直接让任务失败，而是"拖慢 + 逼上层多走几步"，所以必须单独度量。

1.2 工具层：能不能用、坏了能不能自己爬起来

两个指标：工具调用成功率（每次 tool call 是否返回可用观测）与失败恢复率（一次工具失败后，Agent 是否在预算内重试/换路成功，pr03 的 Reflexion / 退避就体现在这里）。工具层是 Agent 区别于纯 LLM 的地方——光看"成功"不够，还要看"失败了会不会自愈"。

1.3 工作流层：业务方真正关心的可运营性

端到端视角：任务成功率 SR、人工介入率（多少任务需要人来兜底，直接决定 ROI）、端到端延迟 p50/p95、单任务成本。这一层是看板首页，但它的每一个异动都必须能下钻到 §1.1/§1.2 的归因，否则就是"知道坏了、不知道哪儿坏"。

2. 关键指标定义（先钉死"指标定义文档"）

这是本篇真正的交付物：一张可粘贴进团队 Wiki 的指标定义表。每个指标给四样东西——口径、计算公式、数据来源列、验收线；没写全这四项的指标，禁止上看板。最后一列标注它落在哪一层，以及是否与 pr11/pr12 打通。

与 pr11/pr12 打通：后三行（峰值显存 / tokens/s / 基准保留率）正是 pr11/pr12 度量四元组的分量，quant_bits 与 spec_decode 两列也原样沿用到本篇 schema（§4），构成同一口径——模型层的配置因子和上层的运行指标落在同一行，pr14 才能把它们当因子做回归。扩展列是"本篇相对通用做法"的增量：通用仪表盘只测上五行，我们把模型层三列也拉进来，让"慢/贵/掉分"能一路归因到"哪一档量化、开没开投机"。

3. 为什么"指标定义先于采集"

看板好搭，定义难齐。若不在埋点之前把口径钉死，两周后一定演变成同名不同义的扯皮——所有人都在看"成功率"，但没有两个人算的是同一个数。

原则驱动的一句话：采集只是"把数抄下来"，定义才是"规定这数允许被怎样抄"。先有定义文档，埋点字段就是定义的可执行投影（§4 的列名 = §2 表格的来源列）；反过来先埋点再补定义，你永远在给自己的历史数据擦屁股。

4. trace 中间件（核心工程交付）

这是本篇唯一"必须能跑"的交付：在 pr03 的 Agent loop 上挂一层中间件，每一步落一条结构化 JSON trace，汇总进统一 traces.csv。字段沿用并汇总 pr03–pr07 各篇新增的列，一次到位。

4.1 统一 traces.csv schema（汇总 pr03–pr07 新增列）

# file: schema.py —— 全系列统一轨迹列定义（pr03 产出，pr08 消费，pr09/11/12/14/15 复用）
#   run_id          一次端到端任务唯一标识            (pr03)
#   timestamp       该步起始 UTC 时间戳               (pr03)
#   task            任务标签/类别，供分组             (pr07)
#   model,quant_bits,spec_decode  模型层配置因子       (pr11/pr12)
#   steps,tokens,latency           工作流层聚合量      (pr03/pr06)
#   sr,tool_success,human_intervene 三大成功率类指标   (pr03/pr05)
#   peak_vram,tps,bench_retention  模型层四元组分量     (pr11)
#   fail_code       稳定错误码，非自由文本            (pr09)
TRACE_FIELDS = [
    "run_id", "timestamp", "task",
    "model", "quant_bits", "spec_decode",
    "steps", "tokens", "latency",
    "sr", "tool_success", "human_intervene",
    "peak_vram", "tps", "bench_retention",
    "fail_code",
]

def append_trace(path, row):
    """一步（或一个 run 聚合后）追加一行；缺列补空，保证列顺序稳定。"""
    import csv, os
    exists = os.path.exists(path)
    with open(path, "a", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=TRACE_FIELDS)
        if not exists:
            w.writeheader
        w.writerow({k: row.get(k, "") for k in TRACE_FIELDS})

4.2 装饰器 / 回调：把 pr03 的 loop 包起来

不改 Agent 逻辑，只在 react_run 外面套一层计时与打点；模型层三列由 pr11 的运行时探针回填，工具/人工事件由回调显式上报，杜绝"事后猜"。

# file: middleware.py —— 给 pr03 的 Agent loop 套观测（pr08 核心）
import time, uuid, functools
from schema import append_trace, TRACE_FIELDS

class AgentTelemetry:
    def __init__(self, path="./runs/traces.csv", model="local-q4",
                 quant_bits=4, spec_decode="off"):
        self.path, self.model = path, model
        self.quant_bits, self.spec_decode = quant_bits, spec_decode

    def instrument(self, react_run):
        """包一层：跑一次任务，落一条聚合 trace。"""
        @functools.wraps(react_run)
        def wrapped(task, **kw):
            run_id = f"r-{time.strftime('%Y%m%d')}-{uuid.uuid4.hex[:6]}"
            t0 = time.time
            st = {"steps": 0, "tokens": 0, "calls": 0, "calls_ok": 0,
                  "fails": 0, "recovered": 0, "intervene": 0}

            def on_step(phase, tool, ok, tokens, tin, tout):
                st["steps"] += 1
                st["tokens"] += tin + tout
                if tool is not None:
                    st["calls"] += 1
                    st["calls_ok"] += int(ok)
                    if not ok:
                        st["fails"] += 1
                        # 本轮最终是否被重试/换路救回，由 reflexion 回调上报

            def on_recover(recovered):
                st["recovered"] += int(recovered)

            def on_intervene:
                st["intervene"] += 1

            # react_run 需接受 on_step/on_recover/on_intervene 回调（pr03 §9 已预留）
            result = react_run(task, on_step=on_step,
                               on_recover=on_recover, on_intervene=on_intervene, **kw)
            latency = round(time.time - t0, 2)
            sr = 1 if result.ok else 0
            tool_success = st["calls_ok"] / st["calls"] if st["calls"] else 1.0
            append_trace(self.path, dict(
                run_id=run_id, timestamp=time.strftime("%Y-%m-%dT%H:%M:%SZ"),
                task=task.tag, model=self.model,
                quant_bits=self.quant_bits, spec_decode=self.spec_decode,
                steps=st["steps"], tokens=st["tokens"], latency=latency,
                sr=sr, tool_success=round(tool_success, 3),
                human_intervene=st["intervene"],
                peak_vram=result.peak_vram_gb,        # pr11 运行时探针回填
                tps=round(result.decode_tps, 1),      # pr11 预热后稳态吞吐
                bench_retention=result.bench_ret,     # pr12 保留率
                fail_code="" if sr else result.fail_code))  # pr09 编码
            return result
        return wrapped

口径即代码：注意 human_intervene 是显式打点（有人在环就 on_intervene 一次），而不是从"最终是否人工重跑"倒推——这正是 §3 反例的工程解药。latency 走的是挂钟且不剔超时，超时任务 sr=0 计入分母，与 §2 表格的 SR 口径一致。

5. 采集与看板

有了 traces.csv，看板就是"读表 + 画图"。下面这段用 pandas 聚合、matplotlib 出四张图：SR 趋势、p50/p95 延迟、成本分布、tokens/s 与显存仪表盘。工具无关（Grafana / Streamlit / 静态 HTML 皆可，§7）。

# file: dashboard.py —— 读统一 traces.csv，出四块看板
import pandas as pd, numpy as np
import matplotlib.pyplot as plt

df = pd.read_csv("./runs/traces.csv")
df["date"] = pd.to_datetime(df["timestamp"]).dt.date
g = df.groupby("date").agg(
    sr=("sr", "mean"),
    p50=("latency", lambda x: np.percentile(x, 50)),
    p95=("latency", lambda x: np.percentile(x, 95)),
    cost=("tokens", "mean"),
    tps=("tps", "mean"),
    vram=("peak_vram", "max"),
    intervene=("human_intervene", lambda x: (x > 0).mean),
).reset_index

fig, ax = plt.subplots(2, 2, figsize=(11, 7))

# ① 任务成功率 SR + 人工介入率
ax[0, 0].plot(g.date, g.sr * 100, marker="o", label="SR %")
ax[0, 0].plot(g.date, g.intervene * 100, marker="s", ls="--", label="人工介入 %")
ax[0, 0].set_title("工作流层：成功率 / 介入率"); ax[0, 0].legend; ax[0, 0].grid(alpha=.3)

# ② p50 / p95 延迟
ax[0, 1].plot(g.date, g.p50, marker="o", label="p50")
ax[0, 1].plot(g.date, g.p95, marker="x", label="p95")
ax[0, 1].set_title("延迟分位数（尾延迟看 p95）"); ax[0, 1].legend; ax[0, 1].grid(alpha=.3)

# ③ 成本分布（箱线图）
ax[1, 0].boxplot([d["tokens"].values for _, d in df.groupby("date")],
                 labels=[str(x) for x in sorted(df.date.unique)])
ax[1, 0].set_title("单任务 token 成本分布"); ax[1, 0].tick_params(axis="x", rotation=30)

# ④ tokens/s vs 峰值显存 仪表盘（按量化档）
for q, d in df.groupby("quant_bits"):
    ax[1, 1].scatter(d["peak_vram"], d["tps"], s=40, label=f"q{q}")
ax[1, 1].set_xlabel("peak_vram (GB)"); ax[1, 1].set_ylabel("tokens/s")
ax[1, 1].set_title("模型层仪表盘（打通 pr11）"); ax[1, 1].legend; ax[1, 1].grid(alpha=.3)

fig.tight_layout; fig.savefig("dashboard.png", dpi=130)
print("看板已输出 dashboard.png")

下钻而不是躺平：四块图分属三层——① ② ③ 是工作流层，④ 是模型层（显式带 quant_bits 分组，直接对接 pr11 的档位）。一旦发现 ① 的 SR 掉、② 的 p95 涨，切到 ④ 看是不是"某档量化把吞吐拖垮、逼上层多走步"。能下钻，才叫可运营；只有一张总分，叫开盲盒。

6. p95 与置信区间

看板上"p95 = 12.3s"这种单点数最危险——样本一小，它抖得厉害。工程上每个分位数都该带区间。分位数用经验分布函数定义，区间用 bootstrap（对分位数没有闭式方差，重采样最稳）。统计推断的细节对接 pr14，本篇只给直觉与算法。

直白说：从 n 条 latency 里有放回抽 n 条，算一次 p95；重复 B 次，把这 B 个 p95 排序，取 2.5% 和 97.5% 两个位置——就是 p95 的 95% 置信区间。成功率 SR 同理，只是它是有界比例，可用更省事的 Wilson 区间：

import numpy as np
def bootstrap_quantile_ci(x, q=95, B=2000, alpha=0.05, seed=0):
    rng = np.random.default_rng(seed)
    x = np.asarray(x, float); n = len(x)
    boots = [np.quantile(rng.choice(x, n, replace=True), q/100) for _ in range(B)]
    lo, hi = np.quantile(boots, [alpha/2, 1-alpha/2])
    return np.quantile(x, q/100), lo, hi   # 点估计, 95%CI 下界, 上界

为什么要区间：本周 p95=12.3s vs 上周 11.8s，若无区间就是"变慢了"的口水战；带区间后可能完全重叠 → 差异不显著，别乱改系统。反之若 SR 区间整体下移且不重叠，才值得拉响 §8 的警报。分位数 + bootstrap 的完整推断对接 pr14。

7. 仪表盘落地

用什么工具不重要，重要的是三块面板共享 §2 那一套口径、共享同一份 traces.csv。任选其一：

• Grafana 把 CSV 灌进数据源，用现成的时序 + 直方图面板；适合长期值守、告警联动。
• Streamlit st.line_chart(g.set_index('date')[['sr','p95']]) 十几行起一个内部看板；适合原型 / 快速迭代。
• 静态 HTML §5 的 dashboard.png 定时刷新贴页；零依赖，适合离线复现与成果交付。

无论哪种，都预留一个失败热力图入口：把 fail_code（由 pr09 定义编码）按 task × tool 交叉聚合成矩阵，红格即"高频失败组合"，点击下钻回 pr09 的失败归因页。

口径统一的判据：同一个数，在 Grafana、在 Streamlit、在周报里必须完全相等。若不相等，多半不是渲染 bug，而是有人绕开了 §2 定义自己算了一遍——这正是 §3 反例换个马甲回来了。看板落地阶段最该守的不是美观，是"单一口径、单一数据源"。

8. 从度量到行动：指标回落触发 CI 阻断

度量的终点不是"看得见"，而是"看得见的劣化能自动拦下来"。把 §2 的验收线写成门禁阈值，接进 pr15 的版本回归门禁：每次模型/提示词/工具配置变更，跑一小批回归任务生成 trace，指标不达标就阻断合入。

# gate.yml —— 度量驱动的回归门禁（对接 pr15）
gate:
  regression_sample: 50          # 每次变更跑 50 个任务生成 traces.csv
  rules:
    - metric: sr
      must_be: ">= baseline_sr - 0.05"      # SR 相对基线掉超 5 个点 → 阻断
    - metric: p95_latency_s
      must_be: "<= baseline_p95 * 1.2"      # 尾延迟恶化 >20% → 阻断
    - metric: human_intervene_rate
      must_be: "<= 0.10"                     # 人工介入率破线 → 阻断
    - metric: bench_retention
      when: { quant_bits: "< 4" }           # 极端量化档额外收紧
      must_be: ">= 80"                      # 基准保留率跌破 80% → 阻断
  on_fail: block_merge          # 合入被拒，附 dashboard.png 与差值报告

闭环：pr03–pr07 产生轨迹 → 本篇（pr08）度量与看板 → pr09 失败编码定位 → pr15 门禁阻断。度量若止于"好看"，就是仪表盘版的自嗨；能触发行动（阻断/回滚/告警）的度量，才把"能跑"变成"可运营"。

9. 诚实标注

10. 接口：本篇在系列里的位置

一页纸总结：观测先于优化，且必须跨三层（模型 / 工具 / 工作流）→ 一切自欺源于定义含糊，先钉死指标定义文档（§2 表）→ trace 中间件（§4）把 pr03 的 loop 落成统一 traces.csv，与 pr11/pr12 四元组同口径 → 看板（§5）能下钻、能带区间（§6）、能触发门禁（§8）→ 这才把"能跑"变成"可运营"。