实时智能体的护栏

Supported in ADKPython v2.8.0

护栏决定了实时语音智能体在生产环境中的行为方式，让对话保持在主题范围内、符合政策 并且安全。在实时连接中，音频持续流动，模型开始实时说话。ADK 提供了多层协同工作的 保护机制，在对话延迟影响有限的前提下保障对话安全。

关于通用回调机制和插件注册，参见 回调类型和插件。

护栏层级

层级	作用	防护重点	延迟影响
系统指令	塑造语气、对话规则和边界	行为指引	不增加应用延迟
安全设置	强制执行内容安全的平台阈值	平台安全过滤器	不增加应用延迟
输入验证	捕获提示词注入和被禁止的话题	用户输入验证	极小（每轮执行一次）
响应验证	按业务策略筛查智能体的响应	智能体响应验证	取决于具体检查
工具护栏	拦截工具执行并校验参数	工具执行安全	极小（每次工具调用执行一次）

这些层级协同工作，提供纵深防御。系统指令塑造自然的对话流程，平台安全过滤器强制执行 基本的伤害边界，应用层验证器独立校验输入、响应和工具调用。将清晰的指令与独立的验证 相结合，可以确保即使某个边界情况绕过了一层，另一层也能捕获它。

指令与安全设置

指令与安全设置为整个会话确立基线行为。这两项设置都属于智能体，你在创建会话时进行 配置：

from google.adk.agents import Agent
from google.genai import types

root_agent = Agent(
    model='gemini-live-2.5-flash-native-audio',
    name='support_agent',
    instruction=(
        'You help customers with billing questions. '
        'Never quote a price; offer to transfer to sales instead. '
        'Never discuss a competitor.'
    ),
    generate_content_config=types.GenerateContentConfig(
        safety_settings=[
            types.SafetySetting(
                category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
            ),
        ]
    ),
)

随着对话上下文不断累积，模型会在每一轮中评估系统指令。为实时智能体设计指令时， 应按顺序定义角色设定、对话规则和对话护栏。对于严格约束，要给出触发边界时的明确 处理规则，例如提供替代方案或转接呼叫方。

安全设置将安全和内容过滤器 应用于实时会话。当模型生成的内容超过配置的阈值时，平台会终止生成。

实时连接在启动时建立这些设置，因此它们在会话期间保持固定。插值了会话状态的指令字符串 在会话启动时渲染一次，恢复的连接会复用该已渲染的指令。

对话验证

对话验证独立评估用户说了什么以及智能体如何回应。对于托管的企业策略， Model Armor 插件会依据 Google Cloud 模板筛查提示词注入、 敏感数据和政策违规。你也可以使用 ADK 模型回调实现自定义验证。

用户输入验证

before_model_callback 钩子拦截用户输入。该回调在文本到达模型之前对其进行评估， 因此阻止操作可以防止消息进入上下文，同时保持连接不断开。对于语音音频，回调是在模型 已经接收到音频之后，才把转写后的语句作为一个整体进行检查；此时阻止会重启实时会话， 以丢弃已经在处理中的回复。

当检查发现政策违规时，返回一个 LlmResponse 会用安全的回退响应替换用户的这一轮：

from typing import Optional

from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.genai import types


def block_input(
    callback_context: CallbackContext,
    llm_request: LlmRequest,
) -> Optional[LlmResponse]:
  """阻止提到竞争对手的用户消息。"""
  text = ''.join(
      part.text or ''
      for content in llm_request.contents
      for part in content.parts or []
  )

  if 'competitor' not in text.lower():
    return None

  return LlmResponse(
      content=types.Content(
          role='model',
          parts=[types.Part(text="I can't discuss that.")],
      )
  )

智能体响应验证

after_model_callback 钩子在模型响应交付给用户之前或交付过程中对其进行校验。 验证可以在语音轮次进行中随着转写内容的累积而检查以便及早发现违规，也可以评估 整轮的完整转写。

def block_output(
    callback_context: CallbackContext,
    llm_response: LlmResponse,
) -> Optional[LlmResponse]:
  """当回答中出现报价时替换该响应。"""
  transcription = llm_response.output_transcription
  text = transcription.text if transcription else ''

  if '$' not in text:
    return None

  return LlmResponse(
      content=types.Content(
          role='model',
          parts=[types.Part(text='Let me connect you with sales.')],
      )
  )

从输出回调返回 LlmResponse 会立即停止模型生成，并把你的替换消息交付给用户。 对于模型已经开始处理输入的语音轮次，ADK 会清除当前轮次，使被拒绝的内容不会 留在对话上下文中。

工具护栏

工具回调用于保护外部系统并校验操作。before_tool_callback 钩子在工具执行之前检查 参数，让你可以拒绝未授权的参数或强制执行业务逻辑。after_tool_callback 钩子在敏感 结果返回给模型之前对其进行脱敏：

from typing import Any, Optional

from google.adk.tools import BaseTool
from google.adk.tools import ToolContext


def validate_refund(
    tool: BaseTool,
    args: dict[str, Any],
    tool_context: ToolContext,
) -> Optional[dict[str, Any]]:
  """阻止超过阈值的未授权退款。"""
  if tool.name == 'issue_refund' and args.get('amount', 0) > 100:
    return {'error': 'Refund exceeds automatic approval limit.'}
  return None

从工具回调返回结果会绕过工具执行，并把返回值直接提供给模型。参见 工具执行回调。

实时音频流无法暂停来等待交互式人工确认。当实时智能体使用需要更高权限的工具时， 应在回调内自动校验参数，或将请求路由到人工交接。

要求与最佳实践

• 确保转写保持开启：基于文本的对话验证依赖语音转写。 RunConfig.input_audio_transcription 和 RunConfig.output_audio_transcription 默认均已启用。将其中任一项设为 None 会禁用对应的筛查层。
• 保持验证器轻量：实时接收循环中的回调与音频处理内联运行。快速的本地检查能让 对话保持流畅。对于较重的检查，应评估完整的轮次转写，或把耗时的分析异步卸载。
• 跨智能体应用策略：在 App 上注册 Model Armor 等插件，可以在不重复编写回调代码的 情况下，为应用中的每个智能体应用一致的安全规则。

其他资源

• Model Armor 插件
• 回调类型
• 插件
• 配置安全过滤器