人工智能AI AgentAI 安全治理策略引擎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点击查看免费下载本指南对应仓库 docs/tutorials/51-cost-governance.md属于agent-sreSRE包的进阶教程阅读时间约 15 分钟适合具备 Python 基础的中级开发者。导读在自主智能体Agent规模化落地的场景中成本失控是比功能缺陷更隐蔽的生产事故一次失控的循环调用可以在数小时内烧穿整个月的 LLM 预算多 Agent 编排还会通过级联调用让成本呈指数级放大。本文基于 Agent Governance Toolkit 中agent-sre包提供的CostGuard成本治理体系系统讲解如何设置单 Agent 与组织级预算、如何利用 50%/75%/85%/90%/95% 分级阈值实现告警—限流—熔断的自动降级以及如何结合内置的异常检测与成本优化器完成从事后救火到事前预防的完整闭环。读完本文你将掌握一套可直接落地到生产环境的 Agent 成本治理方案并能对照源码理解其底层实现原理。为什么需要成本治理在引入CostGuard之前Agent 的成本消耗几乎是黑盒只有事后对账才能发现超支而此时损失已经发生。多 Agent 编排使问题进一步恶化——级联调用会跨越多个 Agent 相互放大费用单个 Agent 的偶发异常也可能拖垮全局月度预算。仓库中的架构决策文档 ADR 0012: Cost Governance via Observability Policies in Agent SRE 记录了这套方案的设计动因与约束成本元数据应来自多个来源工具注解、策略映射、运行时计量并分层叠加预算执行应当分级软上限soft cap触发告警硬上限hard cap阻止后续动作同时支持单 Agent 预算与组织级全局预算执行策略是post-action动作后执行而非动作前预测该能力归属于agent-sre模块而不是策略评估器或新模块。CostGuard是这一设计的具体实现它位于 agent_sre/cost/guard.py对外提供如下能力矩阵能力作用单 Agent 预算每个 Agent 的每日支出上限单任务上限单个任务的最大成本组织级预算所有 Agent 合计的月度全局上限分级告警在 50%、75%、90%、95% 利用率处触发告警自动限流Throttle达到 85% 预算时标记 Agent 为限流状态熔断Kill Switch达到 95% 预算时停止 Agent成本异常检测基于 Z-score 标记异常支出模式核心概念Post-Action 执行为什么选择动作后而非动作前CostGuard采用动作后执行post-action enforcement每个动作完成后记录实际成本并检查预算状态。相比动作前的成本预测这种方式更准确原因在于LLM 的 token 消耗量因请求而异无法可靠预估工具成本取决于参数如数据库查询复杂度、搜索结果量只有来自计费 API 的真实账单数据才是最终依据预测值只是估计。其执行流程可以概括为Action executed - CostGuard.record_cost() - Budget check | Under soft cap: log only Over soft cap: alert Over hard cap: kill agent这一点与 ADR 0012 中的Enforcement should be post-action (observe cost, alert if trending over), not pre-action prediction约束完全一致。分级预算模型预算按利用率从低到高划分为五个状态对应不同的处置动作0% 50% 75% 85% 90% 95% 100% |----------|---------|---------|--------|--------|--------| OK WARN WARN THROTTLE CRIT KILL BLOCKED从源码 guard.py 可以看到实际判定逻辑利用率达到kill_switch_threshold默认 0.95即 95%时Agent 被标记为killed产生BudgetAction.KILL告警利用率达到 85% 但未达 kill 阈值时Agent 被标记为throttled产生BudgetAction.THROTTLE告警一旦处于 throttle 或 kill 状态check_task会直接拒绝后续任务见下文 Step 2。环境准备与安装教程原文以pip install agent-sre安装。需要注意的是根据 agent-sre/pyproject.toml 的说明agent-sre发行版目前是仅依赖的重定向桩dep-only deprecation stub其安装行为是将用户重定向到整合后的agent-governance-toolkit-cli该 CLI 包整合了agent-sre、agt-sandbox与agentmesh-mcp-trust源码仍保留在 agent_sre 目录中并在 CI 中持续验证。因此当前仓库环境下的推荐安装方式为pip install agent-governance-toolkit-cli # 如需 docker 沙箱与 MCP 能力 pip install agent-governance-toolkit-cli[docker,mcp]安装完成后即可从agent_sre.cost导入全部成本治理组件导出清单见 agent_sre/cost/init.pyfrom agent_sre.cost import CostGuard, CostAnomalyDetector, CostOptimizerStep 1基础预算配置创建一个CostGuard实例并设置三层预算单任务上限、单 Agent 日预算、组织月度预算。from agent_sre.cost import CostGuard # Create a cost guard with budget limits guard CostGuard( per_task_limit2.00, # Max $2 per task per_agent_daily_limit50.00, # Max $50/day per agent org_monthly_budget1000.00, # Max $1000/month total auto_throttleTrue, # Auto-throttle at 85%, kill at 95% ) print(fPer-task limit: ${guard.per_task_limit:.2f}) print(fDaily agent limit: ${guard.per_agent_daily_limit:.2f}) print(fOrg monthly: ${guard.org_monthly_budget:.2f})参数说明与默认值均可在 guard.py 的构造器中确认参数默认值说明per_task_limit2.0单个任务允许的最大成本美元per_agent_daily_limit100.0单个 Agent 每日支出上限美元org_monthly_budget5000.0组织月度全局上限美元anomaly_detectionTrue是否启用内置异常检测auto_throttleTrue是否自动限流与熔断kill_switch_threshold0.95触发熔断的利用率0.0~1.0alert_thresholds[0.50, 0.75, 0.90, 0.95]告警触发利用率列表源码中还包含严格的输入校验所有预算参数必须为有限非负数kill_switch_threshold与alert_thresholds中每个阈值必须落在[0.0, 1.0]区间否则构造器直接抛出ValueError。这一点由测试 tests/unit/test_cost.py 中的test_nan_inf_budget_does_not_crash等用例验证——NaN、Inf、负数预算在__init__阶段即被拒绝杜绝了半初始化对象携带损坏状态的风险。Step 2任务预检check_task在运行昂贵任务之前先用check_task确认预算是否允许# Check if a task can proceed allowed, reason guard.check_task(analyst-agent, estimated_cost1.50) print(fTask allowed: {allowed} ({reason})) # Output: Task allowed: True (ok) # Check a task that exceeds per-task limit allowed, reason guard.check_task(analyst-agent, estimated_cost5.00) print(fExpensive task: {allowed} ({reason})) # Output: Expensive task: False (Estimated cost $5.00 exceeds per-task limit $2.00)check_task返回(allowed, reason)二元组其内部判定顺序guard.py为组织级已熔断Organization budget exhausted→ 直接拒绝Agent 已被 killAgent killed — budget exhausted→ 拒绝Agent 处于 throttle 状态Agent throttled — approaching daily limit→ 拒绝预估成本超过per_task_limit→ 拒绝并给出超出金额当日已支出 预估成本超过日预算 → 拒绝并给出剩余额度组织月度已支出 预估成本超过组织预算 → 拒绝并给出组织剩余额度全部通过 → 返回True, ok。重要提示check_task 是咨询性检查源码文档字符串明确警告check_task不会预留reserve预算两个并发调用方可能同时通过检查并同时记账最终双双超支。对于预算敏感的关键路径应改用check_and_charge——它在单把锁内原子地完成检查 记账是真正能在并发下守住预算的原语allowed, reason, alerts guard.check_and_charge( analyst-agent, task-003, cost_usd1.20 ) if not allowed: # 预算不允许不要执行任务 print(fBlocked: {reason}) else: # 成本已记账alerts 为本次触发的告警 print(fCharged: {reason}, alerts{len(alerts)})check_and_charge的完整实现见 guard.py其输入校验拒绝 NaN 与负成本与并发安全同样有测试覆盖test_cost.py 的test_concurrent_record_cost_no_crash用 10 线程 × 100 条记录验证了状态不被破坏。Step 3记录成本与获取告警每个任务完成后调用record_cost记录实际成本# Record a normal task alerts guard.record_cost(analyst-agent, task-001, cost_usd0.50) print(fTask 001: $0.50, alerts: {len(alerts)}) # Record with cost breakdown alerts guard.record_cost( analyst-agent, task-002, cost_usd1.20, breakdown{gpt-4: 1.00, web-search: 0.20}, ) print(fTask 002: $1.20, alerts: {len(alerts)}) # Check budget status budget guard.get_budget(analyst-agent) print(fSpent today: ${budget.spent_today_usd:.2f}) print(fRemaining: ${budget.remaining_today_usd:.2f}) print(fUtilization: {budget.utilization_percent:.1f}%)record_cost返回本次记账触发的CostAlert列表。底层记录对象CostRecordguard.py除agent_id、task_id、cost_usd外还支持breakdown按模型/工具拆分与metadata字段并可通过to_dict()序列化用于审计。每次记账都会同步累加单 Agent 当日支出与组织月度支出_org_spent_month并写入最多 1000 条的滑动成本历史_cost_history供异常检测使用。Step 4观察告警分级升级随着支出增加告警会沿阈值逐级升级。下面的示例模拟连续记账观察 throttle 与 kill 的触发# Simulate spending that triggers alerts for i in range(20): task_id ftask-{i 10:03d} alerts guard.record_cost(analyst-agent, task_id, cost_usd2.00) if alerts: for alert in alerts: print(f [{alert.severity.value.upper()}] {alert.message}) if alert.action.value ! alert: print(f Action: {alert.action.value}) # Check final budget status budget guard.get_budget(analyst-agent) print(f\nFinal status:) print(f Spent: ${budget.spent_today_usd:.2f}) print(f Throttled: {budget.throttled}) print(f Killed: {budget.killed})对照源码 guard.py 可以理解每条告警的触发条件单任务超限单次成本 per_task_limit即产生WARNING分级利用率告警当日利用率跨过alert_thresholds中任一阈值时触发阈值 ≥ 0.90 时为CRITICAL否则为WARNING自动限流auto_throttleTrue且利用率 ≥ 85% 时置throttledTrue产生BudgetAction.THROTTLE自动熔断auto_throttleTrue且利用率 ≥kill_switch_threshold默认 0.95时置killedTrue产生BudgetAction.KILL组织熔断组织月度利用率跨过 kill 阈值时置_org_killedTrue并将已注册的所有 Agent 全部标记为 killed。阈值穿越判定使用prev_util threshold utilization的严格区间逻辑保证每个阈值只触发一次告警避免重复轰炸——测试 test_cost.py 的test_org_kill_alert_fires_once专门验证了这一行为。一旦 Agent 被 kill所有后续任务都会被阻塞allowed, reason guard.check_task(analyst-agent, estimated_cost0.01) print(fAfter kill: allowed{allowed}, reason{reason}) # Output: After kill: allowedFalse, reasonAgent killed — budget exhaustedStep 5组织级预算与全局熔断组织级预算跨所有 Agent 汇总统计是防止多个 Agent 各自正常、合计超支的最后防线guard CostGuard( per_agent_daily_limit100.00, org_monthly_budget200.00, # Low for demo kill_switch_threshold0.95, ) # Multiple agents spending for agent in [agent-a, agent-b, agent-c]: alerts guard.record_cost(agent, task-1, cost_usd60.00) if alerts: for alert in alerts: print(f [{alert.severity.value.upper()}] {alert.message}) # Once org budget is killed, ALL agents are blocked for agent in [agent-a, agent-b, agent-c]: allowed, reason guard.check_task(agent, estimated_cost0.01) print(f {agent}: allowed{allowed})组织级预算的关键语义源码 guard.py 与测试共同确认当组织月度支出跨越 kill 阈值时触发一次CRITICAL的 org kill 告警消息形如Org budget kill switch triggered -- 96% of monthly budget consumed该时刻所有已注册的AgentBudget都会被置为killed之后任意 Agent 的check_task都会被Organization budget exhausted拒绝test_cost.py 的test_org_kill_sets_killed_on_all_agents验证了这一点org_monthly_budget0表示禁用组织级检查零即无限语义而非零预算见test_zero_org_budget_means_no_limit组织级边界采用严格不等式spent estimated budget才拒绝因此恰好等于预算时仍然放行见test_org_budget_boundary_triple49.99 放行、50.00 放行、50.01 拒绝。通过属性org_spent_month、org_remaining_month或summary()可以随时查看组织级状态summary guard.summary() print(fOrg spent: ${summary[org_spent_month]}) print(fOrg remaining:${summary[org_remaining_month]}) print(fTotal alerts: {summary[total_alerts]})Step 6成本异常检测CostGuard内置基于 Z-score 的异常检测anomaly_detectionTrue时启用。当单次成本相对历史均值的偏离超过阈值时产生Anomalous cost detected告警见 guard.py 的_check_anomaly_locked历史样本 ≥ 10 且 Z-score 2.0 时触发。此外仓库还提供独立的CostAnomalyDetectoragent_sre/cost/anomaly.py用于独立的成本流分析from agent_sre.cost import CostAnomalyDetector detector CostAnomalyDetector() # Feed normal cost history for i in range(20): detector.ingest(1.0 (i % 3) * 0.2, agent_iddata-agent) # Check an anomalous cost - returns AnomalyResult if anomaly detected result detector.ingest(50.0, agent_iddata-agent) if result: print(fAnomaly detected!) print(fSeverity: {result.severity.value})CostAnomalyDetector的关键参数与行为anomaly.py参数默认值说明z_threshold2.5Z-score 超过该值判定为异常iqr_multiplier1.5预留的 IQR 方法参数当前实现使用 Z-scoreewma_alpha0.3预留的 EWMA 参数min_samples10至少积累多少样本后才开始判定window_size1000滑动窗口大小判定逻辑样本数不足min_samples时返回None标准差为 0历史恒定时不判定|z| z_threshold时返回AnomalyResult其中 Z-score 3.0 判定为HIGH严重度否则为MEDIUM并附带expected_range期望区间与scoreZ 值。可通过detector.baseline查看当前统计基线均值、标准差、样本数通过detector.anomalies与summary()获取全部异常记录。Step 7成本优化建议CostOptimizer用于在满足质量与延迟约束的前提下为任务推荐成本最低的模型。注意教程示例中的add_model()/optimize()接口在当前源码中已演化为构造器传入模型列表 recommend()查询的形式以下代码以仓库实际实现 agent_sre/cost/optimizer.py 为准from agent_sre.cost import CostOptimizer, ModelConfig, TaskProfile # Register model options (constructor takes the model list) optimizer CostOptimizer(models[ ModelConfig( namegpt-4, provideropenai, cost_per_1k_input_tokens0.03, cost_per_1k_output_tokens0.06, avg_latency_ms900, quality_score0.95, ), ModelConfig( namegpt-3.5-turbo, provideropenai, cost_per_1k_input_tokens0.002, cost_per_1k_output_tokens0.004, avg_latency_ms300, quality_score0.80, ), ]) # Get optimization for a task task TaskProfile( task_typesummarization, avg_input_tokens1500, avg_output_tokens500, min_quality0.75, max_latency_ms2000, ) result optimizer.recommend(task) print(fRecommended: {result.recommendations[0].model_name}) print(fEstimated cost: ${result.recommendations[0].estimated_cost:.4f}) if result.potential_savings_pct is not None: print(fSavings vs default: {result.potential_savings_pct:.0f}%)ModelConfig与TaskProfile均为 Pydantic 模型optimizer.py字段如下ModelConfigname、provider、cost_per_1k_input_tokens、cost_per_1k_output_tokens、avg_latency_ms、quality_score0~1TaskProfiletask_type、avg_input_tokens、avg_output_tokens、min_quality0~1、max_latency_ms可选。recommend()的推荐逻辑是筛选出质量 ≥min_quality且延迟 ≤max_latency_ms的可行模型按估算成本升序排列最优者标记is_optimalTrue若传入current_model还会计算相对当前模型的潜在节省百分比potential_savings_pct。CostOptimizer还提供三个进阶方法# Pareto 前沿分析没有其他模型既更便宜又更高质量 frontier optimizer.pareto_frontier(task) # 批量模拟按请求量投影总成本 projection optimizer.simulate(task, model_namegpt-3.5-turbo, volume10000) print(fTotal cost for 10k requests: ${projection[total_cost]}) # 路由建议为多个任务类型分别推荐最便宜模型 routing optimizer.suggest_routing([task, TaskProfile(task_typecoding, ...)])其中simulate会抛出KeyError未知模型名pareto_frontier返回按成本升序的 Pareto 最优集合——测试 tests/test_cost_optimizer.py 使用gpt-4o-mini、gpt-4o、claude-opus三档模型对推荐、Pareto 前沿、模拟与路由建议做了完整验证。预算配置参考完整参数组合如下默认值已在上文表格中给出CostGuard( per_task_limit2.0, # Max cost per single task per_agent_daily_limit100.0, # Max daily spend per agent org_monthly_budget5000.0, # Global monthly cap anomaly_detectionTrue, # Enable anomaly detection auto_throttleTrue, # Auto-throttle and kill kill_switch_threshold0.95, # Kill at 95% utilization alert_thresholds[ # Alert at these percentages 0.50, 0.75, 0.90, 0.95 ], )API 参考CostGuard方法说明check_task(agent_id, estimated_cost)预检任务是否在预算内咨询性不预留额度check_and_charge(agent_id, task_id, cost_usd)原子地检查 记账并发安全返回(allowed, reason, alerts)record_cost(agent_id, task_id, cost_usd, breakdownNone)记录实际成本并返回触发的告警列表get_budget(agent_id)获取单个 Agent 的预算状态get_all_budgets()获取所有 Agent 的预算状态通过summary()的agents字段get_alerts()获取全部已触发告警通过alerts属性get_summary()获取组织级成本汇总通过summary()方法reset_daily(agent_idNone)重置单 Agent 或全部 Agent 的当日预算清零支出、任务数、throttle/kill 状态AgentBudget字段类型说明spent_today_usdfloat今日累计支出remaining_today_usdfloat今日剩余额度max(0, daily_limit - spent)utilization_percentfloat当日利用率0~100%throttledbool利用率 ≥ 85% 时置位killedbool利用率 ≥ kill 阈值默认 95%时置位task_count_todayint今日任务数avg_cost_per_taskfloat平均单任务成本CostAlert字段说明severityCostAlertSeverityinfo/warning/criticalactionBudgetActionalert/throttle/killmessage人类可读的告警消息current_value/threshold当前值与该告警对应的阈值agent_id/timestamp归属 Agent 与时间戳所有告警与记录对象均提供to_dict()序列化方法便于接入审计链路。生命周期与运维要点每日重置CostGuard不会自动感知新的一天需要由外部调度如 cron在每日开始时调用reset_daily()清零所有 Agent 的当日支出、任务计数与 throttle/kill 状态见 guard.py。组织月度预算则依赖org_spent_month的滚动累计跨月时需要重建CostGuard或自行重置。并发安全CostGuard内部使用threading.Lock保护所有预算状态record_cost、check_task、check_and_charge均可在多线程环境下安全调用有test_concurrent_record_cost_no_crash覆盖。审计序列化CostRecord、CostAlert、AgentBudget、AnomalyResult、BaselineStats均实现to_dict()成本事件可直接落入审计存储。与可观测性体系的结合成本治理不是孤岛。将CostGuard与仓库中其他 SRE 能力串联可以构建完整的成本 可靠性观测闭环结合 SLO 与错误预算成本治理与服务质量目标、错误预算相辅相成——成本告警可视为 SLO 之外的财务健康指标参见 Tutorial 05 - Agent Reliability (SRE)结合多 Agent 策略将成本上限与集体限流collective rate limiting组合使用防止成本合规但调用频次爆表参见 Tutorial 49 - Multi-Agent Policies结合可观测性与链路追踪将成本事件与 OpenTelemetry 链路关联在排障时同时看到哪个任务花了多少钱与调用链路全貌参见 Tutorial 13 - Observability Tracing设计背景成本治理的整体架构决策分层成本元数据、post-action 执行、软/硬上限分级详见 ADR 0012仓库配套示例agent-sre的集成层还提供了面向 LangChain、OpenTelemetry、Prometheus 等生态的导出器见 agent_sre/integrations可将成本指标接入既有监控体系。下一步本文完整覆盖了CostGuard的预算设置、分级告警、自动限流熔断、组织级预算、异常检测与成本优化。建议下一步在生产环境先以仅告警、不熔断auto_throttleFalse观察一周真实成本曲线再逐步开启自动限流与熔断为预算关键路径统一改用check_and_charge原子原语避免并发超支通过 docs/tutorials/05-agent-reliability.md 与 docs/tutorials/13-observability-and-tracing.md 将成本治理纳入完整的 SRE 观测体系。赞分享人工智能AI AgentAI 安全治理策略引擎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点击查看免费下载相关推荐Agent Governance Toolkit 成本治理实战用 CostGuard 实现分级预算、自动节流与成本异常检测Agent Governance Toolkit 成本治理实战用 CostGuard 实现分级预算、自动节流与成本异常检测 本指南以 examples/cos人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权Agent Governance Toolkit A2A 对话治理实战技能级访问控制、信任评分与反馈循环熔断Agent Governance Toolkit A2A 对话治理实战技能级访问控制、信任评分与反馈循环熔断 本文对应仓库教程 Tutorial 44 — A人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权VueGL响应式交互设计从表单输入到3D场景控制VueGL响应式交互设计从表单输入到3D场景控制 VueGL是一个基于Vue.js和three.js的3D WebGL图形渲染组件库它能够帮助开发者轻松实现人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/20 2:30:36 · 阅读 1.2k Agent Governance Toolkit 成本治理实战：CostGuard 预算管控、分级告警与自动熔断
人工智能AI AgentAI 安全治理策略引擎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点击查看免费下载本指南对应仓库 docs/tutorials/51-cost-governance.md属于agent-sreSRE包的进阶教程阅读时间约 15 分钟适合具备 Python 基础的中级开发者。导读在自主智能体Agent规模化落地的场景中成本失控是比功能缺陷更隐蔽的生产事故一次失控的循环调用可以在数小时内烧穿整个月的 LLM 预算多 Agent 编排还会通过级联调用让成本呈指数级放大。本文基于 Agent Governance Toolkit 中agent-sre包提供的CostGuard成本治理体系系统讲解如何设置单 Agent 与组织级预算、如何利用 50%/75%/85%/90%/95% 分级阈值实现告警—限流—熔断的自动降级以及如何结合内置的异常检测与成本优化器完成从事后救火到事前预防的完整闭环。读完本文你将掌握一套可直接落地到生产环境的 Agent 成本治理方案并能对照源码理解其底层实现原理。为什么需要成本治理在引入CostGuard之前Agent 的成本消耗几乎是黑盒只有事后对账才能发现超支而此时损失已经发生。多 Agent 编排使问题进一步恶化——级联调用会跨越多个 Agent 相互放大费用单个 Agent 的偶发异常也可能拖垮全局月度预算。仓库中的架构决策文档 ADR 0012: Cost Governance via Observability Policies in Agent SRE 记录了这套方案的设计动因与约束成本元数据应来自多个来源工具注解、策略映射、运行时计量并分层叠加预算执行应当分级软上限soft cap触发告警硬上限hard cap阻止后续动作同时支持单 Agent 预算与组织级全局预算执行策略是post-action动作后执行而非动作前预测该能力归属于agent-sre模块而不是策略评估器或新模块。CostGuard是这一设计的具体实现它位于 agent_sre/cost/guard.py对外提供如下能力矩阵能力作用单 Agent 预算每个 Agent 的每日支出上限单任务上限单个任务的最大成本组织级预算所有 Agent 合计的月度全局上限分级告警在 50%、75%、90%、95% 利用率处触发告警自动限流Throttle达到 85% 预算时标记 Agent 为限流状态熔断Kill Switch达到 95% 预算时停止 Agent成本异常检测基于 Z-score 标记异常支出模式核心概念Post-Action 执行为什么选择动作后而非动作前CostGuard采用动作后执行post-action enforcement每个动作完成后记录实际成本并检查预算状态。相比动作前的成本预测这种方式更准确原因在于LLM 的 token 消耗量因请求而异无法可靠预估工具成本取决于参数如数据库查询复杂度、搜索结果量只有来自计费 API 的真实账单数据才是最终依据预测值只是估计。其执行流程可以概括为Action executed - CostGuard.record_cost() - Budget check | Under soft cap: log only Over soft cap: alert Over hard cap: kill agent这一点与 ADR 0012 中的Enforcement should be post-action (observe cost, alert if trending over), not pre-action prediction约束完全一致。分级预算模型预算按利用率从低到高划分为五个状态对应不同的处置动作0% 50% 75% 85% 90% 95% 100% |----------|---------|---------|--------|--------|--------| OK WARN WARN THROTTLE CRIT KILL BLOCKED从源码 guard.py 可以看到实际判定逻辑利用率达到kill_switch_threshold默认 0.95即 95%时Agent 被标记为killed产生BudgetAction.KILL告警利用率达到 85% 但未达 kill 阈值时Agent 被标记为throttled产生BudgetAction.THROTTLE告警一旦处于 throttle 或 kill 状态check_task会直接拒绝后续任务见下文 Step 2。环境准备与安装教程原文以pip install agent-sre安装。需要注意的是根据 agent-sre/pyproject.toml 的说明agent-sre发行版目前是仅依赖的重定向桩dep-only deprecation stub其安装行为是将用户重定向到整合后的agent-governance-toolkit-cli该 CLI 包整合了agent-sre、agt-sandbox与agentmesh-mcp-trust源码仍保留在 agent_sre 目录中并在 CI 中持续验证。因此当前仓库环境下的推荐安装方式为pip install agent-governance-toolkit-cli # 如需 docker 沙箱与 MCP 能力 pip install agent-governance-toolkit-cli[docker,mcp]安装完成后即可从agent_sre.cost导入全部成本治理组件导出清单见 agent_sre/cost/init.pyfrom agent_sre.cost import CostGuard, CostAnomalyDetector, CostOptimizerStep 1基础预算配置创建一个CostGuard实例并设置三层预算单任务上限、单 Agent 日预算、组织月度预算。from agent_sre.cost import CostGuard # Create a cost guard with budget limits guard CostGuard( per_task_limit2.00, # Max $2 per task per_agent_daily_limit50.00, # Max $50/day per agent org_monthly_budget1000.00, # Max $1000/month total auto_throttleTrue, # Auto-throttle at 85%, kill at 95% ) print(fPer-task limit: ${guard.per_task_limit:.2f}) print(fDaily agent limit: ${guard.per_agent_daily_limit:.2f}) print(fOrg monthly: ${guard.org_monthly_budget:.2f})参数说明与默认值均可在 guard.py 的构造器中确认参数默认值说明per_task_limit2.0单个任务允许的最大成本美元per_agent_daily_limit100.0单个 Agent 每日支出上限美元org_monthly_budget5000.0组织月度全局上限美元anomaly_detectionTrue是否启用内置异常检测auto_throttleTrue是否自动限流与熔断kill_switch_threshold0.95触发熔断的利用率0.0~1.0alert_thresholds[0.50, 0.75, 0.90, 0.95]告警触发利用率列表源码中还包含严格的输入校验所有预算参数必须为有限非负数kill_switch_threshold与alert_thresholds中每个阈值必须落在[0.0, 1.0]区间否则构造器直接抛出ValueError。这一点由测试 tests/unit/test_cost.py 中的test_nan_inf_budget_does_not_crash等用例验证——NaN、Inf、负数预算在__init__阶段即被拒绝杜绝了半初始化对象携带损坏状态的风险。Step 2任务预检check_task在运行昂贵任务之前先用check_task确认预算是否允许# Check if a task can proceed allowed, reason guard.check_task(analyst-agent, estimated_cost1.50) print(fTask allowed: {allowed} ({reason})) # Output: Task allowed: True (ok) # Check a task that exceeds per-task limit allowed, reason guard.check_task(analyst-agent, estimated_cost5.00) print(fExpensive task: {allowed} ({reason})) # Output: Expensive task: False (Estimated cost $5.00 exceeds per-task limit $2.00)check_task返回(allowed, reason)二元组其内部判定顺序guard.py为组织级已熔断Organization budget exhausted→ 直接拒绝Agent 已被 killAgent killed — budget exhausted→ 拒绝Agent 处于 throttle 状态Agent throttled — approaching daily limit→ 拒绝预估成本超过per_task_limit→ 拒绝并给出超出金额当日已支出 预估成本超过日预算 → 拒绝并给出剩余额度组织月度已支出 预估成本超过组织预算 → 拒绝并给出组织剩余额度全部通过 → 返回True, ok。重要提示check_task 是咨询性检查源码文档字符串明确警告check_task不会预留reserve预算两个并发调用方可能同时通过检查并同时记账最终双双超支。对于预算敏感的关键路径应改用check_and_charge——它在单把锁内原子地完成检查 记账是真正能在并发下守住预算的原语allowed, reason, alerts guard.check_and_charge( analyst-agent, task-003, cost_usd1.20 ) if not allowed: # 预算不允许不要执行任务 print(fBlocked: {reason}) else: # 成本已记账alerts 为本次触发的告警 print(fCharged: {reason}, alerts{len(alerts)})check_and_charge的完整实现见 guard.py其输入校验拒绝 NaN 与负成本与并发安全同样有测试覆盖test_cost.py 的test_concurrent_record_cost_no_crash用 10 线程 × 100 条记录验证了状态不被破坏。Step 3记录成本与获取告警每个任务完成后调用record_cost记录实际成本# Record a normal task alerts guard.record_cost(analyst-agent, task-001, cost_usd0.50) print(fTask 001: $0.50, alerts: {len(alerts)}) # Record with cost breakdown alerts guard.record_cost( analyst-agent, task-002, cost_usd1.20, breakdown{gpt-4: 1.00, web-search: 0.20}, ) print(fTask 002: $1.20, alerts: {len(alerts)}) # Check budget status budget guard.get_budget(analyst-agent) print(fSpent today: ${budget.spent_today_usd:.2f}) print(fRemaining: ${budget.remaining_today_usd:.2f}) print(fUtilization: {budget.utilization_percent:.1f}%)record_cost返回本次记账触发的CostAlert列表。底层记录对象CostRecordguard.py除agent_id、task_id、cost_usd外还支持breakdown按模型/工具拆分与metadata字段并可通过to_dict()序列化用于审计。每次记账都会同步累加单 Agent 当日支出与组织月度支出_org_spent_month并写入最多 1000 条的滑动成本历史_cost_history供异常检测使用。Step 4观察告警分级升级随着支出增加告警会沿阈值逐级升级。下面的示例模拟连续记账观察 throttle 与 kill 的触发# Simulate spending that triggers alerts for i in range(20): task_id ftask-{i 10:03d} alerts guard.record_cost(analyst-agent, task_id, cost_usd2.00) if alerts: for alert in alerts: print(f [{alert.severity.value.upper()}] {alert.message}) if alert.action.value ! alert: print(f Action: {alert.action.value}) # Check final budget status budget guard.get_budget(analyst-agent) print(f\nFinal status:) print(f Spent: ${budget.spent_today_usd:.2f}) print(f Throttled: {budget.throttled}) print(f Killed: {budget.killed})对照源码 guard.py 可以理解每条告警的触发条件单任务超限单次成本 per_task_limit即产生WARNING分级利用率告警当日利用率跨过alert_thresholds中任一阈值时触发阈值 ≥ 0.90 时为CRITICAL否则为WARNING自动限流auto_throttleTrue且利用率 ≥ 85% 时置throttledTrue产生BudgetAction.THROTTLE自动熔断auto_throttleTrue且利用率 ≥kill_switch_threshold默认 0.95时置killedTrue产生BudgetAction.KILL组织熔断组织月度利用率跨过 kill 阈值时置_org_killedTrue并将已注册的所有 Agent 全部标记为 killed。阈值穿越判定使用prev_util threshold utilization的严格区间逻辑保证每个阈值只触发一次告警避免重复轰炸——测试 test_cost.py 的test_org_kill_alert_fires_once专门验证了这一行为。一旦 Agent 被 kill所有后续任务都会被阻塞allowed, reason guard.check_task(analyst-agent, estimated_cost0.01) print(fAfter kill: allowed{allowed}, reason{reason}) # Output: After kill: allowedFalse, reasonAgent killed — budget exhaustedStep 5组织级预算与全局熔断组织级预算跨所有 Agent 汇总统计是防止多个 Agent 各自正常、合计超支的最后防线guard CostGuard( per_agent_daily_limit100.00, org_monthly_budget200.00, # Low for demo kill_switch_threshold0.95, ) # Multiple agents spending for agent in [agent-a, agent-b, agent-c]: alerts guard.record_cost(agent, task-1, cost_usd60.00) if alerts: for alert in alerts: print(f [{alert.severity.value.upper()}] {alert.message}) # Once org budget is killed, ALL agents are blocked for agent in [agent-a, agent-b, agent-c]: allowed, reason guard.check_task(agent, estimated_cost0.01) print(f {agent}: allowed{allowed})组织级预算的关键语义源码 guard.py 与测试共同确认当组织月度支出跨越 kill 阈值时触发一次CRITICAL的 org kill 告警消息形如Org budget kill switch triggered -- 96% of monthly budget consumed该时刻所有已注册的AgentBudget都会被置为killed之后任意 Agent 的check_task都会被Organization budget exhausted拒绝test_cost.py 的test_org_kill_sets_killed_on_all_agents验证了这一点org_monthly_budget0表示禁用组织级检查零即无限语义而非零预算见test_zero_org_budget_means_no_limit组织级边界采用严格不等式spent estimated budget才拒绝因此恰好等于预算时仍然放行见test_org_budget_boundary_triple49.99 放行、50.00 放行、50.01 拒绝。通过属性org_spent_month、org_remaining_month或summary()可以随时查看组织级状态summary guard.summary() print(fOrg spent: ${summary[org_spent_month]}) print(fOrg remaining:${summary[org_remaining_month]}) print(fTotal alerts: {summary[total_alerts]})Step 6成本异常检测CostGuard内置基于 Z-score 的异常检测anomaly_detectionTrue时启用。当单次成本相对历史均值的偏离超过阈值时产生Anomalous cost detected告警见 guard.py 的_check_anomaly_locked历史样本 ≥ 10 且 Z-score 2.0 时触发。此外仓库还提供独立的CostAnomalyDetectoragent_sre/cost/anomaly.py用于独立的成本流分析from agent_sre.cost import CostAnomalyDetector detector CostAnomalyDetector() # Feed normal cost history for i in range(20): detector.ingest(1.0 (i % 3) * 0.2, agent_iddata-agent) # Check an anomalous cost - returns AnomalyResult if anomaly detected result detector.ingest(50.0, agent_iddata-agent) if result: print(fAnomaly detected!) print(fSeverity: {result.severity.value})CostAnomalyDetector的关键参数与行为anomaly.py参数默认值说明z_threshold2.5Z-score 超过该值判定为异常iqr_multiplier1.5预留的 IQR 方法参数当前实现使用 Z-scoreewma_alpha0.3预留的 EWMA 参数min_samples10至少积累多少样本后才开始判定window_size1000滑动窗口大小判定逻辑样本数不足min_samples时返回None标准差为 0历史恒定时不判定|z| z_threshold时返回AnomalyResult其中 Z-score 3.0 判定为HIGH严重度否则为MEDIUM并附带expected_range期望区间与scoreZ 值。可通过detector.baseline查看当前统计基线均值、标准差、样本数通过detector.anomalies与summary()获取全部异常记录。Step 7成本优化建议CostOptimizer用于在满足质量与延迟约束的前提下为任务推荐成本最低的模型。注意教程示例中的add_model()/optimize()接口在当前源码中已演化为构造器传入模型列表 recommend()查询的形式以下代码以仓库实际实现 agent_sre/cost/optimizer.py 为准from agent_sre.cost import CostOptimizer, ModelConfig, TaskProfile # Register model options (constructor takes the model list) optimizer CostOptimizer(models[ ModelConfig( namegpt-4, provideropenai, cost_per_1k_input_tokens0.03, cost_per_1k_output_tokens0.06, avg_latency_ms900, quality_score0.95, ), ModelConfig( namegpt-3.5-turbo, provideropenai, cost_per_1k_input_tokens0.002, cost_per_1k_output_tokens0.004, avg_latency_ms300, quality_score0.80, ), ]) # Get optimization for a task task TaskProfile( task_typesummarization, avg_input_tokens1500, avg_output_tokens500, min_quality0.75, max_latency_ms2000, ) result optimizer.recommend(task) print(fRecommended: {result.recommendations[0].model_name}) print(fEstimated cost: ${result.recommendations[0].estimated_cost:.4f}) if result.potential_savings_pct is not None: print(fSavings vs default: {result.potential_savings_pct:.0f}%)ModelConfig与TaskProfile均为 Pydantic 模型optimizer.py字段如下ModelConfigname、provider、cost_per_1k_input_tokens、cost_per_1k_output_tokens、avg_latency_ms、quality_score0~1TaskProfiletask_type、avg_input_tokens、avg_output_tokens、min_quality0~1、max_latency_ms可选。recommend()的推荐逻辑是筛选出质量 ≥min_quality且延迟 ≤max_latency_ms的可行模型按估算成本升序排列最优者标记is_optimalTrue若传入current_model还会计算相对当前模型的潜在节省百分比potential_savings_pct。CostOptimizer还提供三个进阶方法# Pareto 前沿分析没有其他模型既更便宜又更高质量 frontier optimizer.pareto_frontier(task) # 批量模拟按请求量投影总成本 projection optimizer.simulate(task, model_namegpt-3.5-turbo, volume10000) print(fTotal cost for 10k requests: ${projection[total_cost]}) # 路由建议为多个任务类型分别推荐最便宜模型 routing optimizer.suggest_routing([task, TaskProfile(task_typecoding, ...)])其中simulate会抛出KeyError未知模型名pareto_frontier返回按成本升序的 Pareto 最优集合——测试 tests/test_cost_optimizer.py 使用gpt-4o-mini、gpt-4o、claude-opus三档模型对推荐、Pareto 前沿、模拟与路由建议做了完整验证。预算配置参考完整参数组合如下默认值已在上文表格中给出CostGuard( per_task_limit2.0, # Max cost per single task per_agent_daily_limit100.0, # Max daily spend per agent org_monthly_budget5000.0, # Global monthly cap anomaly_detectionTrue, # Enable anomaly detection auto_throttleTrue, # Auto-throttle and kill kill_switch_threshold0.95, # Kill at 95% utilization alert_thresholds[ # Alert at these percentages 0.50, 0.75, 0.90, 0.95 ], )API 参考CostGuard方法说明check_task(agent_id, estimated_cost)预检任务是否在预算内咨询性不预留额度check_and_charge(agent_id, task_id, cost_usd)原子地检查 记账并发安全返回(allowed, reason, alerts)record_cost(agent_id, task_id, cost_usd, breakdownNone)记录实际成本并返回触发的告警列表get_budget(agent_id)获取单个 Agent 的预算状态get_all_budgets()获取所有 Agent 的预算状态通过summary()的agents字段get_alerts()获取全部已触发告警通过alerts属性get_summary()获取组织级成本汇总通过summary()方法reset_daily(agent_idNone)重置单 Agent 或全部 Agent 的当日预算清零支出、任务数、throttle/kill 状态AgentBudget字段类型说明spent_today_usdfloat今日累计支出remaining_today_usdfloat今日剩余额度max(0, daily_limit - spent)utilization_percentfloat当日利用率0~100%throttledbool利用率 ≥ 85% 时置位killedbool利用率 ≥ kill 阈值默认 95%时置位task_count_todayint今日任务数avg_cost_per_taskfloat平均单任务成本CostAlert字段说明severityCostAlertSeverityinfo/warning/criticalactionBudgetActionalert/throttle/killmessage人类可读的告警消息current_value/threshold当前值与该告警对应的阈值agent_id/timestamp归属 Agent 与时间戳所有告警与记录对象均提供to_dict()序列化方法便于接入审计链路。生命周期与运维要点每日重置CostGuard不会自动感知新的一天需要由外部调度如 cron在每日开始时调用reset_daily()清零所有 Agent 的当日支出、任务计数与 throttle/kill 状态见 guard.py。组织月度预算则依赖org_spent_month的滚动累计跨月时需要重建CostGuard或自行重置。并发安全CostGuard内部使用threading.Lock保护所有预算状态record_cost、check_task、check_and_charge均可在多线程环境下安全调用有test_concurrent_record_cost_no_crash覆盖。审计序列化CostRecord、CostAlert、AgentBudget、AnomalyResult、BaselineStats均实现to_dict()成本事件可直接落入审计存储。与可观测性体系的结合成本治理不是孤岛。将CostGuard与仓库中其他 SRE 能力串联可以构建完整的成本 可靠性观测闭环结合 SLO 与错误预算成本治理与服务质量目标、错误预算相辅相成——成本告警可视为 SLO 之外的财务健康指标参见 Tutorial 05 - Agent Reliability (SRE)结合多 Agent 策略将成本上限与集体限流collective rate limiting组合使用防止成本合规但调用频次爆表参见 Tutorial 49 - Multi-Agent Policies结合可观测性与链路追踪将成本事件与 OpenTelemetry 链路关联在排障时同时看到哪个任务花了多少钱与调用链路全貌参见 Tutorial 13 - Observability Tracing设计背景成本治理的整体架构决策分层成本元数据、post-action 执行、软/硬上限分级详见 ADR 0012仓库配套示例agent-sre的集成层还提供了面向 LangChain、OpenTelemetry、Prometheus 等生态的导出器见 agent_sre/integrations可将成本指标接入既有监控体系。下一步本文完整覆盖了CostGuard的预算设置、分级告警、自动限流熔断、组织级预算、异常检测与成本优化。建议下一步在生产环境先以仅告警、不熔断auto_throttleFalse观察一周真实成本曲线再逐步开启自动限流与熔断为预算关键路径统一改用check_and_charge原子原语避免并发超支通过 docs/tutorials/05-agent-reliability.md 与 docs/tutorials/13-observability-and-tracing.md 将成本治理纳入完整的 SRE 观测体系。赞分享人工智能AI AgentAI 安全治理策略引擎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点击查看免费下载相关推荐Agent Governance Toolkit 成本治理实战用 CostGuard 实现分级预算、自动节流与成本异常检测Agent Governance Toolkit 成本治理实战用 CostGuard 实现分级预算、自动节流与成本异常检测 本指南以 examples/cos人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权Agent Governance Toolkit A2A 对话治理实战技能级访问控制、信任评分与反馈循环熔断Agent Governance Toolkit A2A 对话治理实战技能级访问控制、信任评分与反馈循环熔断 本文对应仓库教程 Tutorial 44 — A人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权VueGL响应式交互设计从表单输入到3D场景控制VueGL响应式交互设计从表单输入到3D场景控制 VueGL是一个基于Vue.js和three.js的3D WebGL图形渲染组件库它能够帮助开发者轻松实现人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考