用时间策略和网关限流,把 Bedrock AgentCore 的智能体成本关进笼子

2026-08-07 50 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:11 分钟

智能体的风险往往不在某一次工具调用,而在一连串看似合理的动作中累积:重复检索、反复重试、持续调用高成本模型,最终让一次任务越过预算边界。Amazon Bedrock AgentCore 新增的时间策略(temporal policies)和网关限流(rate limiting),把控制范围从“单次请求是否允许”扩展到“这段时间内的行为序列是否合规”。

其中,时间策略由开源的 Dogwood 策略语言驱动,可用于表达跨步骤、跨时间窗口的约束;网关限流则为流量和调用频率提供了更直接的成本护栏。两者组合后,即使智能体因为提示词、工具返回结果或重试逻辑而改变执行路径,预算上限仍然可以由平台侧确定性地执行。

单动作授权解决不了行为序列问题

传统的工具授权通常回答一个问题:agent 现在能不能调用 search_orders?这对权限隔离很重要,但它无法约束下面这类过程:

  • 智能体在十分钟内连续执行几十次检索,再把结果交给模型反复总结。
  • 一个失败的外部 API 触发重试循环,导致模型调用和工具请求同步膨胀。
  • 多个并发会话都通过了单次授权检查,但共同耗尽了团队或租户的成本额度。

时间策略关注的是事件之间的关系。例如:某类高成本操作是否在窗口期内超过次数、敏感操作是否在前置审批之后发生、失败后是否出现过量重试。它让控制逻辑能够描述“先发生什么、在多久内发生多少次、之后还能不能继续”,而不是只判断一个孤立动作。

这类能力尤其适合客服自动化、运维代理、数据分析代理等会多轮调用工具的工作流。策略不需要猜测模型下一步会做什么,而是明确规定无论它如何选择,哪些行为轨迹不可接受。

时间策略与限流分别管什么

这两项能力有重叠,但职责不同,部署时应一起设计。

控制手段 主要约束对象 适合解决的问题
时间策略 一段时间内的事件顺序、次数与状态 重试风暴、审批绕过、跨步骤预算规则
网关限流 请求速率、突发流量、并发压力 单租户打满配额、异常客户端、高频工具调用

网关限流是第一道稳定的流量闸门。它可以让同一 API key、租户或调用方在固定窗口中最多发起一定数量的请求,防止短时间突发直接穿透到模型和下游工具。

时间策略则应承接更贴近业务语义的规则。例如,给“生成完整报告”这类昂贵动作设置每个会话的次数限制,或规定“付款操作必须在人工批准事件后 30 分钟内执行”。这类规则仅靠 HTTP 请求计数难以可靠表达。

可以这样实践:给高成本工具加双层护栏

下面示例展示一种可改造的实现思路:在应用层为每个会话记录昂贵工具调用,并在 API 网关层限制客户端速率。

假设说明:Dogwood 的具体策略声明和 AgentCore 资源配置应以当前 AWS 文档及控制台/API 的实际字段为准。下方策略文件用接近声明式策略的伪代码展示应表达的约束,不应直接当作生产配置提交。Python 示例可以直接运行,用于在接入 AgentCore 前验证策略思路。

先创建一个本地的会话预算守卫。它限制单个会话在 10 分钟内最多调用 3 次昂贵报告工具,并对失败重试设置单独上限。

# budget_guard.py
from collections import defaultdict, deque
from dataclasses import dataclass
from time import time

WINDOW_SECONDS = 10 * 60
MAX_REPORTS_PER_WINDOW = 3
MAX_FAILURES_PER_WINDOW = 2


@dataclass
class Event:
    timestamp: float
    action: str
    success: bool


class SessionBudgetGuard:
    def __init__(self):
        self.events = defaultdict(deque)

    def allow(self, session_id: str, action: str) -> bool:
        now = time()
        history = self.events[session_id]

        while history and now - history[0].timestamp > WINDOW_SECONDS:
            history.popleft()

        reports = sum(event.action == "generate_full_report" for event in history)
        failures = sum(
            event.action == action and not event.success
            for event in history
        )

        if action == "generate_full_report" and reports >= MAX_REPORTS_PER_WINDOW:
            return False
        if failures >= MAX_FAILURES_PER_WINDOW:
            return False
        return True

    def record(self, session_id: str, action: str, success: bool) -> None:
        self.events[session_id].append(Event(time(), action, success))


if __name__ == "__main__":
    guard = SessionBudgetGuard()
    session_id = "support-case-42"

    for attempt in range(1, 5):
        action = "generate_full_report"
        if not guard.allow(session_id, action):
            print(f"attempt={attempt}: blocked by budget policy")
            continue

        # Replace this with the actual AgentCore tool invocation.
        success = True
        guard.record(session_id, action, success)
        print(f"attempt={attempt}: report generated")

运行:

python budget_guard.py

预期结果是前三次报告生成被允许,第四次被拒绝。生产环境中,事件状态不应仅放在进程内存:多实例服务应使用具备 TTL 的共享存储,或者将同类规则下沉到 AgentCore 的时间策略中,由平台统一判定。

对应的策略意图可以写成如下形式:

# policy.dogwood (illustrative pseudocode)
policy "session-report-budget" {
  when action == "generate_full_report"
  deny if count(action == "generate_full_report", within: 10m, by: session_id) > 3
}

policy "retry-ceiling" {
  when action == "call_external_tool"
  deny if count(result == "failure", within: 10m, by: session_id) > 2
}

与此同时,在网关上按租户或客户端身份实施限流。以下 YAML 同样是部署意图示例:关键点是让限流键绑定到可信身份,例如 API key、JWT 中的租户声明或已认证的调用方,而不是容易伪造的请求头。

# gateway-rate-limit.yaml
rateLimits:
  - name: tenant-agent-requests
    key: authenticated_tenant_id
    window: 1m
    maxRequests: 120
  - name: report-tool-burst-control
    key: authenticated_tenant_id
    route: /tools/generate-full-report
    window: 1m
    maxRequests: 10

需要按实际流量调整两个阈值。一个常见起点是:网关限制保护基础设施和共享配额,时间策略限制单任务、单会话或单业务实体的累积行为。不要只按全局账户做限流,否则一个活跃租户仍可能挤占其他客户的服务能力。

把拒绝变成可观测事件

策略拒绝不应只是返回一个模糊的失败。应用需要把至少以下字段写入日志、指标或审计事件:

  • 策略名称和版本。
  • 调用方、租户、会话和任务标识。
  • 被拒绝的动作及当前窗口内的计数。
  • 触发时间、请求成本标签或模型类别。

对智能体而言,还要区分“可恢复的限流”和“终止性的策略拒绝”。对于前者,可以在明确的退避次数内等待后重试;对于后者,应结束当前路径或转交人工,不能让模型不断尝试等价调用来规避限制。将策略结果以结构化状态反馈给编排器,也能避免它把拒绝误判为普通工具故障。

落地时的检查清单

时间策略和网关限流并不替代模型选择、工具权限或成本监控,但它们补上了智能体系统最容易失控的维度:随时间累积的行为。

上线前可以检查以下几点:

  • 为高成本模型、写操作和外部 API 定义明确的“每会话/每租户/每任务”上限。
  • 将重试次数、窗口长度和退避策略作为同一套规则评审,避免重试配置绕过预算意图。
  • 限流键使用经过认证的身份字段,并验证匿名流量的处理方式。
  • 为拒绝事件建立告警,区分正常保护触发与策略阈值过低。
  • 在压测和故障演练中模拟工具持续失败、并发会话激增和模型循环调用。

智能体的非确定性不意味着成本控制也必须是概率性的。把行为序列交给时间策略,把流量入口交给网关限流,才能让系统在复杂执行路径下仍保有清晰、可审计的边界。


相关推荐