ADK 的 Parameter Manager¶
Supported in ADKPython v1.30.0
Google Cloud Parameter Manager 集成为 Agent Development Kit (ADK) 智能体提供了标准接口,用于连接 Google Cloud Parameter Manager 服务并在运行时检索渲染后的参数值。此模块使你可以将 Google Cloud Parameter Manager 服务用作智能体指令和工具配置的单一真实来源。
使用场景¶
Parameter Manager 集成支持多种操作:
- 动态指令更新:你可以即时发布参数以防御提示注入攻击、更新强制性免责声明资源,或自动调整智能体语气,而无需完整代码重新部署。
- 功能标志和参数管理:你可以将配置存储为 JSON 负载,并通过工具上下文检索它们,以降低查询速率或在实验性和生产性 API 端点之间切换。
- 通过输入输出对提高准确性:你可以将少样本示例存储为 YAML 文件,并加载到会话状态中,以随时间提高智能体性能。
- 即时工具授权:你可以按需将密钥加载到内存中,而非在初始化代码中使用不安全的静态 API 密钥。
- 安全的多租户工作流:你可以存储映射到用户的 Parameter Manager ID,并使用回调将解析后的 OAuth 令牌恢复到会话状态。
- 加密的系统任务:你可以防止主数据库密码在后台轮询任务期间进入大语言模型 (LLM) 对话历史。
- 多区域部署:你可以在全球部署中维护共享逻辑,同时使用区域 Parameter Manager 覆盖来应用本地货币和联系信息。
前置条件¶
在配置集成之前,你必须满足以下要求:
- 所需软件版本:ADK Python 版本 v1.30.0 或更高版本
- 所需账户/API:一个已启用 Parameter Manager API、Secret Manager API 和 Agent Development Kit API 的 Google Cloud 项目。
完成以下设置步骤:
- 使用 ADK 设置一个智能体。
- 创建一个参数。
- 为你的智能体身份授予 Parameter Manager Parameter Accessor 角色(
roles/parametermanager.parameterAccessor)。此角色允许你的智能体在运行时渲染参数配置。 - 如果你的参数包含嵌入的密钥,请为你的参数资源授予 Secret Manager Secret Accessor 角色(
roles/secretmanager.secretAccessor)。此跨服务权限允许 Parameter Manager 代表智能体解析引用的密钥。更多信息请参阅将 Secret Manager Secret Accessor 角色授予参数。
安装¶
安装 ADK 扩展包以启用 Parameter Manager 集成:
与智能体配合使用¶
以下示例展示了完整的、可工作的代码,使用全局或区域端点在 ADK 智能体中安全检索参数。
全局参数¶
import os
from google.adk import Agent
from google.adk.integrations.parameter_manager.parameter_client import ParameterManagerClient
# 从全局 Parameter Manager 获取参数
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
parameter_id = os.environ.get("ADK_TEST_PARAMETER_ID")
parameter_version = os.environ.get("ADK_TEST_PARAMETER_VERSION", "latest")
if not project_id or not parameter_id:
raise ValueError("必须设置 GOOGLE_CLOUD_PROJECT 和 ADK_TEST_PARAMETER_ID 环境变量。")
resource_name = f"projects/{project_id}/locations/global/parameters/{parameter_id}/versions/{parameter_version}"
print("正在从全局 Parameter Manager 获取参数...")
# 初始化 Parameter Manager 客户端
client = ParameterManagerClient()
# 获取参数
try:
parameter_payload = client.get_parameter(resource_name)
print("成功获取参数。")
except Exception as e:
print(f"获取参数时出错:{e}")
raise e
# 初始化智能体
root_agent = Agent(
model='gemini-2.5-flash',
name='root_agent',
description='一个帮助回答用户问题的助手。',
instruction='尽你所能回答用户的问题',
)
print("智能体初始化成功。")
区域参数¶
import os
from google.adk import Agent
from google.adk.integrations.parameter_manager.parameter_client import ParameterManagerClient
# 从区域 Parameter Manager 获取参数
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_PROJECT_LOCATION")
parameter_id = os.environ.get("ADK_TEST_PARAMETER_ID")
parameter_version = os.environ.get("ADK_TEST_PARAMETER_VERSION", "latest")
if not project_id or not location or not parameter_id:
raise ValueError("必须设置 GOOGLE_CLOUD_PROJECT、GOOGLE_CLOUD_PROJECT_LOCATION 和 ADK_TEST_PARAMETER_ID 环境变量。")
resource_name = f"projects/{project_id}/locations/{location}/parameters/{parameter_id}/versions/{parameter_version}"
print(f"正在从区域 Parameter Manager ({location}) 获取参数...")
# 初始化 Parameter Manager 客户端(区域)
client = ParameterManagerClient(location=location)
# 获取参数
try:
parameter_payload = client.get_parameter(resource_name)
print("成功获取参数。")
except Exception as e:
print(f"获取参数时出错:{e}")
raise e
# 初始化智能体
root_agent = Agent(
model='gemini-2.5-flash',
name='root_agent',
description='一个帮助回答用户问题的助手。',
instruction='尽你所能回答用户的问题',
)
print("智能体初始化成功。")