用于 ADK 的 GKE 代码执行器工具¶

Supported in ADKPython 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。
等待沙箱实例准备就绪。
在认领的沙箱中执行代码。
检索标准输出 (stdout) 和错误 (stderr)。
删除 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，进而创建一个具有强化安全上下文并启用 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 并读取其 Logs。有关 Job 模式完整、现成的配置参考，请参阅 deployment_rbac.yaml 示例。
- 沙箱模式：在部署 Agent Sandbox 的命名空间内，需要拥有创建、获取、查看和删除 SandboxClaim 和 Sandbox 资源的权限。
使用相应的 extra label 安装客户端库：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 文件路径。若不提供，则回退到集群内（In-cluster）配置或默认的本地配置。
`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.0-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.0-flash",
    instruction="你是一个得力的 AI 智能体，负责编写并执行 Python 代码。",
    code_executor=gke_executor,
)