Agent 应用跑起来之后，安全问题是这样的：
你的 Agent 可能在一次请求里跑 20-30 步（规划→调工具→处理结果→再规划→再调工具……）。每一步的风险不一样——用户输入那步要防注入，工具调用那步要防越权，输出那步要过滤敏感信息。
之前的 Bedrock Guardrails 需要你先创建一个 guardrail 资源、配好版本、绑到模型调用上。问题是：
- 一个 guardrail 配置很难覆盖所有步骤（每步风险不同）
- Agent 循环是动态的，你不知道它会跑多少步
- 管理多个 guardrail 版本很烦
现在新出的 InvokeGuardrailChecks API 换了个思路：不创建资源，每次调用时直接告诉它检查什么。
这个 API 干什么的
一句话：你在 Agent 循环的任意位置插一个 API 调用，告诉它"帮我检查这段文本有没有XXX问题"，它返回检测结果（含置信度分数），你自己决定怎么处理。
关键特性：
- 无资源模式 —— 不需要创建 guardrail ID、不需要管版本
- 检测模式 —— 只返回结果，不自动 block（你自己决定 block/pass/log）
- 单项检查 —— 可以只跑"注入检测"不跑"内容过滤"，按需组合
- 数值分数 —— 返回 0-1 的置信度，不是简单的"通过/拒绝"
支持的检查类型
| 类型 | 检查项 | 用途 |
|---|---|---|
| 内容过滤 | 仇恨/暴力/色情/侮辱/不当行为 | 过滤有害输出 |
| 注入检测 | Jailbreak / Prompt Injection / Prompt Leakage | 防护用户输入攻击 |
| 敏感信息 | PII 实体（姓名/电话/身份证/银行卡等） | 防止数据泄露 |
在 Agent 循环里怎么用
import boto3
import jsonbedrock = boto3.client("bedrock-runtime", region_name="us-east-1")def check_prompt_attack(text: str) -> dict:"""检查用户输入是否有注入攻击"""response = bedrock.invoke_guardrail_checks(source="INPUT",content=[{"text": {"text": text}}],checks={"promptAttackCheck": {"attackTypes": ["JAILBREAK", "PROMPT_INJECTION", "PROMPT_LEAKAGE"]}})return response["results"]def check_content_safety(text: str) -> dict:"""检查模型输出是否有有害内容"""response = bedrock.invoke_guardrail_checks(source="OUTPUT",content=[{"text": {"text": text}}],checks={"contentFilter": {"categories": ["HATE", "VIOLENCE", "SEXUAL", "INSULTS", "MISCONDUCT"]}})return response["results"]def check_pii(text: str) -> dict:"""检查输出是否泄露了敏感信息"""response = bedrock.invoke_guardrail_checks(source="OUTPUT",content=[{"text": {"text": text}}],checks={"sensitiveInformationFilter": {"piiEntityTypes": ["PHONE", "EMAIL", "SSN", "CREDIT_CARD", "NAME"]}})return response["results"]
完整 Agent 循环示例
class SafeAgent:def __init__(self):self.bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")self.max_steps = 20self.attack_threshold = 0.7self.content_threshold = 0.8self.pii_threshold = 0.9def run(self, user_input: str) -> str:# 第 1 步：检查用户输入input_check = self.check_input(user_input)if input_check["blocked"]:return f"请求被拒绝：{input_check['reason']}"messages = [{"role": "user", "content": user_input}]for step in range(self.max_steps):# 调用模型model_response = self.invoke_model(messages)# 第 2 步：检查模型输出安全性output_check = self.check_output(model_response)if output_check["blocked"]:return "抱歉，无法生成安全的回复。"# 如果模型要调工具if model_response.get("tool_use"):tool_result = self.execute_tool(model_response["tool_use"])# 第 3 步：检查工具返回结果是否有 PIIpii_check = self.check_pii_in_tool_output(tool_result)if pii_check["has_pii"]:tool_result = self.redact_pii(tool_result, pii_check)messages.append({"role": "tool", "content": tool_result})else:return model_response["text"]return "处理步骤超限，请简化你的请求。"def check_input(self, text: str) -> dict:response = self.bedrock.invoke_guardrail_checks(source="INPUT",content=[{"text": {"text": text}}],checks={"promptAttackCheck": {"attackTypes": ["JAILBREAK", "PROMPT_INJECTION"]}})results = response["results"]max_score = max((r.get("confidenceScore", 0) for r in results.get("promptAttackResults", [])),default=0)return {"blocked": max_score > self.attack_threshold,"reason": "检测到潜在的注入攻击" if max_score > self.attack_threshold else None,"score": max_score}def check_output(self, response: dict) -> dict:text = response.get("text", "")if not text:return {"blocked": False}check_response = self.bedrock.invoke_guardrail_checks(source="OUTPUT",content=[{"text": {"text": text}}],checks={"contentFilter": {"categories": ["HATE", "VIOLENCE", "SEXUAL", "INSULTS"]}})results = check_response["results"]max_score = max((r.get("severityScore", 0) for r in results.get("contentFilterResults", [])),default=0)return {"blocked": max_score > self.content_threshold}def check_pii_in_tool_output(self, tool_output: str) -> dict:response = self.bedrock.invoke_guardrail_checks(source="OUTPUT",content=[{"text": {"text": tool_output}}],checks={"sensitiveInformationFilter": {"piiEntityTypes": ["PHONE", "EMAIL", "SSN", "CREDIT_CARD"]}})results = response["results"]pii_entities = results.get("sensitiveInformationResults", [])return {"has_pii": len(pii_entities) > 0,"entities": pii_entities}def redact_pii(self, text: str, pii_check: dict) -> str:"""把检测到的 PII 替换成占位符"""for entity in pii_check["entities"]:text = text.replace(entity["match"], f"[{entity['type']}]")return text
和原有 Guardrails 的区别
| 维度 | 原有 Guardrails | InvokeGuardrailChecks |
|---|---|---|
| 资源管理 | 需创建 guardrail + version | 无需创建任何资源 |
| 绑定方式 | 和模型调用绑定 | 独立调用，任意位置插入 |
| 执行模式 | 自动 block/allow | 仅检测，返回分数 |
| 灵活度 | 一个配置管所有 | 每次调用可以不同检查组合 |
| 适用场景 | 简单的输入/输出过滤 | 复杂 Agent 循环的精细化控制 |
| 版本管理 | 需要维护 version | 无版本概念 |
什么时候用新 API
用 InvokeGuardrailChecks 的场景：
- 多步 Agent（每步风险不同，需要不同检查策略）
- 快速原型（不想先创建资源就能用）
- 需要自定义阈值和处理逻辑
- 检测结果要记日志做分析
继续用原有 Guardrails 的场景：
- 简单的输入→模型→输出链路
- 团队不想自己写判断逻辑
- 需要统一的 block 行为
可用区域
US East (N. Virginia)、US East (Ohio)、US West (Oregon)、Europe (London)、Europe (Stockholm)、Asia Pacific (Tokyo)、Asia Pacific (Sydney)。
上手建议
- 先从注入检测开始 —— 这是 Agent 安全的核心防线
- 设合理阈值 —— 0.7-0.8 做告警，0.9+ 做 block，根据业务调
- 记录所有检测日志 —— 分数数据是后续调优的基础
- 别每步都全量检查 —— 输入侧查注入，输出侧查内容和 PII，工具返回查 PII
文档：https://docs.aws.amazon.com/bedrock/latest/userguide/guardrails-use-invoke-guardrail-checks.html
Bedrock Guardrails 概览：https://aws.amazon.com/bedrock/guardrails/