使用 agent-sre 交互式探索 AI Agent 的 SLO 与错误预算:Notebook 实战指南
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
本指南以 agent-governance-toolkit 仓库中 agent-sre 模块的 notebooks/README.md 及其配套 slo-exploration.ipynb 为核心骨架,讲解如何用 Jupyter Notebook 为 AI Agent 系统定义 SLO(Service Level Objective)、模拟 200 次 agent 调用、计算 SLI、追踪错误预算(error budget)与燃烧速率(burn rate),并完成告警阈值检查与 what-if 场景分析。读完本文,你将掌握 agent-sre 中 SLI/SLO/ErrorBudget 的完整编程模型,以及一套可复制到生产环境的"定义指标 → 记录事件 → 评估状态 → 触发告警"闭环工作流。
一、Notebook 概览:为什么用交互式方式学习 SLO
agent-sre(Agent SRE)是 agent-governance-toolkit 面向 AI Agent 可靠性工程的 Python 模块,其 SLO 子系统把经典 SRE 方法论迁移到 agent 场景:用 SLI 度量"agent 表现如何",用 SLO 定义"什么叫可靠",用错误预算回答"还能容忍多少次失败"。
notebooks/README.md 目前提供一枚交互式 Notebook:
| Notebook | Description |
|---|---|
| slo-exploration.ipynb | 定义 SLO、模拟 agent 流量、可视化错误预算与燃烧速率 |
该 Notebook 的完整实验流程共六步:定义 SLO(延迟/准确率/成本三种 SLI)→ 模拟 200 次 agent 调用 → 计算 SLI 值与合规度 → 检查错误预算 → 可视化合规性、燃烧速率与延迟分布 → 检查告警阈值 → 运行 what-if 分析。每一步都配有可独立运行的单元格,适合作为新手上手 agent-sre 的交互式训练场。
二、环境准备与启动方式
原文档列出的前置条件如下,结合仓库当前状态有两点校准说明:
- Python 版本:原文档要求 Python ≥ 3.10,但仓库 agent-governance-python/agent-sre/pyproject.toml 中
requires-python = ">=3.11",建议按 ≥ 3.11 准备环境。 - 安装方式:原文档给出
pip install agent-sre与pip install -e .两种方式。需要说明的是,当前仓库中agent_sre包的 5.0.0 发布 wheel 已声明为弃用存根(deprecation stub),仅重定向到agent-governance-toolkit-cli>=5.0.0,<6.0;仓库内的src/agent_sre源码树仍作为事实标准被 CI 通过pip install -e ".[dev]"使用。因此从仓库根目录执行可编辑安装是稳妥做法:
pip install -e . # 在 agent-sre 目录内,等价于 -e ".[dev]" 的核心安装 pip install matplotlib jupyter # 可视化与 Notebook 运行依赖 jupyter notebook notebooks/ # 启动 Notebook 服务,打开 slo-exploration.ipynb启动后依次运行各单元格即可复现下述全部实验。
三、第 1 步:定义 SLO——三种 SLI 与 5% 错误预算
Notebook 的第一段代码为"代码评审 Agent"定义了三种 SLI,并组合进一个带错误预算的 SLO:
| Indicator | Target | Window | Description |
|---|---|---|---|
| Response Latency (p95) | ≤ 3 000 ms | 1 h | 95 分位延迟 |
| Tool-Call Accuracy | ≥ 99 % | 24 h | 工具选择正确率 |
| Cost per Task | ≤ $0.50 | 24 h | 单任务平均美元成本 |
import random import math import matplotlib.pyplot as plt from agent_sre import SLO, ErrorBudget from agent_sre.slo.indicators import ( CostPerTask, ResponseLatency, TaskSuccessRate, ToolCallAccuracy, ) from agent_sre.slo.objectives import ExhaustionAction, SLOStatus from agent_sre.slo.dashboard import SLODashboard random.seed(42) # --- SLI definitions --- latency_sli = ResponseLatency(target_ms=3000.0, percentile=0.95, window="1h") accuracy_sli = ToolCallAccuracy(target=0.99, window="24h") cost_sli = CostPerTask(target_usd=0.50, window="24h") # --- Error budget (5 %) --- budget = ErrorBudget( total=0.05, burn_rate_alert=2.0, # 2× 正常燃烧 → 警告 burn_rate_critical=10.0, # 10× 正常燃烧 → 严重 exhaustion_action=ExhaustionAction.FREEZE_DEPLOYMENTS, ) # --- SLO --- slo = SLO( name="code-review-agent", description="Reliability targets for an AI code-review agent", indicators=[latency_sli, accuracy_sli, cost_sli], error_budget=budget, agent_id="code-review-agent", ) print(slo)源码层面的参数语义
从源码看,这三种 SLI 都在 src/agent_sre/slo/indicators.py 中实现,继承自抽象基类SLI(见 indicators.py):
- 时间窗口:
TimeWindow枚举(indicators.py)定义了1h / 6h / 24h / 7d / 30d五种标准窗口,分别对应 3600、21600、86400、604800、2592000 秒;current_value()只统计窗口内的测量值,实现滑动窗口聚合。 - ResponseLatency(indicators.py):
percentile=0.95时,current_value()返回窗口内延迟排序后的 p95 值;其compliance()继承自基类——窗口内value >= target的测量占比(注意延迟是"上限型"指标,源码中SLIValue.is_good默认按value >= target判断,延迟场景需要结合记录时的target元数据理解)。 - ToolCallAccuracy(indicators.py):通过
record_call(correct: bool)累积计数,返回运行中的正确率。 - CostPerTask(indicators.py):通过
record_cost(cost_usd)累积总成本并返回平均单任务成本。
ErrorBudget与SLO类定义在 src/agent_sre/slo/objectives.py:
ExhaustionAction(objectives.py)提供四种耗尽动作:ALERT(通知)、FREEZE_DEPLOYMENTS(冻结部署)、CIRCUIT_BREAK(打开熔断器)、THROTTLE(限流降级)。ErrorBudget(objectives.py)的关键属性:total(总预算比例)、consumed(已消耗)、remaining_percent(剩余百分比)、is_exhausted(是否耗尽)、burn_rate(window_seconds)(窗口内燃烧速率)。其事件缓冲为deque(maxlen=max_events),默认max_events=100_000,防止长运行 SLO 内存无界增长。SLO(objectives.py):若未显式传error_budget,会用最严格指标的 target 自动推导total = 1.0 - min(target);record_event(good=bool)在记录事件的同时触发一次evaluate()。
四、第 2 步:模拟 200 次 Agent 调用
为了观察错误预算"被消耗"的过程,Notebook 故意把成功率设置在目标值之下(92% 任务成功 < 95% 预算、98.5% 工具准确率 < 99% 目标):
NUM_CALLS = 200 latencies, accuracies_running, costs = [], [], [] good_events, budget_remaining = [], [] for i in range(NUM_CALLS): task_ok = random.random() < 0.92 # 92 % success (below 95 % budget) tool_ok = random.random() < 0.985 # 98.5 % accuracy (below 99 % target) latency_ms = max(100, random.gauss(2400, 700)) cost_usd = max(0.01, random.gauss(0.35, 0.15)) accuracy_sli.record_call(tool_ok) latency_sli.record_latency(latency_ms) cost_sli.record_cost(cost_usd) is_good = task_ok and tool_ok slo.record_event(good=is_good) latencies.append(latency_ms) accuracies_running.append(accuracy_sli.current_value()) costs.append(cost_usd) good_events.append(is_good) budget_remaining.append(slo.error_budget.remaining_percent) print(f"Simulated {NUM_CALLS} agent calls") print(f" Good events: {sum(good_events)} / {NUM_CALLS}") print(f" Bad events: {NUM_CALLS - sum(good_events)} / {NUM_CALLS}")这段代码演示了 agent-sre 的两类记录通道:
- SLI 通道:
record_call/record_latency/record_cost分别把单次测量写入各自 SLI 的测量存储,供后续current_value()与compliance()聚合。 - 预算通道:
slo.record_event(good=...)把"本次调用是否达标"记入ErrorBudget的有界事件缓冲,record_event内部还会调用evaluate()更新 SLO 状态(见 objectives.py)。
random.seed(42)保证结果可复现,这也是 Notebook 教学场景的关键设计:同一份随机数据既用于基线,也用于后续 what-if 重放。
五、第 3~4 步:计算 SLI 与错误预算报告
读取指标值与合规度
print("Indicator Summary") for ind in slo.indicators: val = ind.current_value() comp = ind.compliance() if val is not None and comp is not None: met = "✅" if comp >= 0.95 else "❌" print(f" {met} {ind.name}") print(f" Value: {val:.4f}") print(f" Target: {ind.target}") print(f" Compliance: {comp:.1%}")compliance()的语义见 indicators.py:窗口内满足目标的测量数占比。Notebook 用0.95作为"单指标合规度达标线",与 5% 错误预算口径一致。
评估 SLO 状态与错误预算
status = slo.evaluate() print(f" SLO Status: {status.value}") print(f" Budget Total: {slo.error_budget.total:.2%}") print(f" Budget Consumed: {slo.error_budget.consumed}") print(f" Budget Remaining: {slo.error_budget.remaining_percent:.1f}%") print(f" Exhausted? {slo.error_budget.is_exhausted}") print(f" Burn Rate (1 h): {slo.error_budget.burn_rate(3600):.1f}×") if status == SLOStatus.EXHAUSTED: print(" 🚨 Budget exhausted — action: " f"{slo.error_budget.exhaustion_action.value}") elif status in (SLOStatus.CRITICAL, SLOStatus.WARNING): print(" ⚠️ SLO at risk — consider slowing deployments") else: print(" ✅ Budget healthy — keep shipping")SLOStatus(objectives.py)共有五档:HEALTHY / WARNING / CRITICAL / EXHAUSTED / UNKNOWN。evaluate()的状态判定优先级(objectives.py)为:预算耗尽 →EXHAUSTED;有 critical 级告警 →CRITICAL;有 warning 级告警 →WARNING;无任何指标数据 →UNKNOWN;否则HEALTHY。
燃烧速率的计算(objectives.py):
burn_rate = (窗口内实际错误率) / (允许错误率 = total / window_seconds)burn_rate = 1.0表示按计划速率消耗预算;> 1.0表示消耗快于预期,例如 2× 意味着预算将在窗口期的 1/2 时间内耗尽。
六、第 5 步:可视化——错误预算、延迟分布、准确率与燃烧
Notebook 用matplotlib绘制 2×2 四联图(fig, axes = plt.subplots(2, 2, figsize=(14, 10))):
- 6a 错误预算剩余曲线:横轴为 Agent Call #,纵轴为剩余百分比,红线虚线标记
y=0(预算耗尽线),直观看到预算随"坏事件"单调下降。 - 6b 延迟分布直方图:30 个 bin 的直方图上叠加两条竖线——红色虚线为 3000 ms 目标线,橙色虚线为实际 p95 值,一眼判断 p95 是否越线。
- 6c 运行中的工具准确率:绿色曲线为累计准确率,红色虚线为 99% 目标线,
set_ylim(0.9, 1.01)放大差异区间。 - 6d 累计坏事件曲线:crimson 色折线统计累计未达标事件数,直接对应错误预算的消耗轨迹。
这四张图与 docs/slo-reference.md 中"SLO-Driven Workflows"的运营节奏呼应:周度看燃烧速率趋势、月度看预算消耗、季度回顾 SLO 定义。若需要将这种可视化固化为运维面板,仓库还提供 dashboards/grafana/agent-slo-dashboard.json 等 Grafana 模板,以及SLODashboard(src/agent_sre/slo/dashboard.py)——它支持register_slo()注册、take_snapshot()定时快照、health_summary()汇总健康度,Notebook 中的手工循环恰好是它的交互式雏形。
七、第 6 步:告警阈值检查
错误预算被快速消耗时需要尽早发现,Notebook 的告警配置如下:
| Alert | Threshold | Severity |
|---|---|---|
| Fast burn | 2× normal rate | ⚠️ Warning |
| Critical burn | 10× normal rate | 🚨 Critical |
current_burn = slo.error_budget.burn_rate(3600) print(f" Current burn rate (1 h): {current_burn:.1f}×") for alert in slo.error_budget.alerts(): firing = alert.is_firing(current_burn) print(f" {'🔔 FIRING' if firing else '🔇 ok'} {alert.name} " f"(threshold: {alert.rate:.0f}×, severity: {alert.severity})") firing_alerts = slo.error_budget.firing_alerts() print(f" → {len(firing_alerts)} alert(s) currently firing")底层实现中,alerts()返回两条BurnRateAlert(objectives.py):burn_rate_warning(2×,warning)与burn_rate_critical(10×,critical),窗口均为 86400 秒(24h);is_firing(current_burn_rate)即current_burn_rate >= rate。firing_alerts()则用当前 1h 燃烧速率过滤出正在触发的告警。
若需要多窗口告警与多渠道路由,docs/slo-reference.md 给出了扩展方案:1h/14.4×、6h/6×、24h/3×、72h/1× 的四窗口组合,以及经AlertManager将 P1/P2 路由到 PagerDuty、P3+ 路由到 Slack 的示例。
八、第 7 步:What-If 分析——延迟 +20% 对预算的影响
Notebook 最后一个实验回答一个典型容量问题:"如果延迟上涨 20%,错误预算会怎样?" 做法是复用同一随机种子重放 200 次调用,仅把每次延迟乘以1.20,再对比基线与假设场景的 p95 延迟和剩余预算:
LATENCY_INCREASE = 1.20 # +20 % wif_latency = ResponseLatency(target_ms=3000.0, percentile=0.95, window="1h") wif_accuracy = ToolCallAccuracy(target=0.99, window="24h") wif_cost = CostPerTask(target_usd=0.50, window="24h") wif_budget = ErrorBudget(total=0.05, burn_rate_alert=2.0, burn_rate_critical=10.0, exhaustion_action=ExhaustionAction.FREEZE_DEPLOYMENTS) wif_slo = SLO(name="code-review-agent-whatif", indicators=[wif_latency, wif_accuracy, wif_cost], error_budget=wif_budget, agent_id="code-review-agent") random.seed(42) for i in range(NUM_CALLS): task_ok = random.random() < 0.92 tool_ok = random.random() < 0.985 latency_ms = max(100, random.gauss(2400, 700)) * LATENCY_INCREASE cost_usd = max(0.01, random.gauss(0.35, 0.15)) wif_accuracy.record_call(tool_ok) wif_latency.record_latency(latency_ms) wif_cost.record_cost(cost_usd) wif_slo.record_event(good=task_ok and tool_ok) wif_budget_remaining.append(wif_slo.error_budget.remaining_percent)对比输出包含:基线 vs 假设的 p95 延迟、基线 vs 假设的剩余预算百分比,并绘制左右双联图(左:两条预算剩余曲线对比;右:两组延迟分布直方图叠加,红色虚线标出 3000 ms 目标)。这类"同一流量、单一变量"的重放方法可直接迁移到容量规划、模型升级预评估等场景,也是 examples/canary_rollout.py 等金丝雀发布示例的思路原型。
九、从 Notebook 到生产:下一步路径
Notebook 结尾给出的三条进阶路线,均能在仓库中找到对应落地物:
- 调整耗尽策略:把
ExhaustionAction.FREEZE_DEPLOYMENTS换成CIRCUIT_BREAK或THROTTLE观察不同反应。CIRCUIT_BREAK可配合 src/agent_sre/cascade/circuit_breaker.py 与 src/agent_sre/incidents/circuit_breaker.py 使用,为下游 agent 设置失败阈值与恢复超时,防止级联故障。 - 接入真实 Agent:使用
agent_sre.integrations.langchain.callback中的AgentSRECallback(callback.py)。该回调通过鸭子类型无侵入接入 LangChain 的callbacks=[...],自动采集任务成功率、每次 chain/LLM 调用的延迟、按 token 估算的成本与工具调用成败,Notebook 里手写的record_*循环在真实 Agent 上被它取代。 - CLI 版本:
examples/slo_alerting.py(examples/slo_alerting.py)把同一工作流压缩为脚本:pip install agent-sre && python examples/slo_alerting.py,适合在 CI 或定时任务中输出文本版 SLO 报告。
此外,若要进一步体系化:
- YAML 化 SLO:仓库 specs/slos/ 提供
base.yaml、batch_agent.yaml、critical_agent.yaml等模板,可用agent_sre.slo.spec的load_slo_specs()与resolve_inheritance()(spec.py)加载并解析继承关系,把 Notebook 中的 Python 定义固化为团队共享的配置文件。 - 单元测试佐证:Notebook 中所有 API 均有对应测试覆盖,可参考 tests/unit/test_indicators.py 与 tests/unit/test_objectives.py,了解各 SLI 聚合与预算状态的边界行为。
- 完整概念地图:SLO 设计原则、目标值选取("100% 目标是不对的")、按风险等级分层的目标表格,见 docs/slo-reference.md。
十、小结
本指南完整复刻了slo-exploration.ipynb的实验主线:从定义 p95 延迟、工具准确率、单任务成本三类 SLI 开始,用 200 次带缺陷率的模拟调用驱动错误预算消耗,再通过evaluate()状态机、燃烧速率告警与四联图可视化形成"可观测 → 可预警 → 可决策"的闭环,最后用 what-if 重放验证单一变量变化对预算的影响。这套交互式流程既是学习 agent-sre SLO 模型最快的入口,也是把 SRE 纪律引入 AI Agent 治理(策略合规、零信任身份之外)的第三个支柱——可靠性工程,让"继续上线"与"放慢节奏"不再是拍脑袋决定,而是由错误预算数据说了算。
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考