云成本控制过去主要防范人的误操作:工程师开错实例、忘记关闭测试环境,账单告警通常还有时间介入。但当持有云凭证的是 AI Agent 或攻击者时,资源调用可以在几分钟内并发放大,而成本数据和预算通知可能要到数小时甚至约一天后才反映出来。
一支三人团队在静态 AWS Access Key 被提取后,一天内收到了 14,000 美元账单,攻击者主要消耗了 Amazon Bedrock 上的 Claude 调用。另一起发生在 5 月的 DN42 事件中,自治 Agent 在 24 小时内创建了 6,531 美元的超规格基础设施。这两个案例指向同一个问题:预算告警是观测工具,不是实时熔断器。
账单告警为什么拦不住 Agent
传统成本流程通常是:服务产生用量、计量系统聚合数据、账单系统更新、预算规则计算,最后通过邮件或 SNS 发出通知。这个链路适合处理按天巡检的人工操作,却无法跟上自动化调用的速度。
Agent 还会放大三类风险:
- 高调用频率:模型推理、实例创建和批处理任务可以连续并发执行。
- 错误自我强化:Agent 可能把失败理解为容量不足,继而申请更大实例或重试更多次。
- 凭证失窃后的自动消费:静态密钥一旦泄露,攻击者无需进入控制台即可脚本化消耗 API。
因此,Cost Explorer、AWS Budgets 和账单邮件仍然有价值,但它们属于事后发现层。真正的控制必须落在调用发生之前或调用路径之中。
不要把无限制云凭证交给 Agent
第一道边界是身份设计。Agent 不应持有长期 Access Key,也不应继承开发者或管理员角色。更稳妥的做法是通过工作负载身份获取短期凭证,并为每种任务建立独立角色。
权限策略需要同时约束服务、区域、资源和动作。下面是一份可改造的 IAM 策略示例:它只允许在指定区域调用一个获批的 Bedrock 模型,并显式拒绝其他区域。运行前请替换 REGION、ACCOUNT_ID 和 MODEL_OR_INFERENCE_PROFILE_ARN。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowApprovedBedrockModel",
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
],
"Resource": "MODEL_OR_INFERENCE_PROFILE_ARN",
"Condition": {
"StringEquals": {
"aws:RequestedRegion": "REGION"
}
}
},
{
"Sid": "DenyBedrockOutsideApprovedRegion",
"Effect": "Deny",
"Action": "bedrock:*",
"Resource": "*",
"Condition": {
"StringNotEquals": {
"aws:RequestedRegion": "REGION"
}
}
}
]
}
可以将它保存为 agent-bedrock-policy.json,先用策略模拟器验证,再创建托管策略:
aws iam create-policy \
--policy-name AgentBedrockRestricted \
--policy-document file://agent-bedrock-policy.json
这份策略控制了模型和区域,但没有直接限制花费。模型 ARN 的具体形式会随基础模型、推理配置和 AWS 区域变化,应以目标账户中实际可用的资源为准。
在调用路径中加入同步预算
如果 Agent 可以直接调用 Bedrock,账单系统就成了唯一的成本反馈渠道。更可控的结构是在 Agent 与云 API 之间放置一个调用代理,由代理执行身份校验、并发限制、单任务 token 上限和内部额度扣减。
下面的 Python 示例演示一个进程内的同步护栏。它不依赖延迟账单,而是在每次调用前预留 token 额度。运行前安装 boto3,配置短期 AWS 凭证,并替换模型 ID。示例中的额度是 token 数量,不是美元价格;生产环境可按已批准模型的价格表换算内部成本。
import os
import threading
import boto3
REGION = os.environ.get("AWS_REGION", "us-east-1")
MODEL_ID = os.environ["BEDROCK_MODEL_ID"]
TOKEN_BUDGET = int(os.environ.get("TOKEN_BUDGET", "20000"))
MAX_OUTPUT_TOKENS = int(os.environ.get("MAX_OUTPUT_TOKENS", "512"))
client = boto3.client("bedrock-runtime", region_name=REGION)
lock = threading.Lock()
remaining_tokens = TOKEN_BUDGET
def invoke(prompt: str) -> str:
global remaining_tokens
estimated_input = max(1, len(prompt) // 4)
reservation = estimated_input + MAX_OUTPUT_TOKENS
with lock:
if reservation > remaining_tokens:
raise RuntimeError("Agent token budget exhausted")
remaining_tokens -= reservation
try:
response = client.converse(
modelId=MODEL_ID,
messages=[
{"role": "user", "content": [{"text": prompt}]}
],
inferenceConfig={"maxTokens": MAX_OUTPUT_TOKENS}
)
return response["output"]["message"]["content"][0]["text"]
except Exception:
with lock:
remaining_tokens += reservation
raise
if __name__ == "__main__":
print(invoke("用三句话说明为什么静态云密钥不适合 AI Agent。"))
python -m pip install boto3
export AWS_REGION=us-east-1
export BEDROCK_MODEL_ID='替换为已批准的模型或推理配置 ID'
export TOKEN_BUDGET=20000
python guardrail.py
这个示例适合说明控制思路,但进程重启会重置额度,多实例部署也会各自记账。生产环境应把额度和幂等记录放进 DynamoDB、Redis 或事务数据库,并使用原子扣减。更重要的是,Agent 的 IAM 角色必须只能访问这个代理,不能绕过代理直接调用 Bedrock,否则应用层额度只是建议,不是安全边界。
把预算拆成多层熔断器
可靠方案通常包含多道相互独立的限制:
- 用短期凭证替代静态 Access Key,并自动轮换和撤销会话。
- 用 IAM、权限边界或 AWS Organizations SCP 限定可调用服务、模型、区域和实例规格。
- 在代理层限制每分钟请求数、并发数、最大输出 token 和单任务累计额度。
- 对创建 GPU、超大 EC2 实例、批量集群等高成本动作要求人工批准。
- 同时配置 AWS Budgets、异常检测和 CloudTrail 告警,承担发现与取证职责。
- 准备可立即执行的停机动作,例如撤销角色会话、禁用调用代理或应用紧急拒绝策略。
需要注意,硬限制会降低 Agent 的自治程度,也可能中断合法的突发任务。团队应按任务风险分级:低成本、可回滚操作可以自动执行;高成本或不可逆操作必须经过审批。不要把预算邮件当作最后一道防线,因为在 Agent 的时间尺度上,它抵达时消费可能早已完成。