AI 智能体的安全与保障

在 ADK 中支持PythonTypeScriptGoJava

随着 AI 智能体能力的增强，确保它们安全、可靠地运行并与你的品牌价值观保持一致至关重要。未受控制的智能体可能带来风险，包括执行不符合预期或有害的操作（如数据泄露），以及生成可能损害你品牌声誉的不当内容。风险来源包括模糊的指令、模型幻觉、来自恶意用户的越狱和提示词注入，以及通过工具调用产生的间接提示词注入。

[Google Cloud Agent Platform 提供了一种多层方法来解决这些风险，使你能够构建强大且值得信赖的智能体。它提供了几种机制来建立严格的边界，确保智能体仅执行你明确允许的操作：

1. 身份和授权 (Identity and Authorization)：通过定义智能体和用户身份验证来控制智能体以谁的身份行事。
2. 用于筛选输入和输出的护栏 (Guardrails to screen inputs and outputs)：精确控制你的模型和工具调用。
   • 工具内护栏 (In-Tool Guardrails): 防御性地设计工具，使用开发者设置的工具上下文来强制执行策略（例如，仅允许查询特定表格）。
   • 内置 Gemini 安全功能 (Built-in Gemini Safety Features): 如果使用 Gemini 模型，可受益于内容过滤器以阻止有害输出，以及系统指令以指导模型的行为和安全准则。
   • 回调和插件 (Callbacks and Plugins): 在执行之前或之后验证模型和工具调用，根据智能体状态或外部策略检查参数。
   • 使用 Gemini 作为安全护栏 (Using Gemini as a safety guardrail): 使用通过回调配置的廉价快速模型（如 Gemini Flash Lite）实现额外的安全层，以筛选输入和输出。
3. 沙盒化代码执行 (Sandboxed code execution): 通过沙盒化环境防止模型生成的代码导致安全问题。
4. 评估和追踪 (Evaluation and tracing): 使用评估工具来评估智能体最终输出的质量、相关性和正确性。使用追踪来深入了解智能体操作，分析智能体为达成解决方案所采取的步骤，包括其工具选择、策略和方法的效率。
5. 网络控制和 VPC-SC (Network Controls and VPC-SC): 将智能体活动限制在安全的边界内（如 VPC Service Controls），以防止数据泄露并限制潜在的影响范围。

安全与保障风险

在实施安全措施之前，请针对你的智能体的能力、领域和部署环境进行全面的风险评估。

风险的来源包括：

• 模糊的智能体指令 (Ambiguous agent instructions)
• 来自恶意用户的提示词注入和越狱尝试 (Prompt injection and jailbreak attempts from adversarial users)
• 通过工具使用产生的间接提示词注入 (Indirect prompt injections via tool use)

风险类别包括：

• 错位与目标腐朽 (Misalignment & goal corruption)
  • 追求非预期或代理目标，导致有害结果（“奖励黑客”）
  • 误解复杂或模糊的指令
• 有害内容生成，包括品牌安全 (Harmful content generation, including brand safety)
  • 生成有毒、仇恨、偏见、色情、歧视性或非法内容
  • 品牌安全风险，例如使用与品牌价值观相悖的语言或离题对话
• 不安全操作 (Unsafe actions)
  • 执行损坏系统的命令
  • 进行未经授权的购买或金融交易
  • 泄露敏感个人数据 (PII)
  • 数据外泄 (Data exfiltration)

最佳实践

身份和授权

从安全角度看，工具用于在外部系统上执行操作的身份是一个至关重要的设计考量。同一智能体中的不同工具可以配置不同的策略，因此在讨论智能体配置时需要谨慎。

智能体授权

工具使用智能体自己的身份（例如服务账号）与外部系统交互。必须在外部系统访问策略中明确授权智能体身份，例如将智能体的服务账号添加到数据库的 IAM 策略中以获取读取权限。这些策略约束智能体只执行开发者允许的操作：通过为资源提供只读权限，无论模型决定做什么，工具都将被禁止进行写入操作。

这种方法实现起来很简单，并且适用于所有用户共享相同访问级别的智能体。如果并非所有用户都具有相同的访问级别，那么单纯这种方法并不能提供足够的保护，必须与下文的其他技术相结合。在工具实现中，请确保创建日志以维护用户操作的归属关系，因为所有智能体的操作都将显示为来自智能体本身。

用户授权

工具使用“控制用户”的身份（例如在 Web 应用程序中与前端交互的人类）与外部系统交互。在 ADK 中，这通常通过 OAuth 实现：智能体与前端交互以获取 OAuth 令牌，然后工具在执行外部操作时使用该令牌。如果控制用户被授权自行执行操作，则外部系统会授权该操作。

用户授权的优势在于智能体仅执行用户本人可以执行的操作。这大大降低了恶意用户滥用智能体获取额外数据访问权限的风险。然而，大多数常见的委托实现都有一组固定的权限委托（即 OAuth 范围）。通常，这些范围比智能体实际需要的访问权限更广泛，因此需要下文的技术来进一步约束智能体的操作。

---

用于筛选输入和输出的护栏

工具内护栏

可以在设计工具时考虑安全性：我们可以创建仅公开我们希望模型采取的操作而不公开其他操作的工具。通过限制我们提供给智能体的操作范围，我们可以确定性地消除我们永远不希望智能体采取的一类恶意操作。

这种方法依赖于这样一个事实：工具接收两种类型的输入：由模型设置的参数，以及可以由智能体开发者确定性设置的 工具上下文。我们可以依靠确定性设置的信息来验证模型是否按预期运行。（注：在 TypeScript 中，Tool Context 对应于统一的 Context 类型。）

例如，查询工具可以设计为期望从工具上下文中读取策略。

PythonTypeScriptGoJava

# 概念示例：设置用于工具上下文的策略数据
# 在真实的 ADK 应用程序中，这可能在 InvocationContext.session.state 中设置
# 或在工具初始化期间传递，然后通过 ToolContext 检索。

policy = {} # 假设策略是一个字典
policy['select_only'] = True
policy['tables'] = ['mytable1', 'mytable2']

# 概念：存储策略，以便工具稍后可以通过 ToolContext 访问它。
# 实际应用中，这行代码可能有所不同。
# 例如，存储在会话状态中：
invocation_context.session.state["query_tool_policy"] = policy

# 或者在工具初始化期间传递：
query_tool = QueryTool(policy=policy)
# 对于本例，我们假设它被存储在某个可访问的位置。

// 概念示例：设置用于工具上下文的策略数据
// 在真实的 ADK 应用程序中，这可能在 InvocationContext.session.state 中设置
// 或在工具初始化期间传递，然后通过 Context 检索。

const policy: {[key: string]: any} = {}; // 假设策略是一个对象
policy['select_only'] = true;
policy['tables'] = ['mytable1', 'mytable2'];

// 概念：存储策略，以便工具以后可以通过 Context 访问它。
// 在实际操作中，这一行可能会有所不同。
// 例如，存储在会话状态中：
invocationContext.session.state["query_tool_policy"] = policy;

// 或者在工具初始化期间传递：
const queryTool = new QueryTool({policy: policy});
// 对于本例，我们假设它被存储在某个可访问的位置。

// 概念示例：设置用于工具上下文的策略数据
// 在真实的 ADK 应用程序中，这可能使用会话状态服务进行设置。
// `ctx` 是回调或自定义智能体中可用的 `agent.Context`。

policy := map[string]any{
    "select_only": true,
    "tables":      []string{"mytable1", "mytable2"},
}

// 概念：存储策略，以便工具以后可以通过 ToolContext 访问它。
// 这行代码在实际中可能有所不同。
// 例如，存储在会话状态中：
if err := ctx.Session().State().Set("query_tool_policy", policy); err != nil {
    // 处理错误，例如记录它。
}

// 或者在工具初始化期间传递：
// queryTool := NewQueryTool(policy)
// 对于本例，我们假设它被存储在某个可访问的位置。

// 概念示例：设置用于工具上下文的策略数据
// 在真实的 ADK 应用程序中，这可能在 InvocationContext.session.state 中设置
// 或在工具初始化期间传递，然后通过 ToolContext 检索。

policy = new HashMap<String, Object>(); // 假设策略是一个 Map
policy.put("select_only", true);
policy.put("tables", new ArrayList<>("mytable1", "mytable2"));

// 概念：存储策略，以便工具以后可以通过 ToolContext 访问它。
// 实际上，这行代码可能会有所不同。
// 例如，存储在会话状态中：
invocationContext.session().state().put("query_tool_policy", policy);

// 或者在工具初始化期间传递：
query_tool = QueryTool(policy);
// 对于本例，我们假设它被存储在某个可访问的位置。

在工具执行期间，工具上下文 将被传递给工具（注：在 TypeScript 中，这作为统一的 Context 类型传递）：

PythonTypeScriptGoJava

def query(query: str, tool_context: ToolContext) -> str | dict:
  # 假设 'policy' 是从上下文中检索的，例如通过会话状态：
  # policy = tool_context.invocation_context.session.state.get('query_tool_policy', {})

  # --- 占位符策略强制执行 ---
  policy = tool_context.invocation_context.session.state.get('query_tool_policy', {}) # 示例检索
  actual_tables = explainQuery(query) # 假设的函数动作

  if not set(actual_tables).issubset(set(policy.get('tables', []))):
    # 返回一个错误信息给模型
    allowed = ", ".join(policy.get('tables', ['(未定义)']))
    return f"Error: Query targets unauthorized tables. Allowed: {allowed}"

  if policy.get('select_only', False):
       if not query.strip().upper().startswith("SELECT"):
           return "Error: Policy restricts queries to SELECT statements only."
  # --- 占位符策略强制执行结束 ---

  print(f"Executing validated query (hypothetical): {query}")
  return {"status": "success", "results": [...]} # 示例成功返回

function query(query: string, context: Context): string | object {
    // 假设 'policy' 是从上下文中检索的，例如通过会话状态：
    const policy = context.state.get('query_tool_policy', {}) as {[key: string]: any};

    // --- 占位符策略强制执行 ---
    const actual_tables = explainQuery(query); // 假设的函数动作

    const policyTables = new Set(policy['tables'] || []);
    const isSubset = actual_tables.every(table => policyTables.has(table));

    if (!isSubset) {
        // 为模型返回错误信息
        const allowed = (policy['tables'] || ['(未定义)']).join(', ');
        return `Error: Query targets unauthorized tables. Allowed: ${allowed}`;
    }

    if (policy['select_only']) {
        if (!query.trim().toUpperCase().startsWith("SELECT")) {
            return "Error: Policy restricts queries to SELECT statements only.";
        }
    }
    // --- 策略强制执行结束 ---

    console.log(`执行已验证的查询（假设）：${query}`);
    return { "status": "success", "results": [] }; // 示例成功返回
}

import (
    "fmt"
    "strings"

    "google.golang.org/adk/tool"
)

func query(ctx tool.Context, args QueryArgs) (map[string]any, error) {
    // 假设 'policy' 是从上下文中检索的，例如通过会话状态：
    policyAny, err := ctx.Session().State().Get("query_tool_policy")
    if err != nil {
        return nil, fmt.Errorf("could not retrieve policy: %w", err)
    }
    policy, _ := policyAny.(map[string]any)
    actualTables := explainQuery(args.Query) // 假设的函数动作

    // --- 占位符策略强制执行 ---
    if tables, ok := policy["tables"].([]string); ok {
        if !isSubset(actualTables, tables) {
            // 返回错误以表示失败
            allowed := strings.Join(tables, ", ")
            if allowed == "" {
                allowed = "(未定义)"
            }
            return nil, fmt.Errorf("查询目标未经授权的表。允许的表：%s", allowed)
        }
    }

    if selectOnly, _ := policy["select_only"].(bool); selectOnly {
        if !strings.HasPrefix(strings.ToUpper(strings.TrimSpace(args.Query)), "SELECT") {
            return nil, fmt.Errorf("策略限制查询仅允许 SELECT 语句")
        }
    }
    // --- 占位符策略强制执行结束 ---

    fmt.Printf("执行已验证的查询（假设）：%s\n", args.Query)
    return map[string]any{"status": "success", "results": []string{"..."}}, nil
}

// 辅助函数，检查 a 是否是 b 的子集
func isSubset(a, b []string) bool {
    set := make(map[string]bool)
    for _, item := range b {
        set[item] = true
    }
    for _, item := range a {
        if _, found := set[item]; !found {
            return false
        }
    }
    return true
}

import com.google.adk.tools.ToolContext;
import java.util.*;

class ToolContextQuery {

  public Object query(String query, ToolContext toolContext) {

    // 假设 'policy' 是从上下文中检索的，例如通过会话状态：
    Map<String, Object> queryToolPolicy =
        toolContext.invocationContext.session().state().getOrDefault("query_tool_policy", null);
    List<String> actualTables = explainQuery(query);

    // --- 占位符策略强制执行 ---
    if (!((List<String>) queryToolPolicy.get("tables")).containsAll(actualTables)) {
      List<String> allowedPolicyTables =
          (List<String>) queryToolPolicy.getOrDefault("tables", new ArrayList<String>());

      String allowedTablesString =
          allowedPolicyTables.isEmpty() ? "(未定义)" : String.join(", ", allowedPolicyTables);

      return String.format(
          "Error: Query targets unauthorized tables. Allowed: %s", allowedTablesString);
    }

    if ((Boolean) queryToolPolicy.get("select_only")) {
      if (!query.trim().toUpperCase().startsWith("SELECT")) {
        return "Error: Policy restricts queries to SELECT statements only.";
      }
    }
    // --- 策略强制执行结束 ---

    System.out.printf("执行已验证的查询（假设）%s:", query);
    Map<String, Object> successResult = new HashMap<>();
    successResult.put("status", "success");
    successResult.put("results", Arrays.asList("result_item1", "result_item2"));
    return successResult;
  }
}

内置 Gemini 安全功能

Gemini 模型带有内置的安全机制，可用于提高内容和品牌安全。

• 内容安全过滤器：内容过滤器 可以帮助阻止有害内容的输出。它们独立于 Gemini 模型运行，作为针对试图越狱模型的威胁行为者的分层防御的一部分。Agent Platform 上的 Gemini 模型使用两种类型的内容过滤器：
  • 不可配置的安全过滤器 会自动阻止包含禁止内容的输出，例如儿童性虐待材料 (CSAM) 和个人身份信息 (PII)。
  • 可配置的内容过滤器 允许你在四个有害类别（仇恨言论、骚扰、色情内容和危险内容）中基于概率和严重性分数定义阻止阈值。这些过滤器默认关闭，但你可以根据需要进行配置。

PythonGo

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

agent = Agent(
    # ...
    generate_content_config=types.GenerateContentConfig(
        safety_settings=[
            types.SafetySetting(
                category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=types.HarmBlockThreshold.OFF,
            ),
        ],
    ),
)

import (
    "google.golang.org/adk/agent/llmagent"
    "google.golang.org/genai"
)

agent, _ := llmagent.New(llmagent.Config{
    // ...
    GenerateContentConfig: &genai.GenerateContentConfig{
        SafetySettings: []*genai.SafetySetting{
            {
                Category:  genai.HarmCategoryHateSpeech,
                Threshold: genai.HarmBlockThresholdBlockLowAndAbove,
            },
        },
    },
})

• 用于安全性的系统指令：系统指令 for Gemini models on Agent Platform provide direct guidance to the model on how to behave and what type of content to generate. By providing specific instructions, you can proactively steer the model away from generating undesirable content to meet your organization’s unique needs. You can craft system instructions to define content safety guidelines, such as prohibited and sensitive topics, and disclaimer language, as well as brand safety guidelines to ensure the model's outputs align with your brand's voice, tone, values, and target audience.

虽然这些措施对内容安全非常强大，但你仍需要额外的检查来减少智能体的不一致、不安全行为和品牌安全风险。

安全护栏的回调和插件

回调提供了一种简单的、特定于智能体的方法来为工具和模型 I/O 添加预验证，而插件提供了一种可重用的解决方案，可以在多个智能体中实现通用安全策略。

当无法修改工具以添加护栏时，可以使用 工具执行前回调 (Before Tool Callback) 函数为调用添加预验证。该回调可以访问智能体的状态、请求的工具及其参数。这种方法非常通用，甚至可以用来创建可重用的工具策略通用库。但如果要强制执行护栏所需的信息未直接体现在参数中，则此方法可能不适用于所有工具。

PythonTypeScriptGoJava

# 假设的回调函数
def validate_tool_params(
    callback_context: CallbackContext, # 正确的上下文类型
    tool: BaseTool,
    args: Dict[str, Any],
    tool_context: ToolContext
    ) -> Optional[Dict]: # before_tool_callback 的正确返回类型

  print(f"Callback triggered for tool: {tool.name}, args: {args}")

  # 示例验证：检查状态中所需的 user ID 是否与参数匹配
  expected_user_id = callback_context.state.get("session_user_id")
  actual_user_id_in_args = args.get("user_id_param") # 假设工具接受 'user_id_param'

  if actual_user_id_in_args != expected_user_id:
      print("Validation Failed: User ID mismatch!")
      # 返回字典以阻止工具执行并提供反馈
      return {"error": f"Tool call blocked: User ID mismatch."}

  # 如果验证通过，返回 None 以允许工具调用继续
  print("Callback validation passed.")
  return None

# Hypothetical Agent setup
root_agent = LlmAgent( # Use specific agent type
    model='gemini-flash-latest',
    name='root_agent',
    instruction="...",
    before_tool_callback=validate_tool_params, # 分配回调
    tools = [
      # ... 工具函数或 Tool 实例列表 ...
    ]
)

// 假设的回调函数
function validateToolParams(
    {tool, args, context}: {
        tool: BaseTool,
        args: {[key: string]: any},
        context: Context
    }
): {[key: string]: any} | undefined {
    console.log(`针对工具 ${tool.name} 触发的回调，参数：${JSON.stringify(args)}`);

    // 示例验证：检查状态中所需的用户 ID 是否与参数匹配
    const expectedUserId = context.state.get("session_user_id");
    const actualUserIdInArgs = args["user_id_param"]; // 假设工具接受 'user_id_param'

    if (actualUserIdInArgs !== expectedUserId) {
        console.log("验证失败：用户 ID 不匹配！");
        // 返回对象以阻止工具执行并提供反馈
        return {"error": `工具调用被阻止：用户 ID 不匹配。`};
    }

    // 如果验证通过，返回 undefined 以允许工具调用继续
    console.log("回调验证通过。");
    return undefined;
}

// 假设的智能体设置
const rootAgent = new LlmAgent({
    model: 'gemini-flash-latest',
    name: 'root_agent',
    instruction: "...",
    beforeToolCallback: validateToolParams, // 分配回调
    tools: [
      // ... 工具函数或工具实例列表 ...
    ]
});

import (
    "fmt"

    "google.golang.org/adk/agent/llmagent"
    "google.golang.org/adk/tool"
)

// 假设的回调函数
func validateToolParams(
    ctx tool.Context,
    t tool.Tool,
    args map[string]any,
) (map[string]any, error) {
    fmt.Printf("Callback triggered for tool: %s, args: %v\n", t.Name(), args)

    // 示例验证：检查状态中所需的 user ID 是否与参数匹配
    expectedUserIDVal, err := ctx.Session().State().Get("session_user_id")
    if err != nil {
        // 返回 map 以阻止工具执行并提供反馈给模型
        return map[string]any{"error": "Tool call blocked: User ID not found."}, nil
    }
    expectedUserID, _ := expectedUserIDVal.(string)

    actualUserID, ok := args["user_id_param"].(string)
    if !ok || actualUserID != expectedUserID {
        fmt.Println("Validation Failed: User ID mismatch!")
        return map[string]any{"error": "Tool call blocked: User ID mismatch."}, nil
    }

    // 如果验证通过，返回 nil, nil 以允许工具调用继续
    fmt.Println("Callback validation passed.")
    return nil, nil
}

// Hypothetical Agent setup
// agent, _ := llmagent.New(llmagent.Config{
//  Model: "gemini-flash-latest",
//  Name: "root_agent",
//  Instruction: "...",
//  BeforeToolCallbacks: []llmagent.BeforeToolCallback{validateToolParams},
//  Tools: []tool.Tool{queryToolInstance},

// 假设的回调函数
public Optional<Map<String, Object>> validateToolParams(
  CallbackContext callbackContext,
  Tool baseTool,
  Map<String, Object> input,
  ToolContext toolContext) {

System.out.printf("Callback triggered for tool: %s, Args: %s", baseTool.name(), input);

// 示例验证：检查状态中所需的 user ID 是否与参数匹配
Object expectedUserId = callbackContext.state().get("session_user_id");
Object actualUserIdInput = input.get("user_id_param"); // 假设工具接受 'user_id_param'

if (!actualUserIdInput.equals(expectedUserId)) {
  System.out.println("Validation Failed: User ID mismatch!");
  // 返回以防止工具执行并提供反馈
  return Optional.of(Map.of("error", "Tool call blocked: User ID mismatch."));
}

// 如果验证通过，返回以允许工具调用继续
System.out.println("Callback validation passed.");
return Optional.empty();
}

// Hypothetical Agent setup
public void runAgent() {
LlmAgent agent =
    LlmAgent.builder()
        .model("gemini-flash-latest")
        .name("AgentWithBeforeToolCallback")
        .instruction("...")
        .beforeToolCallback(this::validateToolParams) // Assign the callback
        .tools(anyToolToUse) // Define the tool to be used
        .build();
}

然而，在为你的智能体应用程序添加安全护栏时，插件是实施非特定于单个智能体策略的推荐方法。插件设计为自包含且模块化的，允许你为特定的安全策略创建单独的插件，并在 Runner 级别全局应用它们。这意味着可以配置一次安全插件并将其应用到使用该 Runner 的每个智能体，确保整个应用程序的安全护栏保持一致，而无需重复代码。

一些示例包括：

• Gemini 作为监督插件：此插件使用 Gemini Flash Lite 评估用户输入、工具输入和输出以及智能体的响应是否恰当，并进行提示词注入和越狱检测。该插件将 Gemini 配置为安全过滤器，以缓解内容安全、品牌安全和智能体不一致问题。该插件被配置为将用户输入、工具输入和输出以及模型输出传递给 Gemini Flash Lite，由其决定智能体的输入是否安全。如果 Gemini 认为输入不安全，智能体会返回预定的响应：“抱歉，我无法帮助处理这个问题。我可以帮助你处理其他事情吗？”。
• Model Armor 插件：一个查询 Model Armor API 的插件，用于在智能体执行的指定点检查潜在的内容安全违规。类似于 Gemini 作为监督 插件，如果 Model Armor 发现有害内容匹配，它会向用户返回预定的响应。
• PII 脱敏插件 (PII Redaction Plugin)：一个专门为 工具执行前回调 (Before Tool Callback) 设计的插件，旨在在工具处理或发送到外部服务之前脱敏个人身份信息。

---

沙盒化代码执行

代码执行是一个具有额外安全含义的特殊工具：必须使用沙盒化来防止模型生成的代码危害本地环境，以免造成安全问题。

Google 和 ADK 为安全的代码执行提供了多种选择。Vertex Gemini Enterprise API 代码执行功能 使智能体能够通过启用 tool_execution 工具来利用服务器端的沙盒化代码执行。对于执行数据分析的代码，你可以在 ADK 中使用 代码执行器 (Code Executor) 工具调用 Vertex 代码解释器扩展 (Vertex Code Interpreter Extension)。

Google 和 ADK 为安全的代码执行提供了多种选择。Vertex Gemini Enterprise API 代码执行功能 使智能体能够通过启用 tool_execution 工具来利用服务器端的沙盒化代码执行。对于执行数据分析的代码，你可以在 ADK 中使用 代码执行器 工具调用 Vertex 代码解释器扩展。

---

评估

参见评估智能体。

---

VPC-SC 边界和网络控制

如果你在 VPC-SC 边界内运行智能体，这将确保所有 API 调用只会操作边界内的资源，从而降低数据泄露的可能性。

然而，身份和边界仅能提供对智能体操作的粗略控制。工具内护栏缓解了这些限制，并赋予智能体开发者更多权力来精细控制允许哪些操作。

---

其他安全风险

在 UI 中始终转义模型生成的内容

当智能体输出在浏览器中可视化时必须务必小心：如果在 UI 中没有正确转义 HTML 或 JS 内容，模型返回的文本可能会被执行，导致数据泄露。例如，间接提示词注入可能欺骗模型包含一个 <img> 标签，使浏览器将会话内容发送到第三方站点；或构建恶意 URL，如果被点击，会将数据发送到外部站点。正确转义此类内容可以确保模型生成的文本不会被浏览器解释为代码。