Agent SRE 实战指南用 SLO、熔断器、混沌测试与成本护栏构建可靠的 AI Agentagent-governance-toolkit / agent-sre【免费下载链接】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-toolkitSite Reliability EngineeringSRE是让传统服务在生产环境长期稳定运行的核心方法论而 AI Agent 的失败模式与传统服务截然不同——无限循环、行为漂移、成本爆炸、失控行为与级联故障都需要新的检测与防御手段。本文以 agent-governance-toolkit 仓库中的agent-sre包pip install agent-sre位于 agent-governance-python/agent-sre为对象系统讲解五个核心能力异常 Agent 检测OWASP ASI-10、熔断器模式、SLO 与错误预算、混沌测试、成本控制并给出一个生产级的 SRE 流水线完整代码。读完本文你将能够为自主 AI Agent 搭建一整套检测—隔离—量化—演练—控费的可靠性工程体系。安装方式pip install agent-sre需要 Python 3.10完整教程约 30 分钟。关联阅读部署指南 · OWASP Agentic Top 10 风险映射Windows 用户提示若终端出现乱码先执行chcp 65001或直接使用默认支持 UTF-8 的 Windows Terminal / VS Code 终端。为什么 AI Agent 需要 SRE传统服务的失败是可预测的——超时、崩溃、资源耗尽。Agent 则引入了全新的失败模式失败模式传统服务AI Agent无限循环进程挂起Agent 无限循环调用工具烧光 Token行为漂移新部署引入 Bug模型更新后决策模式悄然改变成本爆炸资源泄漏单个任务消耗 $500 API 调用费失控行为服务被攻破Agent 使用未授权工具或外泄数据级联失败依赖服务宕机Agent A 失败 → Agent B 重试 → Agent C 过载agent-sre提供了应对上述所有场景的构建模块。从 README 的 OWASP 覆盖声明可以看到它直接针对 OWASP Agentic Security Initiative 的ASI08级联 Agent 失败、ASI07失控成本、ASI09缺乏可观测性与ASI10测试不足四大风险分别以熔断器与错误预算、成本限制与杀停开关、7 类 SLI 与 OpenTelemetry 导出、混沌工程与渐进式发布作为应对方案。快速上手异常 Agent 检测Rogue DetectionRogueAgentDetector对应 OWASP ASI-10综合三种信号标记被攻破或异常的 Agent工具调用频率Tool-call frequency——基于滑动窗口的 z-score 尖峰检测动作熵Action entropy——同时标记可疑地重复与异常混乱两类行为能力违规Capability violations——使用了 Agent 允许画像之外的工具。基础配置from agent_sre.anomaly import RogueAgentDetector, RogueDetectorConfig, RiskLevel # 配置检测阈值 config RogueDetectorConfig( frequency_window_seconds60.0, # 频率分析的滑动窗口 frequency_z_threshold2.5, # 超过多少个标准差开始标记 entropy_low_threshold0.3, # 过于重复可能是循环 entropy_high_threshold3.5, # 过于混乱可能是被攻破 quarantine_risk_levelRiskLevel.HIGH, # 达到该风险等级自动建议隔离 ) detector RogueAgentDetector(configconfig)从 rogue_detector.py 的源码可以看到RogueDetectorConfig还包含frequency_min_windows默认 5频率分析最少需要的历史窗口数、entropy_min_actions默认 10熵计算最少需要的动作样本数和capability_violation_weight默认 1.0能力违规得分权重三个补充参数它们决定了检测器在数据不足时的行为——样本不足时相应得分为 0。声明每个 Agent 允许做什么# 为每个 Agent 注册允许使用的工具 detector.register_capability_profile( agent_idsupport-agent, allowed_tools[search_kb, create_ticket, send_email], ) detector.register_capability_profile( agent_idcode-reviewer, allowed_tools[read_file, search_code, post_comment], )喂入动作并评估风险import time # 模拟正常行为 for i in range(20): detector.record_action( agent_idsupport-agent, actionsearch, tool_namesearch_kb, timestamptime.time() i, ) # 评估风险 assessment detector.assess(support-agent) print(fRisk: {assessment.risk_level.value}) # low print(fComposite score: {assessment.composite_score}) # ~0.0 print(fQuarantine? {assessment.quarantine_recommended}) # False检测被攻破的 Agent# Agent 开始快速使用未授权工具 base time.time() 100 for i in range(50): detector.record_action( agent_idsupport-agent, actionexfiltrate, tool_nameshell_exec, # 不在允许工具列表中 timestampbase i * 0.5, # 每秒 2 次调用频率尖峰 ) assessment detector.assess(support-agent, timestampbase 25) print(fRisk: {assessment.risk_level.value}) # high 或 critical print(fFrequency score: {assessment.frequency_score}) # 升高 print(fCapability score: {assessment.capability_score}) # 0违规 print(fQuarantine? {assessment.quarantine_recommended}) # True if assessment.quarantine_recommended: print(f⚠ QUARANTINE agent {assessment.agent_id})源码层面的三信号原理理解三个信号的底层实现有助于你调准阈值频率信号z-score 尖峰ToolCallFrequencyAnalyzer 按时间桶维护每个 Agent 的调用计数用历史桶的均值与标准差计算当前窗口的 z-scoreabs(current_count - mean) / std_dev超过frequency_z_threshold即告警当标准差为 0 或历史窗口不足时返回 0避免冷启动误报。熵信号动作多样性ActionEntropyScorer 基于 Shannon 熵计算动作分布熵低于entropy_low_threshold判定为陷入循环高于entropy_high_threshold判定为行为混乱并分别按偏离程度归一化为异常分。能力违规信号CapabilityProfileDeviation 记录每个 Agent 的总调用数与违规数得分为violations / total * violation_weight取值在 [0, 1]。三者得分直接相加得到composite_score然后按 _classify_risk 映射风险等级 1.0 → LOW、 2.0 → MEDIUM、 3.0 → HIGH、 3.0 → CRITICAL。当风险等级不低于quarantine_risk_level时quarantine_recommended为True。值得特别注意的是assess()返回的每条RogueAssessment都带 SHA-256 哈希链字段previous_hash/entry_hashverify_assessment_chain 可逐条校验链路完整性——这意味着你的风险评估记录具备防篡改能力可以作为审计证据使用。熔断器模式Circuit Breaker当某个 Agent 开始持续失败时你不希望它继续冲击下游服务。熔断器会自动隔离失败的 AgentCLOSED ──(failures threshold)──→ OPEN ──(timeout elapsed)──→ HALF_OPEN ↑ │ └──────────(success)────────────────────────────────────────────←─┘ (failure) ──→ OPEN使用级联熔断器from agent_sre.cascade.circuit_breaker import ( CircuitBreaker, CircuitBreakerConfig, CircuitOpenError, ) # 配置失败 3 次后打开30 秒后试探恢复 config CircuitBreakerConfig( failure_threshold3, recovery_timeout_seconds30.0, half_open_max_calls1, ) breaker CircuitBreaker(agent_iddata-analyst, configconfig)包装 Agent 调用def run_agent_task(task: dict) - str: 你的 Agent 主函数。 # ... agent 逻辑 ... return result # 熔断器包装调用 try: result breaker.call(run_agent_task, {query: revenue Q3}) print(fResult: {result}) except CircuitOpenError as e: print(fAgent isolated: {e}) print(fRetry after: {e.retry_after:.0f}s)手动失败跟踪# 如果调用由你自己管理 try: result run_agent_task(task) breaker.record_success() except Exception: breaker.record_failure() raise # 检查状态 print(fState: {breaker.state}) # CLOSED、OPEN 或 HALF_OPEN print(fFailures: {breaker.failure_count}) # 手动重置例如部署修复后 breaker.reset()为多个 Agent 管理熔断器class AgentFleetBreakers: 为一个 Agent 集群管理熔断器。 def __init__(self, config: CircuitBreakerConfig | None None): self._config config or CircuitBreakerConfig() self._breakers: dict[str, CircuitBreaker] {} def get(self, agent_id: str) - CircuitBreaker: if agent_id not in self._breakers: self._breakers[agent_id] CircuitBreaker(agent_id, self._config) return self._breakers[agent_id] def open_circuits(self) - list[str]: return [aid for aid, cb in self._breakers.items() if cb.state OPEN] fleet AgentFleetBreakers( configCircuitBreakerConfig(failure_threshold5, recovery_timeout_seconds60.0), ) # 使用每个 Agent 独立的熔断器 cb fleet.get(summarizer-agent) cb.record_failure() print(fOpen circuits: {fleet.open_circuits()})源码层面的状态机与级联检测circuit_breaker.py 完整实现了经典三态状态机CLOSED正常放行。每次失败record_failure()递增计数达到failure_threshold默认 5后切换到 OPENrecord_failure。OPEN拒绝所有调用并抛出携带retry_after的CircuitOpenErrorget_state()会在recovery_timeout_seconds默认 30 秒超时后自动转为 HALF_OPEN 试探get_state。HALF_OPEN只允许放行half_open_max_calls默认 1个试探调用试探成功立即回到 CLOSED试探失败立刻重新 OPEN。call()同时支持同步函数与await异步结果还支持fallback参数——电路打开时返回降级值而不是抛异常call。此外同文件还提供了CascadeDetectorcascade_detector 相关实现它持有一组 Agent 的熔断器check_cascade()在 OPEN 状态的 Agent 数达到cascade_threshold默认 3时返回True用于识别级联失败正在扩散。CircuitBreakerConfig还兼容reset_timeout_seconds旧别名并会在两者同时给出且不一致时抛出ValueError方便从早期agent_os.circuit_breakerAPI 平滑迁移。SLO 跟踪用由错误预算与燃烧速率告警支撑的 Service Level Objectives 来定义你的 Agent 到底多可靠。可用的 SLI 类型SLI 类型衡量内容示例目标Latency任务完成时间p99 10sError rate失败任务占比 1%Cost单任务花费 $0.50/任务Token usage每次完成的 Token 数 4096Hallucination事实准确性得分 95%Tool success工具调用成功率 99%Human feedback用户满意度得分 4.0/5.0在 indicators.py 中这些能力被实现为可直接实例化的内建 SLI 类TaskSuccessRate默认目标 0.995、窗口 30d、ToolCallAccuracy默认 0.999、窗口 7d、ResponseLatency默认 p95、5000ms、窗口 1h、CostPerTask默认 $0.50、窗口 24h、PolicyCompliance默认 100%、窗口 24h、DelegationChainDepth默认最大 3 跳、窗口 24h、HallucinationRate默认 0.05、窗口 24h以及新增的CalibrationDeltaSLI校准漂移衡量预测置信度与实际成功率的差距。每个 SLI 通过record_task/record_call/record_latency/record_cost/record_check/record_evaluation等专属方法记录原始事件再通过collect()生成SLIValue。窗口枚举TimeWindow支持1h、6h、24h、7d、30d五种粒度。定义一个 SLI 和 SLOfrom agent_sre.slo import SLI, SLIValue, SLO, ErrorBudget from agent_sre.slo.indicators import TimeWindow from agent_sre.slo.objectives import BurnRateAlert, ExhaustionAction # 通过子类化创建具体 SLI class TaskSuccessRateSLI(SLI): 跟踪任务成功率。 def collect(self) - SLIValue: values self.values_in_window() if not values: return self.record(1.0) good sum(1 for v in values if v.is_good) return self.record(good / len(values)) # 24 小时窗口内 99.5% 成功率目标 success_sli TaskSuccessRateSLI( nametask_success_rate, target0.995, window24h, ) # 创建带错误预算的 SLO slo SLO( namecode-reviewer-reliability, indicators[success_sli], error_budgetErrorBudget( total0.005, # 0.5% 错误预算1 - 0.995 window_seconds2_592_000, # 30 天窗口 burn_rate_alert2.0, # 2 倍燃烧速率时告警 burn_rate_critical10.0, # 10 倍燃烧速率时严重告警 exhaustion_actionExhaustionAction.THROTTLE, # 预算耗尽后自动限流 ), agent_idcode-reviewer, )从源码看ErrorBudgetobjectives.py还有几个值得了解的细节事件存储在deque(maxlenmax_events)默认 100,000的有界缓冲区中长期运行的 SLO 不会无界增长内存最旧事件会被静默淘汰若你未显式传入totalSLO.__init__会用所有 SLI 目标中最严格的一个自动计算total 1.0 - min(sli.target)ExhaustionAction有四个选项ALERT仅告警、FREEZE_DEPLOYMENTS冻结发布、CIRCUIT_BREAK联动熔断器、THROTTLE限流。记录事件并检查状态# 实时记录结果 for _ in range(95): slo.error_budget.record_event(goodTrue) for _ in range(5): slo.error_budget.record_event(goodFalse) # 检查错误预算 budget slo.error_budget print(fBudget remaining: {budget.remaining_percent:.1f}%) print(fExhausted: {budget.is_exhausted}) # 检查燃烧速率预算消耗是否过快 burn_rate budget.burn_rate(window_seconds3600) # 最近一小时 print(f1h burn rate: {burn_rate:.2f}x) # 检查触发的告警 for alert in budget.firing_alerts(): print(f {alert.name}: burn rate {burn_rate:.1f}x (threshold: {alert.rate}x))燃烧速率的含义burn_rate 实际错误率 / 允许错误率等于 1.0 表示按预期速度消耗预算大于 1.0 表示消耗过快。burn_rate 默认以 1 小时为窗口计算窗口内无事件时返回 0.0。SLO 状态上报# 序列化用于仪表盘 / 告警 status slo.error_budget.to_dict() # { # total: 0.005, # consumed: 5.0, # remaining_percent: ..., # is_exhausted: False, # burn_rate: ..., # exhaustion_action: throttle, # firing_alerts: [burn_rate_critical] # }SLO.evaluate()会综合错误预算与燃烧告警给出五档健康状态HEALTHY、WARNING、CRITICAL、EXHAUSTED、UNKNOWN数据不足。如果构造时传入了alert_managerSLO 状态恶化时会自动发出带去重键的告警恢复健康时也会发出 RESOLVED 恢复通知evaluate 与 _maybe_fire_alert。混沌测试在生产事故替你发现薄弱点之前主动向 Agent 管线注入故障来验证其韧性。定义一个实验from agent_sre.chaos import ( ChaosExperiment, Fault, FaultType, AbortCondition, ResilienceScore, ) # 创建要注入的故障 faults [ Fault.latency_injection(openai-api, delay_ms5000, rate0.3), Fault.error_injection(search_tool, errortimeout, rate0.1), Fault.timeout_injection(database, delay_ms30000, rate0.05), ] # 安全兜底成功率低于 50% 时中止 abort_conditions [ AbortCondition(metricsuccess_rate, threshold0.5, comparatorlte), ] experiment ChaosExperiment( namellm-latency-resilience, target_agentcode-reviewer, faultsfaults, duration_seconds1800, # 30 分钟 abort_conditionsabort_conditions, blast_radius0.3, # 影响 30% 流量 description验证 code-reviewer 能优雅应对 LLM 延迟, )运行实验# 启动实验 experiment.start() print(fState: {experiment.state.value}) # running # 在你的 Agent 中间件中注入故障 for fault in experiment.faults: experiment.inject_fault(fault, appliedTrue) # 周期性检查中止条件 metrics {success_rate: 0.85, latency_p99: 8500} if experiment.check_abort(metrics): print(fAborted: {experiment.abort_reason}) else: # 实验正常完成 score experiment.calculate_resilience( baseline_success_rate0.99, experiment_success_rate0.85, ) experiment.complete(resiliencescore) print(fResilience: {experiment.resilience.overall:.0f}/100) print(fPassed: {experiment.resilience.passed})对抗性混沌测试用对抗性故障类型检验 Agent 的安全边界# 安全类故障 security_faults [ Fault.prompt_injection(code-reviewer, techniquedirect_override), Fault.privilege_escalation(code-reviewer, target_roleadmin), Fault.tool_abuse(code-reviewer, tool_nameshell_exec), ] security_experiment ChaosExperiment( namesecurity-boundary-test, target_agentcode-reviewer, faultssecurity_faults, duration_seconds600, description验证 Agent 能拒绝对抗性输入, )使用混沌模板库from agent_sre.chaos import ChaosLibrary, ExperimentTemplate library ChaosLibrary() # 列出可用模板 for template in library.list_templates(): print(f {template.name}: {template.description}) # 序列化实验结果用于报告 report experiment.to_dict() # 包含experiment_id、state、faults、injection_count、resilience scores源码层面的故障类型与模板engine.py 中FaultType枚举定义了三大类故障常规注入LATENCY_INJECTION、ERROR_INJECTION、TIMEOUT_INJECTION、对抗性故障PROMPT_INJECTION、POLICY_BYPASS、PRIVILEGE_ESCALATION、DATA_EXFILTRATION、TOOL_ABUSE、IDENTITY_SPOOFING和行为故障DEADLOCK_INJECTION、CONTRADICTORY_INSTRUCTION、TRUST_PERTURBATION。Fault类为每种类型提供了工厂方法每个Fault由fault_type、target工具名 / Agent ID / 提供商名与rate0.0~1.0受影响的调用比例三元组构成。ChaosExperiment维护一个五态状态机PENDING → RUNNING → COMPLETED / ABORTED / FAILEDblast_radius在构造时被钳制到 [0, 1]。AbortCondition支持lte/gte两种比较器用于在指标越界时紧急中止实验。calculate_resilience 的判定逻辑是experiment_success_rate baseline_success_rate * 0.9即通过overall得分 实验成功率 / 基线成功率 × 100钳制在 0~100。library.py 内置了 11 个可直接实例化的实验模板例如timeout-injection工具超时中危、error-injection错误风暴高危blas radius 0.5、latency-injectionLLM 提供商延迟尖峰、adversarial-injection提示注入、adversarial-escalation权限提升严重、adversarial-exfiltration数据外泄、deadlock-injectionAgent 间死锁、contradictory-instruction矛盾指令、trust-perturbation信任分扰动、delegation-reject委托拒绝与credential-expiry凭据过期。模板可通过library.get(template_id)获取用template.instantiate(target_agent, **overrides)实例化为具体实验也可以按category/severity/tag过滤。成本控制用单任务预算、自动限流和杀停开关阻止失控支出。设置成本护栏from agent_sre.cost import CostGuard, AgentBudget, BudgetAction guard CostGuard( per_task_limit2.00, # 单任务最高 $2 per_agent_daily_limit50.00, # 每 Agent 每天最高 $50 org_monthly_budget5000.00, # 组织级月度上限 auto_throttleTrue, # 达到日预算 85% 时限流 kill_switch_threshold0.95, # 达到日预算 95% 时杀停 Agent anomaly_detectionTrue, # 检测成本尖峰 )源码层面CostGuard.__init__会对所有数值参数做严格校验必须有限且非负、kill_switch_threshold必须在 [0,1]非法值直接抛ValueError避免 NaN/Inf/负数污染状态guard.py。默认告警阈值档位为[0.50, 0.75, 0.90, 0.95]跨过 ≥0.90 的档位按 CRITICAL 告警其余按 WARNING。预检# 运行昂贵任务前先检查预算是否允许 allowed, reason guard.check_task(research-agent, estimated_cost1.50) if not allowed: print(fBlocked: {reason}) else: # 执行任务... pass需要注意check_task是咨询性检查不会预留预算两个并发调用方可能同时通过检查并超额。对于预算敏感的路径应使用check_and_charge(agent_id, task_id, cost_usd, ...)——它在单把锁内原子完成检查 记账从并发角度保证预算约束真实生效。check_task的拒绝原因非常具体组织预算耗尽、Agent 已被杀停、Agent 被限流、预估成本超过单任务上限、会超出日预算、会超出组织月度预算。记录成本并处理告警# 每个任务结束后记录实际成本 alerts guard.record_cost( agent_idresearch-agent, task_idtask-001, cost_usd0.45, breakdown{input_tokens: 0.15, output_tokens: 0.25, tool_calls: 0.05}, ) for alert in alerts: print(f[{alert.severity.value}] {alert.message}) if alert.action BudgetAction.KILL: print( Agent killed — stop all tasks immediately) elif alert.action BudgetAction.THROTTLE: print(⚠ Agent throttled — reduce task rate)监控预算利用率budget guard.get_budget(research-agent) print(fSpent today: ${budget.spent_today_usd:.2f}) print(fRemaining: ${budget.remaining_today_usd:.2f}) print(fUtilization: {budget.utilization_percent:.0f}%) print(fAvg/task: ${budget.avg_cost_per_task:.4f}) print(fThrottled: {budget.throttled}) print(fKilled: {budget.killed})AgentBudget提供remaining_today_usd、utilization_percent、avg_cost_per_task等派生属性且组织级预算被突破时会触发全局杀停——_org_killed置位后所有 Agent 的预算对象都会被标记为killedguard.py。每日开始时调用guard.reset_daily()可重置日消耗与限流/杀停状态。成本异常检测from agent_sre.cost import CostAnomalyDetector anomaly_detector CostAnomalyDetector() # 喂入历史成本数据建立基线 for cost in [0.40, 0.42, 0.38, 0.45, 0.41, 0.39, 0.43, 0.40, 0.42, 0.38]: anomaly_detector.ingest(cost, agent_idresearch-agent) # 检查新数据点是否异常 result anomaly_detector.ingest(4.50, agent_idresearch-agent) if result and result.is_anomaly: print(f Cost anomaly: ${result.value:.2f} (expected {result.expected_range})) print(f Severity: {result.severity.value}, Score: {result.score:.1f})CostAnomalyDetector 采用 z-score 阈值法数据点少于min_samples默认 10时不判定z 值超过z_threshold默认 2.5判定为异常z 3.0 时严重度为 HIGH否则为 MEDIUM并给出期望范围mean ± z_threshold * std_dev。此外CostGuard内置的成本历史最近 1000 条也会在每次record_cost时做同样的 z-score 检查z 2.0 触发 WARNING实现护栏内嵌异常检测。README 中提到的 Z-score、IQR、EWMA 三种方法对应的参数z_threshold、iqr_multiplier、ewma_alpha在构造函数中均有预留便于后续扩展。组合使用生产级 SRE 流水线下面是整合全部组件的生产级 SRE 流水线AI Agent 的生产级 SRE 流水线。 import time from agent_sre.anomaly import RogueAgentDetector, RogueDetectorConfig, RiskLevel from agent_sre.cascade.circuit_breaker import ( CircuitBreaker, CircuitBreakerConfig, CircuitOpenError, ) from agent_sre.slo import SLI, SLIValue, SLO, ErrorBudget from agent_sre.slo.indicators import TimeWindow from agent_sre.slo.objectives import ExhaustionAction from agent_sre.cost import CostGuard, BudgetAction # ── 1. 配置所有组件 ───────────────────────────────────── AGENT_ID production-agent # 异常检测 rogue_detector RogueAgentDetector( configRogueDetectorConfig( frequency_z_threshold3.0, quarantine_risk_levelRiskLevel.HIGH, ), ) rogue_detector.register_capability_profile( AGENT_ID, allowed_tools[search, read_file, write_file, run_tests], ) # 熔断器 breaker CircuitBreaker( agent_idAGENT_ID, configCircuitBreakerConfig( failure_threshold5, recovery_timeout_seconds60.0, ), ) # SLO 跟踪 class SuccessRateSLI(SLI): def collect(self) - SLIValue: values self.values_in_window() if not values: return self.record(1.0) good sum(1 for v in values if v.is_good) return self.record(good / len(values)) slo SLO( namef{AGENT_ID}-reliability, indicators[SuccessRateSLI(namesuccess_rate, target0.995, window24h)], error_budgetErrorBudget( total0.005, exhaustion_actionExhaustionAction.CIRCUIT_BREAK, ), agent_idAGENT_ID, ) # 成本护栏 cost_guard CostGuard( per_task_limit2.00, per_agent_daily_limit100.00, auto_throttleTrue, kill_switch_threshold0.95, ) # ── 2. Agent 执行包装器 ────────────────────────────────────── def execute_task(agent_id: str, task: dict) - dict: 让任务穿过完整的 SRE 流水线。 # 预检成本检查 allowed, reason cost_guard.check_task(agent_id, estimated_costtask.get(est_cost, 0)) if not allowed: return {status: blocked, reason: reason} # 预检异常检查 assessment rogue_detector.assess(agent_id) if assessment.quarantine_recommended: return { status: quarantined, risk_level: assessment.risk_level.value, score: assessment.composite_score, } # 通过熔断器执行 try: result breaker.call(_run_agent, agent_id, task) except CircuitOpenError as e: return {status: circuit_open, retry_after: e.retry_after} except Exception as exc: slo.error_budget.record_event(goodFalse) return {status: error, error: str(exc)} # 飞行后记录成功 成本 slo.error_budget.record_event(goodTrue) cost_alerts cost_guard.record_cost( agent_idagent_id, task_idtask[id], cost_usdresult.get(cost, 0), ) # 记录动作供异常检测使用 for tool in result.get(tools_used, []): rogue_detector.record_action(agent_id, actiontool_call, tool_nametool) # 检查关键告警 for alert in cost_alerts: if alert.action BudgetAction.KILL: breaker.record_failure() # 同时触发熔断器 return {status: success, result: result, alerts: [a.to_dict() for a in cost_alerts]} def _run_agent(agent_id: str, task: dict) - dict: 占位你的真实 Agent 逻辑。 return { output: task completed, cost: 0.35, tools_used: [search, read_file], } # ── 3. 运行任务 ──────────────────────────────────────────────────── tasks [ {id: t1, query: Review PR #42, est_cost: 0.50}, {id: t2, query: Summarize docs, est_cost: 0.30}, {id: t3, query: Run test suite, est_cost: 1.00}, ] for task in tasks: result execute_task(AGENT_ID, task) print(fTask {task[id]}: {result[status]}) # 上报 SRE 健康状态 print(f\nCircuit state: {breaker.state}) print(fError budget remaining: {slo.error_budget.remaining_percent:.1f}%) budget cost_guard.get_budget(AGENT_ID) print(fCost today: ${budget.spent_today_usd:.2f} / ${budget.daily_limit_usd:.2f})这条流水线为你提供什么层级组件保护能力预检CostGuard.check_task拦截会超预算的任务预检RogueAgentDetector.assess隔离被攻破的 Agent执行CircuitBreaker.call隔离持续失败的 Agent飞行后ErrorBudget.record_event随时间跟踪可靠性飞行后CostGuard.record_cost检测成本异常、自动限流飞行后RogueAgentDetector.record_action建立行为基线注意流水线中的两个巧妙联动一是ErrorBudget的exhaustion_actionExhaustionAction.CIRCUIT_BREAK预算耗尽时与熔断器协同二是成本告警出现KILL动作时同时调用breaker.record_failure()让熔断器也跳闸形成成本失控 → 执行隔离的双重防御。下一步把 SRE 流水线延伸到发布与告警渐进式交付使用agent_sre.delivery.BlueGreenManager以验证与自动回滚安全发布新的 Agent 版本对应 delivery/blue_green.pyREADME 中还展示了声明式的AgentRolloutYAML 规格支持 shadow → 5% → 25% → 100% 的 canary 阶段与自动回滚条件错误预算燃烧速率 5.0、策略违规 0、检测到成本异常。告警将agent_sre.alerts.AlertManager接入你的通知系统PagerDuty、Slack、Teamsalerts/dedup.py 提供告警去重SLO 告警自带dedup_key。仪表盘通过agent_sre.slo.dashboard导出 SLO 数据实现实时可视SLO 状态也可通过slo.to_dict()/error_budget.to_dict()序列化后接入你现有的 Grafana/Prometheusintegrations/prometheus/exporter.py或 OpenTelemetryintegrations/otel链路。定时混沌使用agent_sre.chaos.ChaosSchedulerchaos/scheduler.py执行周期性韧性演练并支持黑窗blackout window避开业务高峰。事件响应事故发生时agent_sre.incidents提供信号关联、去重、按 Agent 熔断与自动 runbook 执行内置重启 Agent、撤销信任、回滚版本、限流四个 runbook见 incidents/runbooks并自动生成带时间线与行动项的事后报告postmortem。小结AI Agent 的可靠性不能靠运气 重试来保证。agent-sre用一套可组合、可验证的 SRE 原语覆盖了从失控检测、故障隔离、可靠性量化到成本管控的完整闭环RogueAgentDetector用频率 z-score 动作熵 能力画像三信号识别失控 Agent 并输出防篡改审计链CircuitBreaker用 CLOSED/OPEN/HALF_OPEN 状态机隔离失败并支持级联检测SLOErrorBudget把可靠变成可度量的目标并驱动告警与限流ChaosExperiment用 11 个内置模板在事故之前验证韧性CostGuard用三级预算与内嵌 z-score 异常检测掐灭成本爆炸。将它们串成一条预检 → 执行 → 飞行后的流水线你的自主 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），仅供参考