AI 智能体的安全与保障¶
随着 AI 智能体能力的增长，确保它们安全、可靠地运行并与你的品牌价值观保持一致至关重要。未受控制的智能体可能带来风险，包括执行不符合预期或有害的操作，如数据泄露，以及生成可能影响你品牌声誉的不适当内容。风险来源包括模糊的指令、模型幻觉、来自恶意用户的越狱和提示注入，以及通过工具使用产生的间接提示注入。
Google Cloud Vertex AI 提供了多层方法来缓解这些风险，使你能够构建强大且可信的智能体。它提供了几种机制来建立严格的边界，确保智能体只执行你明确允许的操作：
- 身份和授权：通过定义智能体和用户身份验证来控制智能体以谁的身份行事。
-
用于筛选输入和输出的防护栏：精确控制你的模型和工具调用。
-
身份和授权 (Identity and Authorization)：通过定义智能体和用户身份验证来控制智能体以谁的身份行事。
-
用于筛选输入和输出的防护栏 (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）实现额外的安全层，以筛选输入和输出。
-
沙盒化代码执行 (Sandboxed code execution): 通过沙盒化环境防止模型生成的代码导致安全问题。
- 评估和追踪 (Evaluation and tracing): 使用评估工具来评估智能体最终输出的质量、相关性和正确性。使用追踪来深入了解智能体操作，分析智能体为达成解决方案所采取的步骤，包括其工具选择、策略和方法的效率。
- 网络控制和 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 范围）。通常，这些范围比智能体实际需要的访问权限更广泛，需要下面的技术来进一步约束智能体的操作。
用于筛选输入和输出的防护栏¶
工具内防护栏¶
可以在设计工具时考虑安全性：我们可以创建只暴露我们希望模型采取的操作而不暴露其他操作的工具。通过限制我们提供给智能体的操作范围，我们可以确定性地消除我们永远不希望智能体采取的一类恶意操作。
This approach relies on the fact that tools receive two types of input: arguments, which are set by the model, and Tool Context, which can be set deterministically by the agent developer. We can rely on the deterministically set information to validate that the model is behaving as-expected. (Note: In TypeScript, Tool Context corresponds to the unified Context type.)
这种方法依赖于这样一个事实：工具接收两种类型的输入：由模型设置的参数，以及可以由智能体开发者确定性设置的工具上下文。我们可以依靠确定性设置的信息来验证模型是否按预期行为。
例如，查询工具可以设计为期望从 Tool Context 中读取策略。
# 概念示例：设置用于工具上下文的策略数据
# 在真实的 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)
# 对于本例，我们假设它被存储在某个可访问的位置。
// Conceptual example: Setting policy data intended for tool context
// In a real ADK app, this might be set in InvocationContext.session.state
// or passed during tool initialization, then retrieved via Context.
const policy: {[key: string]: any} = {}; // 假设策略是一个对象
policy['select_only'] = true;
policy['tables'] = ['mytable1', 'mytable2'];
// Conceptual: Storing policy where the tool can access it via Context later.
// This specific line might look different in practice.
// For example, storing in session state:
invocationContext.session.state["query_tool_policy"] = policy;
// 或者在工具初始化期间传递：
const queryTool = new QueryTool({policy: policy});
// 对于本例，我们假设它被存储在某个可访问的位置。
// 概念示例：设置用于工具上下文的策略数据
// 在真实的 ADK 应用程序中，这可能使用会话状态服务进行设置。
// `ctx` 是回调或自定义智能体中可用的 `agent.Context`。
policy := map[string]interface{}{
"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);
// 对于本例，我们假设它被存储在某个可访问的位置。
During the tool execution, Tool Context will be passed to the tool (Note: In TypeScript, this is passed as the unified Context type):
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 {
// Assume 'policy' is retrieved from context, e.g., via session state:
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(query string, toolContext *tool.Context) (any, error) {
// 假设 'policy' 是从上下文中获取的，例如通过会话状态：
policyAny, err := toolContext.State().Get("query_tool_policy")
if err != nil {
return nil, fmt.Errorf("无法检索策略：%w", err)
}
policy, _ := policyAny.(map[string]interface{})
actualTables := explainQuery(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(query)), "SELECT") {
return nil, fmt.Errorf("策略限制查询只能是 SELECT 语句")
}
}
// --- 占位符策略强制执行结束 ---
fmt.Printf("执行已验证的查询（假设）：%s\n", query)
return map[string]interface{}{"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 (!queryToolPolicy.get("tables").containsAll(actualTables)) {
List<String> allowedPolicyTables =
(List<String>) queryToolPolicy.getOrDefault("tables", new ArrayList<String>());
String allowedTablesString =
allowedPolicyTables.isEmpty() ? "(None defined)" : String.join(", ", allowedPolicyTables);
return String.format(
"Error: Query targets unauthorized tables. Allowed: %s", allowedTablesString);
}
if (!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 模型带有内置的安全机制，可用于提高内容和品牌安全。
- 内容安全过滤器 (Content safety filters): 内容过滤器 可以帮助阻止有害内容的输出。它们作为分层防御的一部分，独立于 Gemini 模型运行，以对抗试图越狱模型的威胁行为者。Vertex AI 上的 Gemini 模型使用两种类型的内容过滤器：
- 不可配置的安全过滤器 (Non-configurable safety filters) 自动阻止包含违禁内容（例如儿童性虐待材料 (CSAM) 和个人身份信息 (PII)）的输出。
- 可配置的内容过滤器 (Configurable content filters) 允许你根据概率和严重性分数在四个危害类别（仇恨言论、骚扰、色情和危险内容）中定义阻止阈值。这些过滤器默认关闭，但你可以根据需要进行配置。
- 安全系统指令 (System instructions for safety): Vertex AI 中 Gemini 模型的 系统指令 为模型提供了关于如何行为以及生成何种内容的直接指导。通过提供具体指令，你可以主动引导模型避免生成不良内容，以满足你组织的独特需求。你可以制定系统指令来定义内容安全准则，例如违禁和敏感主题，以及免责声明语言，以及品牌安全准则，以确保模型的输出与你的品牌声音、语调、价值观和目标受众保持一致。
虽然这些措施对内容安全非常强大，但你需要额外的检查来减少智能体不一致、不安全行为和品牌安全风险。
安全防护措施的回调和插件¶
回调提供了一种简单的、特定于智能体的方法来为工具和模型 I/O 添加预验证，而插件提供了一种可重用的解决方案来在多个智能体中实现通用安全策略。
当无法修改工具以添加防护栏时，可以使用 Before Tool Callback 函数为调用添加预验证。该回调可以访问智能体的状态、请求的工具和参数。这种方法非常通用，甚至可以用来创建可重用的工具策略通用库。但如果要强制执行防护栏的信息不直接体现在参数中，则可能不适用于所有工具。
# 假设的回调函数
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
# 假设的智能体设置
root_agent = LlmAgent( # 使用特定的智能体类型
model='gemini-2.0-flash',
name='root_agent',
instruction="...",
before_tool_callback=validate_tool_params, # 分配回调
tools = [
# ... 工具函数或 Tool 实例列表 ...
# 例如 query_tool_instance
]
)
// 假设的回调函数
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-2.5-flash',
name: 'root_agent',
instruction: "...",
beforeToolCallback: validateToolParams, // 分配回调
tools: [
// ... 工具函数或工具实例列表 ...
// 例如，queryToolInstance
]
});
import (
"fmt"
"reflect"
"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 是否与参数匹配
expectedUserID, err := ctx.State().Get("session_user_id")
if err != nil {
// 这是一个意外的失败，返回错误。
return nil, fmt.Errorf("内部错误：状态中未找到 session_user_id：%w", err)
}
expectedUserIDVal, ok := expectedUserID.(string)
if !ok {
return nil, fmt.Errorf("内部错误：状态中的 session_user_id 不是字符串，得到 %T", expectedUserID)
}
actualUserIDInArgs, exists := args["user_id_param"]
if !exists {
// 处理 user_id_param 不在参数中的情况
fmt.Println("验证失败：参数中缺少 user_id_param！")
return map[string]any{"error": "工具调用被阻止：参数中缺少 user_id_param。"}, nil
}
actualUserID, ok := actualUserIDInArgs.(string)
if !ok {
// 处理 user_id_param 不是字符串的情况
fmt.Println("验证失败：user_id_param 不是字符串！")
return map[string]any{"error": "工具调用被阻止：user_id_param 不是字符串。"}, nil
}
if actualUserID != expectedUserID {
fmt.Println("验证失败：User ID 不匹配！")
// 返回一个 map 以阻止工具执行并向模型提供反馈。
// 这不是 Go 错误，而是给智能体的消息。
return map[string]any{"error": "工具调用被阻止：User ID 不匹配。"}, nil
}
// 如果验证通过，返回 nil, nil 以允许工具调用继续
fmt.Println("回调验证通过。")
return nil, nil
}
// 假设的智能体设置
// rootAgent, err := llmagent.New(llmagent.Config{
// Model: "gemini-2.0-flash",
// 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();
}
// 假设的智能体设置
public void runAgent() {
LlmAgent agent =
LlmAgent.builder()
.model("gemini-2.0-flash")
.name("AgentWithBeforeToolCallback")
.instruction("...")
.beforeToolCallback(this::validateToolParams) // 分配回调
.tools(anyToolToUse) // 定义要使用的工具
.build();
}
然而，在为你的智能体应用程序添加安全防护措施时，插件是实施不特定于单个智能体的策略的推荐方法。插件设计为自包含和模块化，允许你为特定安全策略创建单独的插件，并在运行器级别全局应用它们。这意味着可以配置一次安全插件并将其应用于使用该运行器的每个智能体，确保整个应用程序的安全防护措施保持一致，而无需重复代码。
一些示例包括：
-
Gemini 作为监督插件：此插件使用 Gemini Flash Lite 评估用户输入、工具输入和输出以及智能体的响应是否适当、提示注入和越狱检测。该插件配置 Gemini 作为安全过滤器来缓解内容安全、品牌安全和智能体不一致问题。该插件配置为将用户输入、工具输入和输出以及模型输出传递给 Gemini Flash Lite，由它决定智能体的输入是否安全。如果 Gemini 认为输入不安全，智能体会返回预定的响应："抱歉，我无法帮助处理这个问题。我可以帮助你处理其他事情吗？"。
-
Model Armor 插件：一个查询 Model Armor API 的插件，用于在智能体执行的指定点检查潜在的内容安全违规。类似于 Gemini 作为监督 插件，如果 Model Armor 发现有害内容匹配，它会向用户返回预定的响应。
-
PII 编辑插件：一个专门为 工具前置回调 设计的专用插件，专门用于在工具处理或发送到外部服务之前编辑个人身份信息。
沙盒化代码执行¶
代码执行是一个具有额外安全含义的特殊工具：必须使用沙盒化来防止模型生成的代码危害本地环境，可能造成安全问题。
Google 和 ADK 提供了多种安全代码执行的选项。Vertex Gemini Enterprise API 代码执行功能 使智能体能够通过启用 tool_execution 工具来利用服务器端的沙盒代码执行。对于执行数据分析的代码，你可以在 ADK 中使用 代码执行器 工具来调用 Vertex 代码解释器扩展。
如果这些选项都不能满足你的需求，你可以使用 ADK 提供的构建模块构建自己的代码执行器。我们建议创建封闭的执行环境：不允许网络连接和 API 调用，以避免不受控制的数据泄露；并在执行之间完全清理数据，以避免跨用户泄露问题。
评估¶
参见评估智能体。
VPC-SC 边界和网络控制¶
如果你在 VPC-SC 边界内执行智能体，这将保证所有 API 调用只会操作边界内的资源，从而降低数据泄露的可能性。
然而，身份和边界只提供对智能体操作的粗略控制。工具使用防护栏缓解了这些限制，并赋予智能体开发者更多权力来精细控制允许哪些操作。
其他安全风险¶
在 UI 中始终转义模型生成的内容¶
当智能体输出在浏览器中可视化时必须小心：如果在 UI 中没有正确转义 HTML 或 JS 内容，模型返回的文本可能会被执行，导致数据泄露。例如，间接提示注入可能欺骗模型包含一个 img 标签，使浏览器将会话内容发送到第三方站点；或构建 URL，如果点击，会将数据发送到外部站点。正确转义此类内容必须确保模型生成的文本不会被浏览器解释为代码。