llama.cpp 安全策略全解：漏洞披露流程、覆盖范围与安全使用指南
面向 AI 工程师的 Agent SRE 实战指南：用 SLI/SLO、错误预算与故障注入守护自主智能体
项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit 本文是 Agent Governance Toolkit 中
agent-sre子项目（位于 agent-governance-python/agent-sre）配套的 SRE 概念手册的深度展开。它面向"懂 AI 但不懂 SRE"的团队，回答一个核心问题：当监控面板显示 HTTP 200、而你的 Agent 正在一本正经地编造医学剂量或放行一笔欺诈交易时，你如何量化"不可靠"、预算"可接受的失败"、并在出问题时自动熔断降级。读完本文，你将掌握七类 Agent 专属 SLI、四档 SLO 状态、错误预算与燃烧率的自动告警、混沌故障注入、金丝雀模型发布、Agent 级熔断器与信号关联式事件响应的完整落地写法。
什么是 SRE，为什么 Agent 团队必须重新学一遍
Site Reliability Engineering（SRE）是 Google 开创的一门学科：用软件工程的方法解决运维问题。它不指望系统永远在线，而是为可靠性设定可度量的目标、为"可接受的失败"做预算、并用自动化手段在出问题时及时响应。
SRE 的核心理念只有一句话：100% 可靠性是错误的目标。任何系统都会失败，真正的问题是你的用户能容忍多少失败，以及你如何把这份"失败预算"花在快速迭代上而不透支信任。
SRE 交付三样东西：
- 一套共享的可靠性语言 —— SLI、SLO、错误预算，把"应该能用"这种模糊说法变成可度量的契约；
- 自动化的护栏 —— 熔断器、金丝雀发布、混沌测试，在用户发现问题之前先发现问题；
- 事件处理纪律 —— 当故障发生时（一定会发生），用 runbook、告警和事后复盘（postmortem）快速恢复并防止复发。
在 AI Agent 场景下，这套方法论的对象从"服务器"迁移到了"智能体"：agent-sre 模块（源码位于 agent-governance-python/agent-sre/src/agent_sre）正是这一思路的工程化实现。
为什么 AI Agent 需要 SRE：五种"看不见"的故障模式
传统服务以 APM 能捕获的方式失败——崩溃、超时、500 错误。AI Agent 的失败方式完全不同，而且更危险，因为它们不可见。
Agent 会无声地失败
监控显示 "HTTP 200, all green"，而你的 Agent 刚刚幻觉出一个医疗剂量、批准了一笔欺诈交易、或编造了一条引用。"错误答案"没有堆栈跟踪。
没有错误预算
多数团队对 Agent 错误的态度是"理论上零容忍，实践上零度量"。没有错误预算，就无法在速度与可靠性之间做理性取舍——每个 bug 都变成危机，或者更糟，每个 bug 都被无视。
没有面向 AI 的 SLO
传统 SLO 度量可用性与延迟。但一个 200ms 返回幻觉答案的 Agent 并不"可靠"——它只是快而错。AI Agent 的 SLO 必须度量"决策质量"而不仅仅是可用性。 这正是 agent-sre 的 SLO 引擎 的设计出发点。
多 Agent 系统的级联失败
当 Agent A 调用 Agent B、B 再调用 C 时，C 的失败会反向级联。没有熔断器，一个不稳定的工具或一个过载的模型就能拖垮整条 Agent 工作流。这对应 OWASP ASI08 —— Cascading Agent Failures（见 docs/compliance/owasp-agentic-top10-mapping.md）。
没有护栏的成本失控
一个卡在重试循环里的 Agent 能在几分钟内烧掉数千美元的 API 调用费。没有成本护栏，一个坏提示词或死循环就能在任何人注意到之前耗尽你的月度预算。agent-sre 为此提供了 CostGuard 成本守卫。
SLI：度量 Agent 的"行为"，而不是基础设施
传统 SRE 的 SLI 度量请求延迟、错误率、吞吐量。Agent SRE 的 SLI 度量 Agent 的行为——问题不是"服务器响应了吗"，而是"Agent 给出好答案了吗"。
agent-sre 定义了七类为 AI Agent 量身打造的 SLI：
| SLI | 度量什么 | 示例 |
|---|---|---|
| Task Success Rate（任务成功率） | Agent 是否正确完成任务 | 95% 的任务成功完成 |
| Hallucination Rate（幻觉率） | Agent 多久编造一次信息 | 少于 5% 的响应包含幻觉 |
| Cost Per Task（单任务成本） | 每个 Agent 任务花费多少 | 平均成本低于 $0.50/任务 |
| Latency（延迟） | 首 token 时间与总响应时间 | p99 延迟低于 3 秒 |
| Policy Compliance（策略合规） | Agent 是否遵守治理规则 | 100% 符合安全策略 |
| Tool Call Success（工具调用成功率） | Agent 的工具调用是否成功 | 超过 98% 的工具调用返回有效结果 |
| User Satisfaction（用户满意度） | 显式或隐式用户反馈 | 平均评分 > 4.0/5.0 |
from agent_sre.slo.indicators import TaskSuccessRate, CostPerTask, HallucinationRate
# 定义你要度量什么
success = TaskSuccessRate(target=0.95, window="24h")
cost = CostPerTask(target_usd=0.50, window="24h")
hallucination = HallucinationRate(target=0.05, window="24h")
# 每个 Agent 任务结束后记录观测
success.record_task(success=True)
cost.record_cost(cost_usd=0.35)
hallucination.record_evaluation(hallucinated=False)
源码透视：SLI 的底层契约
从 SLI 基类与内置指标实现 可以看到这套 API 的完整设计：
- 时间窗口：
TimeWindow枚举定义了1h/6h/24h/7d/30d五档标准窗口（DAY_30 = "30d"对应 2,592,000 秒），任何字符串窗口都会在构造时被归一化为TimeWindow； - 测量与合规：基类
SLI通过record()把每次测量写入MeasurementStore（默认InMemoryMeasurementStore，也支持SQLiteMeasurementStore以跨重启存活），current_value()返回窗口内均值，compliance()返回窗口内达标测量的占比； - 内置指标：除了文档表格中的三类，源码还内置了
ToolCallAccuracy、ResponseLatency（支持按百分位取值，默认 p95/5000ms）、PolicyCompliance（默认 100% 目标）、DelegationChainDepth（作用域链深度，越低越好，反向比较）与CalibrationDeltaSLI（校准漂移，度量预测置信度与实际成功率的差距）； - 方向感知的合规：
HallucinationRate.compliance()与DelegationChainDepth.compliance()都覆写了基类逻辑——前者按value <= target判定"好"（越低越好），后者按value <= max_depth判定；这与 TaskSuccessRate 的"越高越好"方向相反，接入自定义 SLI 时务必注意目标方向； - 注册表：
SLIRegistry自动注册全部内置指标类型，并支持按agent_id注册实例、批量collect_all()采集，方便与观测面板对接。
单元测试见 agent-governance-python/agent-sre/tests/test_indicators.py 与 test_sli_persistence.py，可作为自定义 SLI 的参照实现。
SLO：把指标组合成可靠性契约
传统 SRE 说："99.9% 的请求必须在 200ms 内完成。"
Agent SRE 说："99.5% 的 Agent 响应必须准确，且 95% 的任务成本低于 $0.50。"
SLO 把多个 SLI 组合成一个带时间窗口的可靠性目标，回答一个问题："我的 Agent 足够可靠吗？"
from agent_sre import SLO, ErrorBudget
slo = SLO(
name="customer-support-agent",
description="Production support agent reliability targets",
indicators=[success, cost, hallucination],
error_budget=ErrorBudget(total=0.05), # 5% error budget
)
# 每个任务结束后，记录这是一个 "good" 事件
slo.record_event(good=True)
# 评估 SLO 健康状态
status = slo.evaluate() # HEALTHY | WARNING | CRITICAL | EXHAUSTED
SLO 状态机与告警联动
| 状态 | 含义 | 应采取的动作 |
|---|---|---|
| HEALTHY | 预算内，所有指标达标 | 继续部署 |
| WARNING | 预算消耗快于可持续速度 | 调查，放慢部署 |
| CRITICAL | 预算几乎耗尽 | 停止新功能开发，先修可靠性 |
| EXHAUSTED | 预算已耗尽 | 冻结部署，直到预算恢复 |
源码细节值得注意（见 objectives.py）：
SLO.__init__中有一条自动推导规则：如果未显式传入ErrorBudget，则从所有指标里最严格（最小）的 target 推导预算总量：total = 1.0 - min(sli.target)，即最严格指标决定可靠性下限；SLOStatus枚举实际包含五档：HEALTHY / WARNING / CRITICAL / EXHAUSTED / UNKNOWN，其中UNKNOWN表示窗口内尚无任何测量数据；evaluate()的判定优先级为：预算耗尽 → CRITICAL → 燃烧率告警 → 数据不足（UNKNOWN）→ HEALTHY；- 若在构造时传入
AlertManager，evaluate()会在状态恶化时发送WARNING/CRITICAL告警、在恢复时发送RESOLVED告警，并携带dedup_key去重——告警按 Agent 与 SLO 维度去重，避免状态抖动造成告警风暴。
依赖推导、状态判定与告警去重的对应测试
仓库中 test_objectives.py 覆盖了预算推导、状态迁移与告警联动；test_alert_dedup.py 与 test_alert_persistence.py 进一步验证了去重与持久化行为，可作为接入告警链路时的回归保障。
错误预算：把"可靠性"变成可花销的资源
传统 SRE 说："我们每月可容忍 0.1% 的宕机（43 分钟）。如果已经用了 40 分钟，就停止部署有风险的变更。"
Agent SRE 说："我们每月可容忍 5% 的坏响应。如果已经用掉 4.5% 的预算，就冻结模型升级与提示词变更，直到预算恢复。"
错误预算把可靠性从"好/坏"二元判断变成可花销的资源，让你做出理性权衡：
- 预算还有余额？ 上线那个新提示词模板。
- 预算耗尽了？ 专注可靠性。质量恢复之前不上新功能。
budget = ErrorBudget(
total=0.05, # 5% 错误预算（95% 可靠性目标）
burn_rate_alert=2.0, # 燃烧率 2 倍时告警
burn_rate_critical=10.0, # 燃烧率 10 倍时呼叫
)
# 查看预算状态
print(f"Budget remaining: {budget.remaining_percent:.1f}%")
print(f"Exhausted: {budget.is_exhausted}")
预算耗尽后的动作
| 动作 | 描述 |
|---|---|
ALERT | 通知团队 |
FREEZE_DEPLOYMENTS | 阻止新版本 Agent 上线 |
CIRCUIT_BREAK | 打开熔断器，缩小爆炸半径 |
THROTTLE | 降低 Agent 流量或能力 |
源码透视：错误预算的实现细节
从 ErrorBudget 实现 可以确认：
- 有界事件缓冲：事件存储在
collections.deque(maxlen=100_000)中，防止长运行 SLO 的内存无限增长；缓冲满时最旧事件被静默淘汰； - 剩余量计算：
remaining = max(0.0, 1.0 - consumed/total)，因此remaining_percent永远不会为负；is_exhausted判断consumed >= total； - ExhaustionAction 枚举正是上文表格四个动作的机器可读形式，可通过
ErrorBudget.exhaustion_action配置； - 燃烧率告警：
firing_alerts()会基于当前窗口计算燃烧率并与burn_rate_alert（默认 2.0，warning）和burn_rate_critical（默认 10.0，critical）比对，返回正在触发的告警列表。
燃烧率（Burn Rate）：预算消耗的速度计
传统 SRE 说："按当前错误率，我们会在 3 天内耗尽月度预算。"
Agent SRE 说："按当前幻觉率，我们的错误预算将在 6 小时后归零。"
燃烧率度量你消耗错误预算的速度。燃烧率 1.0 表示你恰好会在窗口结束时耗尽预算，更高则更糟。
| 燃烧率 | 含义 | 到耗尽的时间（30 天预算） |
|---|---|---|
| 1.0x | 可持续 | 30 天 |
| 2.0x | 令人担忧 | 15 天 |
| 5.0x | 糟糕 | 6 天 |
| 10.0x | 危急 | 3 天 |
| 30.0x | 宕机 | 1 天 |
# 查看最近一小时的燃烧率
rate = slo.error_budget.burn_rate(window_seconds=3600)
print(f"Burn rate (1h): {rate:.1f}x")
# 查看正在触发的告警
for alert in slo.error_budget.firing_alerts():
print(f"⚠️ {alert.name}: burn rate = {alert.rate:.1f}x")
实现要点：burn_rate() 的计算公式为 实际错误率 / 允许错误率（源码见 objectives.py#L106-L130），其中"允许错误率"由 total / window_seconds 得出；默认评估窗口为 1 小时，但告警阈值按 24 小时窗口评估（alerts() 中两个 BurnRateAlert 均使用 86400 秒窗口），这遵循了 SRE 社区"快速燃烧 + 长窗口观测"的通行实践。
混沌测试：给 Agent 的依赖注入故障
传统 SRE 说："我们随机杀死服务器，证明系统能扛住故障。"
Agent SRE 说："我们向 Agent 的依赖注入故障——慢速 LLM 响应、工具报错、上下文损坏——证明 Agent 能优雅降级。"
AI Agent 的混沌测试不止于基础设施故障，你需要测试这些场景：
- LLM 以 5 秒延迟（而不是 500ms）响应
- 工具调用返回错误或垃圾数据
- 上下文被截断或损坏
- Agent 的内存不可用
- 链中的下游 Agent 不可达
from agent_sre.chaos.engine import ChaosExperiment, Fault, AbortCondition
# 定义实验
experiment = ChaosExperiment(
name="llm-latency-spike",
description="What happens when the LLM takes 5x longer?",
duration_seconds=300,
blast_radius=0.5, # 影响 50% 的请求
abort_conditions=[
AbortCondition(metric="success_rate", threshold=0.80, comparator="lte"),
],
)
# 定义要注入的故障
latency_fault = Fault.latency_injection(
target="llm-provider",
delay_ms=5000,
rate=0.5,
)
# 运行实验
experiment.start()
experiment.inject_fault(latency_fault, applied=True, details="5s latency added")
# 评估韧性
score = experiment.calculate_resilience(
baseline_success_rate=0.95,
experiment_success_rate=0.88,
)
print(f"Resilience score: {score.score}/100 ({'PASS' if score.passed else 'FAIL'})")
源码透视：故障注入的内置原语
混沌引擎源码 提供了四类开箱即用的故障构造器：
Fault.latency_injection(target, delay_ms=5000, rate=1.0)—— 注入延迟；Fault.error_injection(target, error="internal_error", rate=1.0)—— 注入错误返回；Fault.timeout_injection(target, delay_ms=30000, rate=1.0)—— 注入超时；Fault.prompt_injection(target, technique="direct_override", rate=1.0)—— 注入对抗性提示词（对应 OWASP 提示注入场景）。
此外，chaos_scheduler.py 提供 ChaosScheduler，支持按 cron 调度实验、配置黑窗（blackout）、按严重度渐进加压，并记录每次执行的韧性趋势；adversarial.py 与 adversarial_policy.py 则将混沌测试扩展到对抗性攻击剧本（AttackPlaybook）与对抗策略评估。可运行示例见 examples/chaos_test.py，配套测试在 tests/test_chaos.py 与 tests/test_chaos_scheduler.py。
金丝雀发布：先让 5% 的真实流量试新模型
传统 SRE 说："把 5% 的流量路由到新版本服务器。如果错误率飙升，回滚。"
Agent SRE 说："把 5% 的流量路由到新模型/新提示词。如果幻觉率上升或准确率下降，自动回滚。"
金丝雀发布对 AI Agent 至关重要，因为模型升级和提示词变更可能带来不可预测的回归——一个在基准测试上表现更好的模型，可能在你特定的工作负载上表现更差。
from agent_sre.delivery.rollout import CanaryRollout, RollbackCondition, RolloutStep
rollout = CanaryRollout(
name="assistant-v2",
steps=[
RolloutStep(weight=0.05, duration_seconds=60, name="5% canary"),
RolloutStep(weight=0.25, duration_seconds=120, name="25% ramp"),
RolloutStep(weight=1.0, duration_seconds=0, name="full rollout"),
],
rollback_conditions=[
RollbackCondition(metric="error_rate", threshold=0.10, comparator="gte"),
RollbackCondition(metric="hallucination_rate", threshold=0.08, comparator="gte"),
],
)
rollout.start()
# 每个阶段评估候选版本
metrics = evaluate_new_version()
if rollout.check_rollback(metrics):
rollout.rollback(reason="Hallucination rate exceeded threshold")
else:
rollout.advance()
发布策略的工程支撑
delivery 模块 不只提供金丝雀：rollout.py 定义了 DeploymentStrategy（部署策略枚举）与 RolloutState 状态机；blue_green.py 提供蓝绿发布，支持 deploy → validate(health_check_fn) → switch → rollback 的完整生命周期；gitops.py 则提供声明式 RolloutSpec 校验与 default_canary / default_shadow 模板生成。这些机制共同覆盖"模型/提示词"这一新型部署物件的灰度与回滚诉求。示例与测试分别见 examples/canary_rollout.py 与 tests/test_delivery.py、tests/test_blue_green.py。
熔断器：阻止多 Agent 系统的级联雪崩
传统 SRE 说："如果下游服务失败 5 次，停止调用它 30 秒。"
Agent SRE 说："如果 Agent 失败 5 次，停止向它路由任务并使用兜底方案。"
熔断器阻止多 Agent 系统中的级联失败。当 Agent A 依赖 Agent B、而 B 开始失败时，熔断器阻止 A 继续向 B 发送注定失败的请求，转而返回兜底响应。
三种状态：
CLOSED (正常) → 失败超过阈值 → OPEN (阻断)
↓
超时到期
↓
HALF_OPEN (试探)
↓ ↓
成功 失败
↓ ↓
CLOSED OPEN
from agent_sre.cascade.breaker import CircuitBreaker, CircuitBreakerConfig
breaker = CircuitBreaker(
agent_id="research-agent",
config=CircuitBreakerConfig(
failure_threshold=5, # 5 次失败后打开
recovery_timeout_seconds=30, # 30 秒后再试
half_open_max_calls=1, # 允许 1 次试探调用
),
)
# 用熔断器包裹 Agent 调用
result = breaker.call(
func=research_agent.run,
query="summarize this paper",
fallback="Unable to process request. Please try again later.",
)
# 熔断器自动：
# - 跟踪成功与失败
# - 连续 5 次失败后打开熔断
# - 打开时抛出 CircuitOpenError（或返回兜底）
# - 30 秒后放行一次试探调用检查恢复
源码透视：状态机、同步/异步与级联检测
CircuitBreaker 实现 有几个值得注意的工程细节：
- 配置兼容：
CircuitBreakerConfig同时接受规范名recovery_timeout_seconds与旧别名reset_timeout_seconds（兼容agent_os.circuit_breaker旧 API），两者不一致时抛ValueError； - 状态机：
CircuitState为CLOSED / OPEN / HALF_OPEN；_maybe_transition_to_half_open()在 OPEN 状态下按recovery_timeout_seconds到期自动转 HALF_OPEN；HALF_OPEN 下成功即回 CLOSED，失败立即回 OPEN； - 同步/异步双支持：
call()对返回 awaitable 的结果会包装为协程，并自动在协程成功/失败时调用record_success()/record_failure()，因此breaker.call(awaitable_fn)在异步调用方中可直接await； - 兜底语义：OPEN 状态且传入
fallback时直接返回兜底值，否则抛出CircuitOpenError（内含agent_id与retry_after，便于上层降级提示）； - 级联检测：同一文件还提供
CascadeDetector，可注册多个 Agent 的熔断器，当处于 OPEN 的 Agent 数达到cascade_threshold（默认 3）时判定发生级联失败（对应 OWASP ASI08）。
测试见 tests/test_circuit_breaker.py 与 tests/test_cascade.py。
事件管理：把多个信号关联成一次事故
传统 SRE 说："PagerDuty 告警 → 值班工程师 → Runbook → 事后复盘。"
Agent SRE 说："检测到 SLO 违约 → 信号关联 → 创建事件 → 自动执行 runbook → 通知团队。"
Agent SRE 通过关联多个信号来检测事件——一次 SLO 违约 + 一次成本飙升 + 一次延迟升高，很可能是一次事故而不是三个独立告警。
from agent_sre.incidents.detector import IncidentDetector, Signal, SignalType
detector = IncidentDetector(correlation_window_seconds=60)
# 从不同来源接入信号
signal = Signal(
signal_type=SignalType.SLO_BREACH,
source="customer-support-agent",
message="Task success rate dropped below 95% target",
)
incident = detector.ingest_signal(signal)
if incident:
print(f"🚨 Incident: {incident.title}")
print(f" Severity: {incident.severity.value}") # P1, P2, P3, P4
incident.acknowledge()
# ... 执行修复 ...
incident.resolve()
Agent SRE 监控的信号类型
| 信号 | 触发条件 |
|---|---|
SLO_BREACH | 违反 SLO 目标 |
ERROR_BUDGET_EXHAUSTED | 错误预算完全耗尽 |
COST_ANOMALY | 花费超过 z-score 阈值 |
LATENCY_SPIKE | 响应时间超过基线 |
TOOL_FAILURE_SPIKE | 工具调用失败率飙升 |
POLICY_VIOLATION | Agent 违反治理策略 |
TRUST_REVOCATION | Agent 信任分下降（经 AgentMesh） |
源码透视：事件检测器的信号关联逻辑
从 detector.py 的实现可以看到：
- 信号到严重度的自动映射：
Signal.severity_hint将ERROR_BUDGET_EXHAUSTED / POLICY_VIOLATION / TRUST_REVOCATION映射为 P1，SLO_BREACH / COST_ANOMALY / LATENCY_SPIKE映射为 P2，其余为 P3； - 事件只对 P1/P2 信号创建：P3/P4 信号仅记录，不建事件，避免噪音；
- 去重与关联：
dedup_window_seconds（默认 600s）内同源同类型信号不重复建事件；correlation_window_seconds（默认 300s）内来自同一来源的不同类型信号会被合并为"Correlated"事件，取最高严重度、自动应用所有信号类型注册的响应动作（通过register_response(signal_type, actions)注册，事件创建时自动执行并标记auto-triggered）； - 事件生命周期：
Incident内置DETECTED → ACKNOWLEDGED → INVESTIGATING → MITIGATING → RESOLVED状态机，duration_seconds自动统计从检测到解决的时间。
测试见 tests/test_incidents.py；事件运行手册（runbook）相关支持可查看 src/agent_sre/incidents 目录下的其他模块与 tests/test_runbooks.py。
传统 SRE → Agent SRE 映射表
| 传统 SRE | Agent SRE | 为什么不同 |
|---|---|---|
| Uptime（服务器在跑吗？） | Response quality（答案对吗？） | 返回 200 但内容幻觉不算"在线" |
| Request latency（p50, p99） | Token latency（首 token 时间、总生成时间） | LLM 推理与 CRUD API 的延迟画像不同 |
| Error rate（5xx 响应） | Hallucination rate（编造信息） | "编造一条假引用"没有 HTTP 状态码 |
| Throughput（请求/秒） | Task completion rate（任务/小时） | Agent 任务是多步骤的，不是单次请求-响应 |
| CPU/Memory | Token usage / Cost per task | 昂贵资源是 LLM 推理而非算力 |
| Deployment rollback | Model/prompt rollback | 回滚的是提示词变更，不是二进制产物 |
| Circuit breaker（服务→服务） | Circuit breaker（Agent→Agent） | Agent 失败是语义性的，不只是超时 |
| Canary deployment（5% 服务器） | Canary deployment（5% 流量到新模型） | 全量上线前先用真实流量试新模型 |
| Chaos testing（杀一台服务器） | Chaos testing（注入 LLM 延迟、工具错误） | 测试 Agent 对依赖故障的韧性 |
| Runbook（重启服务、扩容） | Runbook（切换模型、禁用工具、限流） | 修复动作发生在模型/提示词层，而非基础设施层 |
| SLO（99.9% 可用性） | SLO（95% 任务成功、<5% 幻觉） | 多维：质量 + 成本 + 安全 |
| Error budget（每月 43 分钟宕机） | Error budget（每月 5% 坏响应） | 预算花在坏答案上，不是宕机上 |
| On-call rotation | Agent-aware on-call | 告警携带 Agent 上下文、追踪与决策日志 |
快速上手：30 行代码跑通完整 SRE 闭环
下面是一个完整示例——定义 SLO、设置成本护栏、检测事件，全部在 30 行内完成：
"""Monitor an AI agent with SRE best practices."""
from agent_sre import SLO, ErrorBudget
from agent_sre.slo.indicators import TaskSuccessRate, CostPerTask, HallucinationRate
from agent_sre.cost.guard import CostGuard
from agent_sre.cascade.breaker import CircuitBreaker, CircuitBreakerConfig
from agent_sre.incidents.detector import IncidentDetector, Signal, SignalType
# ── 1. 定义"可靠"意味着什么 ────────────────────────────────────
success = TaskSuccessRate(target=0.95, window="24h")
cost = CostPerTask(target_usd=0.50, window="24h")
hallucination = HallucinationRate(target=0.05, window="24h")
slo = SLO(
name="my-agent",
indicators=[success, cost, hallucination],
error_budget=ErrorBudget(total=0.05, burn_rate_critical=10.0),
)
# ── 2. 设置成本护栏 ─────────────────────────────────────────────
guard = CostGuard(
per_task_limit=2.00,
per_agent_daily_limit=50.0,
org_monthly_budget=1000.0,
)
# ── 3. 为下游 Agent 添加熔断器 ──────────────────────────────────
breaker = CircuitBreaker(
agent_id="my-agent",
config=CircuitBreakerConfig(failure_threshold=5, recovery_timeout_seconds=30),
)
# ── 4. 接入事件检测 ─────────────────────────────────────────────
detector = IncidentDetector(correlation_window_seconds=60)
# ── 5. 带着 SRE 护栏运行你的 Agent ──────────────────────────────
def run_agent_task(query: str):
# 检查成本预算
allowed, reason = guard.check_task("my-agent", estimated_cost=0.50)
if not allowed:
return f"Blocked: {reason}"
# 通过熔断器执行
result = breaker.call(
func=your_agent.run,
query=query,
fallback="Service temporarily unavailable.",
)
# 记录 SLI 观测
is_good = evaluate_response(result)
success.record_task(success=is_good)
slo.record_event(good=is_good)
guard.record_cost("my-agent", "task-1", cost_usd=0.35)
# 检查 SLO 健康状态
status = slo.evaluate()
if status.value == "EXHAUSTED":
signal = Signal(
signal_type=SignalType.ERROR_BUDGET_EXHAUSTED,
source="my-agent",
message="Error budget consumed — freeze deployments",
)
detector.ingest_signal(signal)
return result
安装与运行
# 在 agent-governance-python/agent-sre 目录下（仓库内源码安装）
pip install -e ".[dev]"
python examples/quickstart.py
仓库内的可运行版本见 examples/quickstart.py，它模拟 100 次任务，刻意让成功率低于 95% 目标、幻觉率高于 5% 目标，演示 SLO 状态评估、预算剩余、燃烧率、成本护栏与事件创建的完整输出。
包发布与源码布局说明
需要说明的是：agent-sre 已随 Agent Governance Toolkit 的包整合迁移。当前 agent-sre/pyproject.toml 是一个弃用占位包（dep-only deprecation stub），发布时只依赖并重定向到 agent-governance-toolkit-cli；agent_sre 顶层包源码由 agent-governance-toolkit-cli 在构建时通过 force-include 从 ../agent-sre/src/agent_sre 合入。因此，在仓库内开发测试请使用 pip install -e（可编辑安装）方式直接引用源码树；发布安装则统一走 agent-governance-toolkit-cli，其提供的 agent-sre 控制台入口指向 agent_sre.cli.main:cli，并支持 docker / hyperlight / policy / mcp / api / otel / langchain / langfuse 等丰富的可选依赖组。
成本护栏的实战深化：CostGuard 的预算层级与异常检测
快速上手示例里只展示了 CostGuard 的基本用法，这里补充其完整能力（源码见 guard.py）：
- 三层预算：单任务上限（
per_task_limit，默认 $2.0）、单 Agent 日限额（per_agent_daily_limit，默认 $100）、组织月预算（org_monthly_budget，默认 $5000），check_task()返回(allowed, reason)元组，reason 明确说明被拦截原因； - 原子化"检查并扣费"：文档示例使用的
check_task()是咨询性检查（不预留预算，并发场景可能超限）；对预算敏感路径应改用check_and_charge()，它在同一把锁内完成"检查 + 记账"，杜绝并发穿透； - 阈值告警与自动限流/熔断：默认在日预算消耗 50% / 75% / 90% / 95% 时告警（
alert_thresholds可自定义）；auto_throttle=True时，利用率 ≥85% 自动THROTTLE，达到kill_switch_threshold（默认 0.95）自动KILL；组织月预算同样触发全局 kill switch，一次性熔断所有 Agent； - Z-score 异常检测：
anomaly_detection=True（默认开启）时，基于最近 1000 条成本记录的历史，对 z-score > 2.0 的单笔成本触发COST_ANOMALY告警——这正是事件检测信号表中COST_ANOMALY的底层来源； - 输入防护：所有数值参数在构造时校验
isfinite与非负，kill_switch_threshold与各阈值必须落在 [0,1]，防止 NaN/Inf/负数破坏预算状态。
配套测试见 tests/test_cost.py 与 tests/test_cost_hardened.py。
从概念到生产：进一步阅读
SRE 基础
- Google SRE Book —— Site Reliability Engineering 的权威指南
- Google SRE Workbook —— 实践案例与练习
- The Art of SLOs —— 如何设定有意义的 SLO
Agent SRE 文档（仓库内）
- getting-started.md —— 安装并定义你的第一个 SLO
- concepts.md —— Agent SLI 与基础设施指标的区别
- slo-reference.md —— 完整 SLO 配置指南与模板
- integration-guide.md —— 与 LangChain、CrewAI、LangGraph 等框架集成
- deployment.md —— 生产部署模式
- security.md —— Agent 可靠性的安全考量
示例代码（仓库内）
- examples/quickstart.py —— 30 行完整可运行示例
- examples/canary_rollout.py —— 自动回滚的安全模型升级
- examples/cost_guard.py —— 预算强制与异常检测
- examples/chaos_test.py —— 面向 Agent 依赖的故障注入
- examples/slo_alerting.py —— 多渠道告警路由
生态关联
- agent-governance-python/agent-mesh —— 多 Agent 系统的身份与信任（提供
TRUST_REVOCATION信号来源） - agent-governance-python/agent-os —— AI Agent 治理内核
- docs/compliance/owasp-agentic-top10-mapping.md —— Agent SRE 如何对应 OWASP Agentic Top 10（含 ASI08 级联失败）
说明：本文引用的外部 Google SRE 链接均出自原文档"Further Reading"一节，仅作背景阅读指引；仓库内的实现与示例请以本文给出的相对路径为准。
北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台
更多推荐
24
0- 0
已为社区贡献12条内容
所有评论(0)