Best practices for applying Amazon Bedrock Guardrails to code generation workflows

将 Amazon Bedrock 安全防护措施应用于代码生成工作流的最佳实践 | 人工智能

将 Amazon Bedrock 安全防护措施应用于代码生成工作流的最佳实践

本文延续我们关于 Amazon Bedrock 安全防护措施最佳实践的系列文章。如需查看上一篇文章，请参阅《像专业人士一样构建安全的生成式 AI 应用：Amazon Bedrock 安全防护措施最佳实践》。

由 AI 驱动的代码助手和代码生成工作流（如 Claude Code、Kiro 和 OpenAI Codex）正在改变开发人员编写软件的方式。这些工具通过流式响应实时生成代码，通常在长时间会话中生成数千个字符。随着组织大规模采用这些助手进行代码生成 AI 工作流，当需要时检测并阻止危险代码模式变得至关重要。Amazon Bedrock 安全防护措施通过内容过滤器（用于内容审核）、提示攻击预防（包括越狱、提示注入和提示泄露）、敏感信息过滤器（用于删除和阻止个人身份信息 PII）等防护机制，帮助检测和过滤危险及不期望的代码内容。

然而，代码工作流和智能代理循环具有独特的吞吐量特征，包括长流式输出、并发开发者会话和重复的上下文评估。如果未正确配置，将 Amazon Bedrock 安全防护措施应用于具有这些特征的工作流可能会导致诸如限流错误、成本增加和延迟不佳等限制。在本文中，我们将解释如何为使用代码助手的代码生成工作流配置 Amazon Bedrock 安全防护措施以克服这些限制。通过这些最佳实践，您可以构建一个高效的蓝图，帮助您进行有效的容量规划并实现全面的安全覆盖。

为什么安全防护对代码生成工作流至关重要

Amazon Bedrock 安全防护措施提供全面的防护机制，可检测并过滤用户输入和模型响应中的有害及不期望内容，帮助构建安全的生成式 AI 应用。虽然这些防护机制可以配置并实施在各种应用中，但有一些特定的防护措施可用于保护有害代码模式，例如：

• 检测提示攻击 – 阻止试图操纵模型生成恶意代码或绕过指令的尝试。

• 过滤敏感信息 – 可帮助阻止或隐藏敏感信息，如个人身份信息（PII）、用户输入和模型响应中的自定义正则表达式，以及在生成代码中出现前的硬编码 AWS 访问密钥、数据库连接字符串或私钥。

• 内容审核 – 检测并过滤用户提示和模型响应中的有害内容，包括违反组织可接受使用政策的内容。

• 阻止禁止主题 – 定义自定义主题作为阻止与禁止活动相关对话的基础，例如生成绕过认证机制的代码或协助未经授权的访问模式。

当 AI 生成的代码用于生产系统时，这些防护措施至关重要。

情景：当良好的安全防护措施出现偏差时

一个客户系统团队刚刚在Amazon Bedrock上为15名开发人员部署了Claude Code。他们配置了一个包含三个安全机制的防护措施：用于防止注入攻击的提示攻击检测、用于隐藏或阻止泄露凭证的敏感信息过滤器，以及用于阻止不安全代码模式的内容过滤器。在两名开发人员参与的试点阶段，所有功能都运行完美。试点成功后，所有15名开发人员同时开始编码会话。几分钟后，团队开始报告错误：来自Amazon Bedrock模型推理的ThrottlingException响应。代码补全在传输过程中突然中断。开发人员感到沮丧，他们的Slack频道瞬间被消息淹没。

发生了什么？每个开发人员的会话平均每个函数生成约5000个字符的代码。在默认的流式传输配置下，Amazon Bedrock会针对每50个字符评估一次防护措施，导致每个开发人员每个函数触发100次API调用。当15个并发会话同时运行时，系统每秒会产生1500次评估请求。更糟糕的是，由于他们启用了3个安全机制，每次评估会消耗3个文本单位而非1个，使吞吐量消耗增加三倍。

根本原因并非配额不足，而是架构不匹配。他们将原本为短对话设计的模式应用于高吞吐量的代码生成流水线。本文将提供架构模式来解决这个问题。

文本单位：防护措施的计量标准

在深入架构设计之前，理解防护措施消耗的计算方式至关重要，因为这是所有后续优化的关键。

文本单位被定义为文本中的1000个字符。当使用ApplyGuardrail API调用对包含1000个字符的文本进行评估，并应用3个安全机制时，会产生3个文本单位的消耗。

关键的一点是，消耗是乘法关系。它会随着内容长度和启用的安全机制数量同步增长。

在代码生成工作流中，输出通常非常冗长（每个函数可达数千个字符），这种乘法关系是容量规划中的关键因素。配置了内容过滤、禁止主题和敏感信息过滤的防护措施，其计费方式基于每个安全机制或过滤器处理的文本单位数量。

注意：内容过滤的计费方式为每1000个字符计为一个文本单位，无论过滤器中启用了多少类别（仇恨、侮辱、性内容、暴力、不当行为、提示攻击）。例如，如果启用了所有六个内容过滤类别，仍计为一个文本单位，而非六个。乘法关系适用于不同类型的策略（内容过滤、禁止主题、敏感信息过滤），而非同一策略类型内的类别。

挑战：代码生成工作流可能触发限流

传统的对话式AI工作流涉及简短、离散的用户提示和模型响应。代码生成工作流在影响防护措施架构的方面存在显著差异：

下表说明了代码生成工作流为何具有独特挑战性：

| 特性 | 对话式AI | 代码生成 | |------------------|------------------|----------------------| | 输出长度 | 100-500字符 | 5000-50000+字符 | | 会话时长 | 单轮或少数几轮 | 延长的多轮会话 | | 并发用户 | 通常异步 | 团队同时编码 | | 上下文复用 | 最小 | 系统提示、工具定义和先前代码在每轮中重复发送 |

中间输出

广泛的推理过程

这些差异意味着，采用内联扫描方法（即在生成过程中对每一块流式输出进行评估）会导致评估次数与实际安全价值的不匹配。

最佳实践：代码生成工作流的架构模式

为了解决这些挑战，我们建议采用一系列优化代码生成工作流中防护机制使用情况的架构模式。每个模式都针对问题的特定方面——从减少评估频率到仅选择性扫描高风险内容。您可以根据工作负载特征和安全要求单独应用这些模式或组合使用。

架构模式1：预提交钩子模型

理解内联扫描：默认方法

当您使用Converse或InvokeModel API通过guardrailConfig将防护机制直接附加到模型调用时，您使用的是所谓的内联扫描方法。在此模式下，Amazon Bedrock防护机制会实时连续评估完整输入提示和流式输出，与所有主动安全措施进行比对。对于传统的对话式AI，这种方法效果良好：提示简短，响应简洁，评估开销可以忽略不计。

但对于代码生成工作流，内联扫描会变得低效。它会增加成本并消耗配额，而不会显著提升您的安全态势。代码助手生成的不只是简短响应，而是会流式传输数千个字符的代码，其中夹杂着推理、注释和迭代优化。当启用内联评估时，输出的每个片段都会与所有配置的安全措施进行比对，包括静态系统提示和自上次评估以来未发生变化的先前生成上下文。结果是冗余工作：相同的模板指令、工具定义和先前对话历史会在每一轮中被重新评估，消耗文本单元却未增加安全价值。

核心最佳实践是将连续内联扫描转变为在战略检查点进行选择性验证，类似于Git工作流中的预提交钩子。与其在流式传输过程中扫描每个token，不如在风险特征发生变化的明确定义边界处验证内容。

为什么“预提交钩子”思路有效

软件开发人员不会在每次输入一行代码后就运行代码检查器、格式化工具和安全扫描器。他们会在提交时进行验证。他们不会验证每个中间编辑、每个删除的行或每个未完成的函数，而是在关键时刻验证最终结果。

在Git工作流中，预提交钩子是一个在特定明确定义边界自动运行的脚本：即您尝试将代码提交到仓库的时刻。这是您的更改成为共享代码库一部分前的最后防线。在此检查点，您一次性对即将持久化的最终产物运行所有验证和测试。

这种模式有效是因为它平衡了两种相互竞争的需求：彻底性（每个提交的更改都得到充分验证）和效率（仅在风险特征发生变化时进行验证，当代码从“进行中的草稿”转变为“即将持久化的产物”时）。

将该模式应用于Amazon Bedrock防护机制

核心最佳实践是将这一原则应用于 Amazon Bedrock Guardrails：从持续的内联扫描转变为在关键检查点进行选择性验证。

将代码生成流程视为具有自然的提交点，这些时刻内容会从一个信任级别过渡到另一个信任级别：

• 接收用户输入 – 内容从不可信来源进入系统（在此处验证）。

• 组装最终代码产物 – 模型的完整响应已准备好展示或保存（在此处验证）。

• 代码写入文件或提交到仓库 – AI 生成的内容即将变得持久化并可能被执行（在此处验证）。

在这些检查点之间，当模型进行推理、生成中间标记或生成推理过程解释时，内容是短暂的。它尚未跨越信任边界。持续扫描会增加成本并消耗配额，但不会显著提升安全态势。

关键原则：使用解耦的 ApplyGuardrail API 在内容到达模型前验证用户提供的输入，并在代码提交前验证聚合后的最终代码产物，而非每个中间推理标记。

实现：文件操作的提交前验证

对于最高风险的检查点，当 AI 生成的代码即将写入文件或提交到仓库时，执行全面的 guardrail 评估。这是直接应用的提交前钩子模式：

import hashlib
from pathlib import Path

class CodeCommitGuardrail:
    """
    在代码持久化前验证所有 AI 生成的代码更改。
    类似于在 Git 提交前钩子中运行安全扫描器。

    使用场景：
    - 编码助手将代码写入磁盘时
    - AI 生成的更改准备提交时
    - 生成的代码即将部署时
    """

    def __init__(self, client, guardrail_id, guardrail_version):
        self.client = client
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version
        self.validated_hashes = {}  # 缓存：无需重新验证未更改的文件

    def validate_file_changes(self, staged_files: dict) -> dict:
        """
        提交前评估所有已暂存的 AI 生成代码更改。

        参数:
            staged_files: {文件路径: 文件内容} 的字典，包含所有更改的文件

        返回:
            包含 'safe' 布尔值和发现违规项的字典
        """
        violations = []
        skipped = []

        for file_path, content in staged_files.items():
            # 跳过自上次验证后未更改的文件
            content_hash = hashlib.sha256(content.encode()).hexdigest()
            if self.validated_hashes.get(file_path) == content_hash:
                skipped.append(file_path)
                continue

            # 按 1,000 字符对齐块进行评估
            file_violations = self._evaluate_file(file_path, content)

            if file_violations:
                violations.extend(file_violations)
            else:
                # 缓存成功验证
                self.validated_hashes[file_path] = content_hash

        return {
            'safe': len(violations) == 0,
            'violations': violations,
            'files_evaluated': len(staged_files) - len(skipped),
            'files_skipped_cached': len(skipped)
        }

def _evaluate_file(self, file_path: str, content: str) -> list: """评估单个文件内容是否符合防护规则""" violations = []

for offset in range(0, len(content), 1000): chunk = content[offset:offset + 1000]

response = self.client.apply_guardrail( guardrailIdentifier=self.guardrail_id, guardrailVersion=self.guardrail_version, source='OUTPUT', content=[{'text': {'text': chunk}}] )

if response['action'] == 'GUARDRAIL_INTERVENED': violations.append({ 'file': file_path, 'offset': offset, 'length': len(chunk), 'assessments': response['assessments'], 'snippet': chunk[:200] + '...' if len(chunk) > 200 else chunk })

return violations

示例：与代码助手文件写入操作的集成

def write_ai_generated_code(file_path: str, content: str, guardrail: CodeCommitGuardrail): """ 在持久化文件前验证内容的文件写入包装器 """ result = guardrail.validate_file_changes({file_path: content})

if not result['safe']: print(f"防护规则阻止了对 {file_path} 的写入") for v in result['violations']: print(f" 偏移量 {v['offset']} 处的违规: {v['assessments']}") return False

安全写入

Path(file_path).write_text(content) print(f"{file_path} 验证通过并成功写入") return True

## 架构模式 2：将流式传输间隔增加到 1,000 个字符

当需要实时流式评估时（例如交互式编码会话中希望在检测到违规时立即停止生成），请优化流式传输间隔。将防护规则间隔从默认的 50 个字符增加到 1,000 个字符，可以将 API 调用量减少 20 倍。请参阅以下配置：

使用 InvokeModelWithResponseStream 与防护规则时

response = bedrock_runtime.invoke_model_with_response_stream( modelId='anthropic.claude-sonnet-4-20250514', body=json.dumps(request_body), guardrailIdentifier='your-guardrail-id', guardrailVersion='1',

关键配置：增加流式传输间隔

streamingConfigurations={ 'guardrailStreamingInterval': 1000 # 默认是 50！ } )

影响：5,000 字符的函数从 100 次评估减少到 5 次。50,000 字符的文件从 1,000 次评估减少到 50 次。

## 架构模式 3：使用独立的 ApplyGuardrail API 进行选择性评估

通过独立的 ApplyGuardrail API，您可以使用 Amazon Bedrock 防护规则，而无需依赖基础模型。您可以在不调用基础模型的情况下评估文本。

### 模式：仅输入验证与无防护推断

当您主要关注防止提示注入和阻止恶意输入时，在内容到达模型之前仅验证动态用户内容。在这种情况下，不使用内联防护规则调用模型：

import boto3 import json

bedrock_runtime = boto3.client('bedrock-runtime')

def process_coding_request(user_input: str, system_prompt: str, conversation_history: list): """ 通过防护机制验证用户输入，然后调用模型时不进行内联扫描。 系统提示和对话历史记录在每次交互中不会重新评估。 """

步骤1：仅评估新的动态用户输入

guardrail_response = bedrock_runtime.apply_guardrail( guardrailIdentifier='your-guardrail-id', guardrailVersion='1', source='INPUT', content=[ { 'text': { 'text': user_input # 仅新内容 - 不是完整提示 } } ] )

if guardrail_response['action'] == 'GUARDRAIL_INTERVENED': return { 'blocked': True, 'message': guardrail_response['outputs'][0]['text'], 'assessments': guardrail_response['assessments'] }

步骤2：可以安全继续 - 调用模型时不使用内联防护机制

系统提示和历史记录跳过冗余评估

messages = conversation_history + [{'role': 'user', 'content': user_input}]

response = bedrock_runtime.converse( modelId='anthropic.claude-sonnet-4-20250514', messages=messages, system=[{'text': system_prompt}]

注意：此处没有 guardrailConfig - 我们已经验证了输入

)

return { 'blocked': False, 'response': response['output']['message']['content'][0]['text'] } ``

为什么这对编码工作流很重要：在典型的编码会话中，系统提示（通常包含2000-5000字符的工具定义和指令）和对话历史（随着每次交互增长）会在每次调用时被重新发送。使用内联评估时，这些静态内容会在每次交互中被重新扫描。采用解耦方法后，您只需评估实际更改的约200字符用户消息。

模式：完成时仅输出验证

当需要扫描生成的代码以查找敏感信息（泄露的凭证、硬编码的密钥），但信任用户输入是安全的（例如，受身份验证保护的内部开发工具），只需验证最终输出：

def generate_and_validate_code(user_input: str, system_prompt: str):
    """
    不使用内联防护机制生成代码，然后验证完整输出。
    适用于检测生成代码中的密钥、PII或不安全模式。
    """

    # 步骤1：不使用内联防护机制开销生成代码
    response = bedrock_runtime.converse(
        modelId='anthropic.claude-sonnet-4-20250514',
        messages=[{'role': 'user', 'content': user_input}],
        system=[{'text': system_prompt}]
    )

    generated_code = response['output']['message']['content'][0]['text']

    # 步骤2：验证完整的代码工件 - 不是中间片段
    guardrail_response = bedrock_runtime.apply_guardrail(
        guardrailIdentifier='your-guardrail-id',
        guardrailVersion='1',
        source='OUTPUT',
        content=[
            {
                'text': {
                    'text': generated_code
                }
            }
        ]
    )

    if guardrail_response['action'] == 'GUARDRAIL_INTERVENED':
        return {
            'blocked': True,
            'message': '生成的代码包含被阻止的敏感内容。',
            'assessments': guardrail_response['assessments']
        }

return {
        'blocked': False,
        'code': generated_code
    }

模式：具有选择性作用域的双向验证

为了实现最大覆盖范围，在避免内联扫描开销的同时，应验证输入以防止注入攻击并验证输出以检查敏感内容：

def secure_code_generation_pipeline(user_input: str, system_prompt: str, history: list):
    """
    无内联扫描开销的完整双向验证。
    输入：检查提示注入/操控
    输出：检查泄露的密钥和不安全模式
    """

    # 第1步：验证输入以防止提示注入
    input_check = bedrock_runtime.apply_guardrail(
        guardrailIdentifier='your-guardrail-id',
        guardrailVersion='1',
        source='INPUT',
        content=[{'text': {'text': user_input}}]
    )

    if input_check['action'] == 'GUARDRAIL_INTERVENED':
        return {'blocked': True, 'stage': 'input', 'reason': input_check['outputs'][0]['text']}

    # 第2步：生成代码 - 不需要内联防护栏
    messages = history + [{'role': 'user', 'content': user_input}]
    response = bedrock_runtime.converse(
        modelId='anthropic.claude-sonnet-4-20250514',
        messages=messages,
        system=[{'text': system_prompt}]
    )
    generated_code = response['output']['message']['content'][0]['text']

    # 第3步：验证输出中的敏感信息
    output_check = bedrock_runtime.apply_guardrail(
        guardrailIdentifier='your-guardrail-id',
        guardrailVersion='1',
        source='OUTPUT',
        content=[{'text': {'text': generated_code}}]
    )

    if output_check['action'] == 'GUARDRAIL_INTERVENED':
        return {'blocked': True, 'stage': 'output', 'reason': output_check['outputs'][0]['text']}

    return {'blocked': False, 'code': generated_code}

架构模式4：批量输出对齐文本单元边界

由于600字符的片段仍需消耗一个完整的文本单元（1,000字符），应始终按1,000字符的倍数进行批量处理以避免部分单元浪费：

import asyncio

class GuardrailBatcher:
    """累积流式输出并在1,000字符边界进行评估。"""

    def __init__(self, client, guardrail_id, guardrail_version):
        self.client = client
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version
        self.buffer = ""
        self.BATCH_SIZE = 1000  # 对齐到文本单元边界

    async def process_chunk(self, chunk: str):
        """缓冲片段并在达到文本单元边界时进行评估。"""
        self.buffer += chunk

        while len(self.buffer) >= self.BATCH_SIZE:
            # 提取正好1,000字符用于评估
            batch = self.buffer[:self.BATCH_SIZE]
            self.buffer = self.buffer[self.BATCH_SIZE:]

            response = self.client.apply_guardrail(
                guardrailIdentifier=self.guardrail_id,
                guardrailVersion=self.guardrail_version,
                source='OUTPUT',
                content=[{'text': {'text': batch}}]
            )

            if response['action'] == 'GUARDRAIL_INTERVENED':
                raise GuardrailViolation(response)

async def flush(self): """在流结束时评估任何剩余内容。""" if self.buffer: response = self.client.apply_guardrail( guardrailIdentifier=self.guardrail_id, guardrailVersion=self.guardrail_version, source='OUTPUT', content=[{'text': {'text': self.buffer}}] ) self.buffer = "" return response

## 架构模式 5：基于风险的评估深度

并非所有生成的代码都具有相同的风险。涉及 IAM 策略、密钥或认证的代码需要比 UI 布局代码更彻底的扫描。根据内容信号实现自适应评估深度：

import re

class RiskBasedGuardrail: """ 根据生成代码的风险特征调整防护规则评估策略。

高风险代码：启用所有安全防护的完整评估 标准代码：轻量级评估（仅敏感信息） 低风险代码：跳过中间评估；仅在提交时验证 """

表示高风险代码的模式

HIGH_RISK_PATTERNS = [ r'iam[_\.]', # IAM 策略操作 r'(access_key|secret_key|password)', # 凭据处理 r'(exec|eval|subprocess|os\.system)', # 代码执行 r'(BEGIN.*PRIVATE KEY)', # 加密密钥 r'(security_group|nacl|firewall)', # 网络安全 r'(grant|revoke|permission)', # 授权操作 r'(encrypt|decrypt|kms)', # 加密操作 ]

表示低风险代码的模式

LOW_RISK_PATTERNS = [ r'(\.css|style|className|fontSize)', # 样式设置 r'(render|component|jsx|tsx)', # UI 组件 r'(test|spec|mock|fixture)', # 测试文件 r'(README|docs|comment)', # 文档内容 ]

def __init__(self, client, guardrail_id_full, guardrail_id_lightweight): self.client = client self.guardrail_id_full = guardrail_id_full # 启用所有安全防护 self.guardrail_id_lightweight = guardrail_id_lightweight # 仅检测敏感信息

def classify_risk(self, code: str, file_path: str = "") -> str: """确定生成代码的风险等级。""" combined = code + " " + file_path

for pattern in self.HIGH_RISK_PATTERNS: if re.search(pattern, combined, re.IGNORECASE): return 'HIGH'

for pattern in self.LOW_RISK_PATTERNS: if re.search(pattern, combined, re.IGNORECASE): return 'LOW'

return 'STANDARD'

def evaluate(self, code: str, file_path: str = "") -> dict: """ 根据风险分类以适当深度评估代码。 """ risk_level = self.classify_risk(code, file_path)

if risk_level == 'LOW':

低风险：延迟到提交时验证

return { 'action': 'PASS', 'risk_level': 'LOW', 'evaluation': 'deferred_to_commit', 'reason': '内容被归类为低风险（UI/测试/文档）' }

根据风险选择防护规则

guardrail_id = ( self.guardrail_id_full if risk_level == 'HIGH' else self.guardrail_id_lightweight )

response = self.client.apply_guardrail( guardrailIdentifier=guardrail_id, guardrailVersion='1', source='OUTPUT', content=[{'text': {'text': code}}] )

return { 'action': response['action'], 'risk_level': risk_level, 'evaluation': 'full' if risk_level == 'HIGH' else 'lightweight', 'assessments': response.get('assessments', []) }

示例：在编码流水线中使用基于风险的评估

risk_guardrail = RiskBasedGuardrail( client=bedrock_runtime, guardrail_id_full='guardrail-all-safeguards', guardrail_id_lightweight='guardrail-sensitive-info-only' )

高风险：IAM 策略代码 → 全面评估（3 个防护措施）

警告：以下是一个不安全的反模式示例，仅用于演示防护措施检测

请勿在生产环境中使用通配符权限

(Action: '*', Resource: '*')。始终遵循最小权限原则

详情请参阅：https://docs.aws.amazon.com/IAM/latest/UserGuide/best-practices.html#grant-least-privilege

iam_code = """ policy = iam.create_policy( PolicyName='AdminAccess', PolicyDocument=json.dumps({ 'Statement': [{'Effect': 'Allow', 'Action': '*', 'Resource': '*'}] }) ) """ result = risk_guardrail.evaluate(iam_code, "deploy/iam_policies.py")

→ risk_level: HIGH, evaluation: full

低风险：React 组件 → 延迟到提交时评估

ui_code = """ function Button({ label, onClick }) { return <button className="btn-primary" onClick={onClick}>{label}</button>; } """ result = risk_guardrail.evaluate(ui_code, "src/components/Button.tsx")

→ risk_level: LOW, evaluation: deferred_to_commit

## 架构模式 6：多阶段代理流水线

代理式编码工作流（模型使用工具、跨多步推理并生成中间输出）需要不同于单次生成的评估策略：

class AgentPipelineGuardrail: """ 多步骤代理式编码工作流的防护措施架构。

关键洞察：在代理循环中，模型可能需要进行 5-10 次推理步骤 （工具调用、观察、反思）才能生成最终代码。 评估每个中间步骤效率低下 - 大多数包含内部推理 且不会到达用户或文件系统。

策略：

• INPUT: 在代理开始之前验证用户请求
• INTERMEDIATE: 跳过推理链和工具使用
• TOOL INPUTS: 在代理写入文件或执行代码时验证
• FINAL OUTPUT: 验证展示给用户的完整响应

"""

def __init__(self, client, guardrail_id, guardrail_version): self.client = client self.guardrail_id = guardrail_id self.guardrail_version = guardrail_version

def run_agent_with_guardrails(self, user_request: str, agent): """ 在战略性防护措施位置执行代理式编码工作流。 """

阶段 1：在代理开始工作前验证用户输入

input_check = self._evaluate(user_request, 'INPUT') if input_check['action'] == 'GUARDRAIL_INTERVENED': return {'blocked': True, 'stage': 'input', 'response': input_check}

运行代理 - 推理步骤期间不进行防护措施评估

agent_steps = [] final_output = None

for step in agent.run(user_request):
    if step['type'] == 'thinking':
        # 跳过：内部推理不需要评估
        agent_steps.append(step)
        continue

    elif step['type'] == 'tool_call':
        # 防御点2：仅验证危险工具的输入
        if self._is_dangerous_tool(step['tool_name']):
            tool_check = self._evaluate(
                json.dumps(step['tool_input']), 'OUTPUT'
            )
            if tool_check['action'] == 'GUARDRAIL_INTERVENED':
                return {
                    'blocked': True,
                    'stage': f"tool_call:{step['tool_name']}",
                    'response': tool_check
                }
        agent_steps.append(step)

    elif step['type'] == 'final_response':
        final_output = step['content']

# 防御点3：在向用户展示前验证最终输出
if final_output:
    output_check = self._evaluate(final_output, 'OUTPUT')
    if output_check['action'] == 'GUARDRAIL_INTERVENED':
        return {'blocked': True, 'stage': 'output', 'response': output_check}

return {
    'blocked': False,
    'output': final_output,
    'steps_taken': len(agent_steps),
    'steps_evaluated': sum(
        1 for s in agent_steps
        if s['type'] == 'tool_call' and self._is_dangerous_tool(s.get('tool_name', ''))
    )
}

def _is_dangerous_tool(self, tool_name: str) -> bool:
    """需要防护栏评估的工具是那些会写入磁盘或执行代码的工具。"""
    dangerous_tools = {
        'write_file', 'execute_code', 'run_command',
        'create_file', 'edit_file', 'bash'
    }
    return tool_name in dangerous_tools

def _evaluate(self, content: str, source: str) -> dict:
    return self.client.apply_guardrail(
        guardrailIdentifier=self.guardrail_id,
        guardrailVersion=self.guardrail_version,
        source=source,
        content=[{'text': {'text': content}}]
    )

完整的决策框架

下表展示了各种检查点的决策框架：

| 检查点 | 需要验证的内容 | 验证方式 | 文本单元影响 | |---|---|---|---| | 用户输入（每次交互） | 仅动态用户内容 | ApplyGuardrail（source=INPUT） | 低 – 仅评估新内容 | | 流式输出 | 流式传输时的活动内容 | 设置间隔为1,000字符 | 相比默认值最多减少20倍评估频率 | | 完成的响应 | 最终聚合的代码制品 | ApplyGuardrail（source=OUTPUT） | 一次性全面检查 | | 提交前/文件保存 | AI生成的代码更改 | 全面但频率较低 | | | 代理工具调用 | 仅危险工具（写入/执行） | 针对性 – 跳过无害工具 | | | 中间推理 | 完全跳过 | 不评估CoT标记 | 0 |

关键要点

• 从内联扫描转向选择性评估 – 使用解耦的ApplyGuardrail API来精确控制扫描内容和时机。在信任边界进行评估，而非持续评估。

• 将流式传输间隔设置为1,000字符 – 当需要流式评估时，这一更改可将评估频率减少多达20倍。

• 仅评估动态内容 – 不要每次交互都重新评估系统提示、工具定义和对话历史。对需要重新评估的内容使用基于哈希的缓存机制。

• 采用基于风险的评估深度 – 使用安全措施扫描 IAM 策略和凭证处理代码。将 UI 组件扫描推迟到提交时进行。

• 跳过中间推理标记 – 在智能体工作流中，仅在内容跨越信任边界时（用户输入、危险工具调用、最终输出）进行评估。

• 按 1,000 字符边界批量处理 – 600 字符片段的费用与 1,000 字符相同。将评估对齐到文本单元边界以减少浪费。

• 在提交边界设置防护 – 像 Git 提交前钩子一样，当 AI 生成的代码即将变为持久化可执行代码时执行全面验证。

结论

在本文中，我们提出了使用 Amazon Bedrock Guardrails 优化 AI 协助代码生成工作流的最佳实践。以下是您如何有效为代码工作流实施 Guardrails 的总体总结：

• 审核当前配置：打开 Amazon Bedrock 控制台，查看现有的 Guardrails。检查您的流式传输间隔设置。如果未从默认的 50 字符更改设置，您可能错失了潜在的 20 倍效率提升。

• 检查账户配额：在扩展代码工作流之前，请通过 AWS 服务配额控制台验证您实际分配的限制。不要假设发布的默认值适用，尤其是新账户。请参阅 Amazon Bedrock 终端节点和配额文档以获取最新的区域限制。

• 实施解耦的 ApplyGuardrail API：从内联评估切换为使用 ApplyGuardrail API 进行选择性验证。查阅配置流式传输响应行为文档以微调您的流式传输间隔。

• 探索完整的 Guardrails 功能集：深入阅读 Amazon Bedrock Guardrails 文档，了解内容过滤器、禁止主题、敏感信息检测以及如何为您的使用案例创建和修改 Guardrails。

作者简介

'"` /think