用时序策略保护 Amazon Bedrock AgentCore:让 AI 智能体按正确顺序行动

2026-08-07 61 预计阅读时间: 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 分钟

传统授权通常只回答一个问题:当前主体能不能调用这个工具。但对 AI 智能体来说,仅检查单次调用往往不够。一次转账本身可能合法,可如果智能体尚未验证收款账户、没有获得用户确认,或者本次会话的累计金额已经超限,这次调用仍然不应该执行。

Amazon Bedrock AgentCore 的时序策略(temporal policies)把会话历史纳入授权判断。策略不再只看“现在要做什么”,还会检查“之前发生过什么”,从而约束工作流顺序、阻止无依据的数据进入后续操作、限制财务风险,并在高价值动作前要求人工批准。

从静态权限转向会话级约束

静态访问控制适合表达稳定权限,例如客服智能体可以读取订单,但不能修改支付记录。时序策略处理的是另一类问题:同一个动作在不同会话状态下,授权结果可能不同。

以退款流程为例,合理的调用顺序可能是:

  1. 查询订单,并获得可信系统返回的订单数据。
  2. 验证订单状态和可退款金额。
  3. 获得用户确认。
  4. 金额超过阈值时,等待人工审批。
  5. 执行退款。

这里的关键不是让模型“记住流程”,而是在执行工具调用的边界上强制检查流程。提示词可以指导模型,但不能承担最终授权责任;智能体可能误解上下文、受到提示注入影响,或者根据缺失信息生成看似合理的参数。时序策略需要把关键前置条件变成系统可验证的事实。

这类策略通常围绕三种信息做判断:

  • 当前动作:准备调用哪个工具,参数和金额是多少。
  • 会话历史:此前调用过哪些工具,哪些调用成功,返回了什么受信任结果。
  • 累计状态:本次会话已经执行多少次敏感操作,累计金额是多少,是否存在有效审批。

四类值得优先落地的规则

1. 强制工作流顺序

智能体不能绕过必要步骤。例如,只有在身份验证成功后才能修改地址;只有在库存检查完成后才能提交订单。

规则应检查成功事件,而不只是检查“工具曾经被调用”。一次失败的身份验证不能成为后续动作的授权依据。

2. 阻止数据编造进入执行层

模型生成的账号、订单号或报价不应自动变成可信数据。敏感动作使用的关键字段,应该能够追溯到当前会话中的可信工具响应,或者来自经过验证的用户输入。

例如,创建发票前可以要求客户 ID 必须出现在客户系统查询结果中。这样即使模型生成了格式正确但不存在的客户 ID,执行层也会拒绝请求。

3. 限制累计财务暴露

单笔限额无法覆盖多次小额操作造成的风险。时序策略可以统计会话内已经批准或执行的金额,并对累计值设限。

需要提前定义统计口径:失败交易是否计入、撤销操作是否冲销额度、并行调用如何避免竞争条件,以及限额按会话、用户还是更长的时间窗口计算。高风险场景不能只依赖智能体本地内存维护计数。

4. 对高价值动作要求人工审批

达到阈值的退款、采购、转账或基础设施变更,可以进入暂停状态,等待人工审批事件。审批记录至少应绑定动作类型、关键参数、金额和有效期,防止智能体拿一张旧审批去执行另一项操作。

审批不是一个简单的布尔值。更稳妥的设计是把审批视为有作用域、可过期、只能消费一次的授权凭证。

一个可改造的策略与执行示例

下面的 YAML 是说明性策略模型,不代表 AgentCore 的固定配置语法。它展示了如何把顺序、数据来源、累计限额和人工审批写成可审查的规则;接入时需要映射到实际的 AgentCore 策略接口和事件结构。

version: 1
workflow: vendor_payment

rules:
  - id: verified-vendor-required
    action: payment.execute
    require:
      successful_event:
        type: vendor.lookup
        match:
          result.vendor_id: request.vendor_id

  - id: session-exposure-limit
    action: payment.execute
    require:
      expression: session.sum("payment.execute", "amount", status="succeeded") + request.amount <= 10000

  - id: human-approval-for-high-value-payment
    action: payment.execute
    when:
      expression: request.amount >= 5000
    require:
      approval:
        type: finance.payment
        match:
          subject_id: request.vendor_id
          amount: request.amount
        max_age_seconds: 1800
        single_use: true

可以先在应用侧实现一个最小授权网关,以便验证事件模型和拒绝逻辑。下面的 Python 示例可直接运行,它模拟了三项检查:供应商必须来自可信查询结果、会话累计付款不得超过 10000、高于或等于 5000 的付款必须有匹配审批。

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Session:
    events: list[dict[str, Any]] = field(default_factory=list)

    def successful_payments(self) -> float:
        return sum(
            float(event["amount"])
            for event in self.events
            if event.get("type") == "payment.execute"
            and event.get("status") == "succeeded"
        )

    def vendor_was_verified(self, vendor_id: str) -> bool:
        return any(
            event.get("type") == "vendor.lookup"
            and event.get("status") == "succeeded"
            and event.get("vendor_id") == vendor_id
            for event in self.events
        )

    def has_matching_approval(self, vendor_id: str, amount: float) -> bool:
        return any(
            event.get("type") == "finance.payment.approved"
            and event.get("vendor_id") == vendor_id
            and float(event.get("amount", -1)) == amount
            and not event.get("consumed", False)
            for event in self.events
        )


def authorize_payment(session: Session, vendor_id: str, amount: float) -> None:
    if not session.vendor_was_verified(vendor_id):
        raise PermissionError("Vendor must be verified in this session")

    if session.successful_payments() + amount > 10_000:
        raise PermissionError("Session payment limit exceeded")

    if amount >= 5_000 and not session.has_matching_approval(vendor_id, amount):
        raise PermissionError("Matching human approval is required")


session = Session(events=[
    {
        "type": "vendor.lookup",
        "status": "succeeded",
        "vendor_id": "vendor-42",
    },
    {
        "type": "finance.payment.approved",
        "vendor_id": "vendor-42",
        "amount": 6_000,
        "consumed": False,
    },
])

authorize_payment(session, vendor_id="vendor-42", amount=6_000)
print("Authorized")

运行方式:

python temporal_policy_demo.py

生产实现还需要在支付成功后原子地写入事件,并把审批标记为已消费。否则,并行工具调用可能同时通过检查,突破累计限额或重复使用同一审批。

策略执行边界决定安全效果

时序规则应该部署在工具真正执行之前,而不是只作为提示词的一部分。模型负责提出动作,策略层负责裁决,工具层只接受带有有效授权结果的请求。

事件来源也必须区分信任级别。模型在对话中声称“用户已经批准”不等于审批事件;工具返回的结构化记录、身份系统签发的证明或人工审批服务产生的凭证,才适合作为授权依据。

还要避免把完整敏感响应无节制地写入历史。策略通常只需要订单 ID、金额、状态、哈希或审批标识等最小字段。事件日志应设置访问控制、保留周期和脱敏规则,并为拒绝结果记录规则 ID,方便审计和排障。

落地时的检查清单

采用时序策略时,可以从少量高风险动作开始,而不是一次覆盖所有工具:

  • 列出会造成资金、数据或基础设施变化的工具。
  • 为每个动作定义必须先发生的可信事件。
  • 明确关键参数必须来自哪个数据源。
  • 同时设置单笔限额和累计限额。
  • 让人工审批绑定具体动作、参数、金额和有效期。
  • 使用原子状态更新处理并发执行。
  • 测试跳步、失败后继续、重复审批、参数替换和并行调用。
  • 默认拒绝历史不完整、状态冲突或策略无法求值的请求。

时序策略的价值在于把“智能体应该遵守流程”变成“系统只允许符合流程的动作”。对于能够转账、退款、修改数据或操作云资源的智能体,这种会话级授权应当与身份权限、工具参数校验和审计日志共同组成执行边界,而不能只依赖模型自身的判断。


相关推荐