ADK 的 GKE 代码执行器工具¶

ADK 已支持Python v1.14.0

GKE 代码执行器 (GkeCodeExecutor) 通过利用 Google Kubernetes Engine (GKE) 为运行 LLM 生成的代码提供了一种安全且可扩展的方法。对于 GKE 上对安全性和隔离性要求极高的生产环境，你应该使用此执行器。它支持两种执行模式：

沙箱模式 (推荐)： 利用 Agent Sandbox 客户端，在根据模板按需创建的沙箱实例中执行代码。此模式通过使用预热沙箱提供更低的延迟，并支持与沙箱环境更直接的交互。
任务 (Job) 模式： 使用带有 gVisor 的 GKE Sandbox 环境进行工作负载隔离。对于每个代码执行请求，它都会通过强化的 Pod 配置动态创建一个临时的、沙箱化的 Kubernetes Job。此模式用于向后兼容。

执行模式¶

沙箱模式 (`executor_type="sandbox"`)¶

这是推荐的模式。它使用 k8s-agent-sandbox 客户端库与 GKE 集群中的 Agent Sandbox 进行通信。当发出执行代码的请求时，它将执行以下步骤：

使用指定的模板创建 SandboxClaim。
等待沙箱实例准备就绪。
在认领的沙箱中执行代码。
检索标准输出和错误。
删除 SandboxClaim，进而清理沙箱实例。

这种方法比任务 (Job) 模式快，因为它利用了预热沙箱并优化了 Agent Sandbox 控制器提供的启动时间。

主要优势：

除了 Job 模式的所有优势外，沙箱模式还提供以下功能：

更低延迟： 旨在减少与创建完整 Kubernetes Job 相比的启动时间。
托管环境： 利用 Agent Sandbox 框架进行沙箱生命周期管理。

先决条件：

在你的 GKE 集群中已有 Agent Sandbox 部署，包括沙箱控制器及其扩展（例如沙箱认领控制器和沙箱预热池控制器）、路由器、网关以及相关的 SandboxTemplate 资源（例如 python-sandbox-template）。
ADK 智能体创建和删除 SandboxClaim 资源所需的 RBAC 权限。

任务 (Job) 模式 (`executor_type="job"`)¶

此模式用于向后兼容。当发出执行代码的请求时，GkeCodeExecutor 将执行以下步骤：

创建 ConfigMap: 创建一个 Kubernetes ConfigMap 来存储需要执行的 Python 代码。
创建沙箱化 Pod: 创建一个新的 Kubernetes Job，该 Job 反过来创建一个具有强化安全上下文并启用 gVisor 运行时的 Pod。ConfigMap 中的代码被挂载到此 Pod 中。
执行代码： 代码在沙箱化的 Pod 中执行，与底层节点和其他工作负载隔离。
检索结果： 从 Pod 的日志中捕获执行的标准输出和错误流。
清理资源： 执行完成后，Job 和关联的 ConfigMap 会自动删除，确保不会留下任何制品。

主要优势：

增强的安全性： 代码在具有内核级隔离的 gVisor 沙箱环境中执行。
临时环境： 每次代码执行都在其自己的临时 Pod 中运行，以防止执行之间的状态转移。
资源控制： 你可以为执行 Pod 配置 CPU 和内存限制，以防止资源滥用。
可扩展性： 允许并行运行大量代码执行，由 GKE 处理底层节点的调度和缩放。
最简设置： 依赖于标准的 GKE 功能和 gVisor。

系统要求¶

要成功部署带有 GKE 代码执行器工具的 ADK 项目，必须满足以下要求：

具有 启用 gVisor 的节点池 的 GKE 集群（Job 模式的默认镜像和典型的 Agent Sandbox 模板均需要）。
智能体的服务账号需要特定的 RBAC 权限：
- Job 模式： 创建、查看和删除 Jobs；管理 ConfigMaps；列出 Pods 并读取其日志。有关 Job 模式完整、现成的配置，请参阅 deployment_rbac.yaml 示例。
- 沙箱模式： 在部署 Agent Sandbox 的命名空间内创建、获取、查看和删除 SandboxClaim 和 Sandbox 资源的权限。
使用适当的 extra 安装客户端库：pip install google-adk[gke]

配置参数¶

GkeCodeExecutor 可以使用以下参数进行配置：

参数	类型	描述
`namespace`	`str`	创建执行资源（Jobs 或 SandboxClaims）的 Kubernetes 命名空间。默认为 `"default"`。
`executor_type`	`Literal["job", "sandbox"]`	指定执行模式。默认为 `"job"`。
`image`	`str`	(Job 模式) 用于执行 Pod 的容器镜像。默认为 `"python:3.11-slim"`。
`timeout_seconds`	`int`	(Job 模式) 代码执行的超时时间（秒）。默认为 `300`。
`cpu_requested`	`str`	(Job 模式) 为执行 Pod 请求的 CPU 量。默认为 `"200m"`。
`mem_requested`	`str`	(Job 模式) 为执行 Pod 请求的内存量。默认为 `"256Mi"`。
`cpu_limit`	`str`	(Job 模式) 执行 Pod 可以使用的最大 CPU 量。默认为 `"500m"`。
`mem_limit`	`str`	(Job 模式) 执行 Pod 可以使用的最大内存量。默认为 `"512Mi"`。
`kubeconfig_path`	`str`	用于身份验证的 kubeconfig 文件的路径。回退到集群内配置或默认的本地 kubeconfig。
`kubeconfig_context`	`str`	要使用的 `kubeconfig` 上下文。
`sandbox_gateway_name`	`str \| None`	(沙箱模式) 要使用的沙箱网关名称。可选。
`sandbox_template`	`str \| None`	(沙箱模式) 要使用的 `SandboxTemplate` 名称。默认为 `"python-sandbox-template"`。

使用示例¶

Python - 沙箱模式 (推荐)Python - 任务 (Job) 模式

from google.adk.agents import LlmAgent
from google.adk.code_executors import GkeCodeExecutor
from google.adk.code_executors import CodeExecutionInput
from google.adk.agents.invocation_context import InvocationContext

# 初始化沙箱模式的执行器
# 命名空间应具有 SandboxClaims 和 Sandbox 的 RBAC 权限
gke_sandbox_executor = GkeCodeExecutor(
    namespace="agent-sandbox-system",  # 通常是安装 agent-sandbox 的位置
    executor_type="sandbox",
    sandbox_template="python-sandbox-template",
    sandbox_gateway_name="your-gateway-name", # 可选
)

# 直接执行示例：
ctx = InvocationContext()
result = gke_sandbox_executor.execute_code(ctx, CodeExecutionInput(code="print('来自沙箱模式的问候')"))
print(result.stdout)

# 配合智能体使用的示例：
gke_sandbox_agent = LlmAgent(
    name="gke_sandbox_coding_agent",
    model="gemini-2.5-flash",
    instruction="你是一个乐于助人的 AI 智能体，负责编写并使用沙箱执行 Python 代码。",
    code_executor=gke_sandbox_executor,
)

from google.adk.agents import LlmAgent
from google.adk.code_executors import GkeCodeExecutor
from google.adk.code_executors import CodeExecutionInput
from google.adk.agents.invocation_context import InvocationContext

# 初始化 Job 模式的执行器
# 命名空间应具有 Jobs, ConfigMaps, Pods, Logs 的 RBAC 权限
gke_executor = GkeCodeExecutor(
    namespace="agent-ns",
    executor_type="job",
    timeout_seconds=600,
    cpu_limit="1000m",  # 1 个 CPU 核心
    mem_limit="1Gi",
)

# 直接执行示例：
ctx = InvocationContext()
result = gke_executor.execute_code(ctx, CodeExecutionInput(code="print('来自 Job 模式的问候')"))
print(result.stdout)

# 配合智能体使用的示例：
gke_agent = LlmAgent(
    name="gke_coding_agent",
    model="gemini-2.5-flash",
    instruction="你是一个乐于助人的 AI 智能体，负责编写并执行 Python 代码。",
    code_executor=gke_executor,
)