# AgentCube插件 随着生成式AI和大语言模型(LLM)应用的快速发展，AI Agent工作负载正在成为云原生环境中的重要负载类型。这些工作负载具有独特的特征：持续交互、状态保持、间歇性活跃的长会话特性。然而，现有的工作负载管理模式无法满足AI Agent的这些特殊需求： - **冷启动延迟高**：Agent的交互通常要求毫秒级响应。然而，原生的K8s Pod启动流程（调度、IP 分配、镜像拉取、容器启动）往往在秒级甚至分钟级。对于需要频繁拉起Code Interpreter（代码解释器）或临时子Agent的场景，这种冷启动延迟是用户无法接受的。 - **资源利用效率低**：Agent是典型的IO密集型负载。在一次会话中，90%的时间Agent可能都在等待LLM生成Token或等待外部工具响应。如果在K8s上为每个Agent 独占一个Pod，会导致大量的CPU/Memory资源在等待期间被闲置浪费，却无法被其他任务复用。 - **状态管理缺失**：K8s对"无状态（Stateless）"工作负载天然友好，但 Agent 高度依赖"上下文（Context/Memory）"。在原生K8s中，Pod重启意味着内存数据丢失，开发者被迫在应用层通过外部存储重建上下文，这带来了巨大的复杂性和网络开销。 - **安全隔离难题** ：高级Agent（如Data Analyst）需要运行由LLM生成的不可信代码。但普通的runC容器如果运行*rm -rf /*具有极高风险。企业级Agent平台迫切需要一种既能快速启动，又能提供强隔离（如MicroVM）的沙箱环境。 为了解决上述问题，CCE Autopilot集群基于[AgentCube](https://github.com/volcano-sh/agentcube)提供了AgentCube插件。AgentCube是一个高性能AI Agent编排工具，用于在Kubernetes上原生支持和管理AI Agent工作负载，提供极速启动、极速调度、原生会话管理、Serverless 化的弹性伸缩等创新能力。 ![](https://support.huaweicloud.com/usermanual-cce-autopilot/public_sys-resources/caution_3.0-zh-cn.png) - AgentCube是一个开源项目，目前处于Alpha阶段，CCE不对软件的适用性、可靠性或性能做出任何保证，用户需自行承担使用过程中的风险（包括但不限于数据安全、系统稳定性、资源消耗等）。 - 该特性当前正处于上线阶段，已发布区域请以控制台实际为准。 #### 功能介绍 AgentCube引入了两个核心的CRD来定义Agent工作负载： 1. **AgentRuntime**：面向长会话、复杂的对话式Agent，支持定义会话的生命周期、资源配额以及持久化策略。 2. **CodeInterpreter**：面向短任务、高频的代码执行环境，强调"用完即毁"和极致的安全隔离。 ![](https://support.huaweicloud.com/usermanual-cce-autopilot/zh-cn_image_0000002538507982.png "点击放大") **核心功能：** - 极速启动 为了消除冷启动的挑战，AgentCube 实现了 Warm Pool（预热池） 机制。系统会预先启动并暂停一组持有基础环境的 MicroVM 沙箱。当 Agent 请求到来时，AgentCube 能够通过 "Claim-and-Go" 的方式，在毫秒级将预热的沙箱分配给会话，实现近乎零延迟的启动体验。 - 原生会话管理 AgentCube 引入了 Session ID 作为核心路由标识，便于保证业务上下文的连续性。 - 请求路由： AgentCube Router 能够识别请求中的 x-agentcube-session-id，自动将其路由到对应的活跃沙箱。 - 自动的沙箱激活： 当前会话对应的沙箱处于休眠状态时，AgentCube Router 能够自动激活沙箱。 - 基于会话的端到端隔离： AgentCube 会自动为每个会话分配独立的沙箱环境，确保计算、内存与文件系统的完全隔离，防止跨租户的数据泄露。 更多关于AgentCube的详细介绍，请参见[AgentCube](https://github.com/volcano-sh/agentcube/blob/main/README.md)。 #### 前提条件 - 已创建v1.31及以上的CCE Autopilot集群。 - 已经部署了Redis，并确保AgentCube可访问Redis。 #### 安装插件 1. 登录[CCE控制台](https://console.huaweicloud.com/cce2.0/?#/cce/cluster/list)，单击集群名称进入集群。 2. 单击左侧导航栏的"插件中心"，在右侧找到**AgentCube**插件，单击"安装"。 3. 在安装插件页面进行相关配置。 表1参数说明 | 参数 | 说明 | 示例 | |:---|:---|:---| | redis.addr | Redis的地址，必须配置。 | ip地址:端口 | | redis.password | Redis的密码，必须配置。 | - | | agentSandbox.install | 是否自动安装agent-sandbox。AgentCube插件运行时依赖[agent-sandbox](https://github.com/kubernetes-sigs/agent-sandbox)，当参数配置为true时会自动安装。若您的集群已手动安装了agent-sandbox，则可配置为false跳过安装。 | true | | agentSandbox.extensions | 是否启用agent-sandbox的extension controller。 | true | | router.service.type | agentcube-router组件service访问的类型。 | ClusterIP | | workloadmanager.service.type | workloadmanager组件service访问的类型。 | ClusterIP | 4. 单击"安装"。当该插件状态为"运行中"时，则说明插件安装成功。 #### 组件说明 表2AgentCube组件 | 容器组件 | 说明 | 资源类型 | |:---|:---|:---| | workloadmanager | 管理Sandbox和CodeInterpreter的生命周期。 | Deployment | | agentcube-router | API 网关，代理客户端请求到沙箱实例。 | Deployment | | agent-sandbox-controller | 管理AgentSandbox资源。 | Deployment | workloadmanager、agentcube-router、agent-sandbox-controller组件均部署在agentcube命名空间中。 ![](https://support.huaweicloud.com/usermanual-cce-autopilot/public_sys-resources/note_3.0-zh-cn.png) 插件安装后如果卸载，会将agentcube命名空间下的创建的组件、Service删除，但不会删除插件创建的CRD。 #### 使用示例 插件安装完成后，您可以参考以下示例进行使用。 #### 示例1：使用CodeInterpreter 1. 创建code-interpreter-demo.yaml。 ``` apiVersion: runtime.agentcube.volcano.sh/v1alpha1 kind: CodeInterpreter metadata: name: my-codeinterpreter namespace: default spec: template: # 使用 PicoD 镜像 image: ghcr.io/volcano-sh/picod:latest # 资源限制 resources: requests: cpu: "100m" memory: "128Mi" limits: cpu: "500m" memory: "512Mi" # 会话配置 sessionTimeout: "15m" # 空闲 15 分钟后超时 maxSessionDuration: "8h" # 最大会话时长 8 小时 # 关键功能：预热池 warmPoolSize: 5 # 预热 5 个 Pod（近零延迟） ``` 2. 部署CodeInterpreter。 ``` kubectl apply -f code-interpreter-demo.yaml ``` 创建完成后，使用以下命令查看状态。 ``` kubectl get codeinterpreter my-codeinterpreter -o yaml ``` 3. 验证预热池创建。 ``` # 查看创建的 SandboxWarmPool kubectl get sandboxwarmpool -n default # 查看预热池的 Pod kubectl get pods |grep my-codeinterpreter ``` 4. 创建测试脚本（需本地已安装python\>=3.11）。 1. 创建文件夹test_codeinterpreter。 在其中创建文件test_codeinterpreter.py、requirements.txt，内容如下： - test_codeinterpreter.py文件： ``` import time from agentcube import CodeInterpreterClient import os WORKLOAD_MANAGER_URL = os.getenv('WORKLOAD_MANAGER_URL', 'http://workloadmanager.agentcube.svc.cluster.local:8080') ROUTER_URL = os.getenv('ROUTER_URL', 'http://agentcube-router.agentcube.svc.cluster.local:8080') CODEINTERPRETER_NAME = os.getenv('CODEINTERPRETER_NAME', 'my-codeinterpreter') CODEINTERPRETER_NAMESPACE = os.getenv('CODEINTERPRETER_NAMESPACE', 'default') def basic_operations(): """演示基本操作""" print("=== basic operations ===\n") with CodeInterpreterClient( name=CODEINTERPRETER_NAME, namespace=CODEINTERPRETER_NAMESPACE, workload_manager_url=WORKLOAD_MANAGER_URL, router_url=ROUTER_URL, verbose=True, ) as client: print(f"session ID: {client.session_id}\n") # 1. 执行 Shell 命令 print("1. Shell command (whoami):") output = client.execute_command("whoami") print(f"output: {output.strip()}\n") # 2. 执行 Python 代码 print("2. Python code execution:") code = """ import sys print(f"Python version: {sys.version}") import math print(f"π value: {math.pi:.10f}") """ result = client.run_code("python", code) print(f"output:\n{result}\n") # 3. 文件操作 print("3. file operate:") client.write_file("Hello, AgentCube!", "hello.txt") client.write_file("42", "number.txt") files = client.list_files(".") print(f"file list: {[f['name'] for f in files]}\n") print("session auto delete") def session_reuse_workflow(): """ 演示会话复用 - 对 AI 工作流很重要 场景：分步处理数据，需要在会话间保持文件系统状态 """ print("=== session reuse workflow ===\n") # 步骤 1：创建会话并写入数据 print("step 1: Create a session and write initial data.") client1 = CodeInterpreterClient( name=CODEINTERPRETER_NAME, namespace=CODEINTERPRETER_NAMESPACE, workload_manager_url=WORKLOAD_MANAGER_URL, router_url=ROUTER_URL, verbose=True, ) # 写入多个文件 client1.write_file("100", "counter.txt") client1.write_file("[]", "results.json") session_id = client1.session_id print(f"session ID: {session_id}") print("The file system status has been saved.\n") # 注意：不调用 client1.stop()，让会话保持活跃 # 步骤 2：复用会话，读取并处理数据 print("step 2: Reusing sessions and processing data.") client2 = CodeInterpreterClient( name=CODEINTERPRETER_NAME, namespace=CODEINTERPRETER_NAMESPACE, workload_manager_url=WORKLOAD_MANAGER_URL, router_url=ROUTER_URL, session_id=session_id, # 复用会话 verbose=True, ) code = """ import json import time # 读取计数器 with open('counter.txt') as f: counter = int(f.read().strip()) print(f"current count: {counter}") # 增加计数 counter += 1 with open('counter.txt', 'w') as f: f.write(str(counter)) # 读取结果列表 with open('results.json') as f: results = json.load(f) # 添加新结果 results.append({ 'timestamp': time.time(), 'counter': counter }) # 保存结果 with open('results.json', 'w') as f: json.dump(results, f, indent=2) print(f"new count: {counter}") print(f"result count: {len(results)}") """ result = client2.run_code("python", code) print(f"{result}\n") # 步骤 3：查看文件系统状态 print("step 3: Verifying the File System Statuses") files = client2.list_files(".") print(f"Files in session: {[f['name'] for f in files]}\n") # 清理会话 client2.stop() print("session is deleted") if __name__ == "__main__": basic_operations() print("\n" + "="*50 + "\n") session_reuse_workflow() ``` - requirements.txt文件： ``` agentcube_sdk ``` 2. 进入test_codeinterpreter文件夹，执行以下命令创建python虚拟环境。 ``` python -m venv venv ``` 3. 执行以下命令激活虚拟环境（Windows系统请在powershell下执行.\\venv\\Scripts\\activate）。 ``` source venv/bin/activate ``` 4. 执行以下命令安装依赖。 ``` pip install -r requirements.txt ``` 5. 运行测试脚本。 1. 设置环境变量（如果在集群外访问，需要进行设置，可为workloadmanager和agentcube-router的service配置ELB Ingress，通过ELB的弹性公网IP访问） - Linux环境执行： ``` export WORKLOAD_MANAGER_URL="http://addr:workloadmanager-port" export ROUTER_URL="http://addr:agentcube-router-port" ``` - Windows环境在powershell下执行： ``` $env:ROUTER_URL = "http://addr:agentcube-router-port" $env:WORKLOAD_MANAGER_URL = "http://addr:workloadmanager-port" ``` 其中addr为弹性公网IP地址，agentcube-router-port为agentcube-router Service对应Ingress的端口，workloadmanager-port为workloadmanager Service对应Ingress的端口。 2. 运行测试 ``` python test_codeinterpreter.py ``` 输出： ![](https://support.huaweicloud.com/usermanual-cce-autopilot/zh-cn_image_0000002538348044.png "点击放大") #### 示例二：使用AgentRuntime 1. 创建文件夹test_agentruntime。 在文件夹下创建agent.py、Dockerfile、requirements.txt文件，内容如下： - agent.py文件： ``` import json import os from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any, Dict from agentcube import CodeInterpreterClient WORKLOAD_MANAGER_URL = os.getenv('WORKLOAD_MANAGER_URL', 'http://workloadmanager.agentcube.svc.cluster.local:8080') ROUTER_URL = os.getenv('ROUTER_URL', 'http://agentcube-router.agentcube.svc.cluster.local:8080') CODEINTERPRETER_NAME = os.getenv('CODEINTERPRETER_NAME', 'my-codeinterpreter') CODEINTERPRETER_NAMESPACE = os.getenv('CODEINTERPRETER_NAMESPACE', 'default') class HelloAgentHandler(BaseHTTPRequestHandler): """HTTP handler for the Hello Agent.""" def do_GET(self): """Handle GET requests.""" if self.path == '/health': self._send_json_response({"status": "healthy", "agent": "hello-agent"}) elif self.path == '/': self._send_json_response({ "message": "Hello from Hello Agent!", "endpoints": [ "GET /health - Health check", "POST /greet - Greet someone", "POST /runcmd - Execute command using CodeInterpreterClient", "GET / - Agent information" ] }) else: self._send_error(404, "Endpoint not found") def do_POST(self): """Handle POST requests.""" if self.path == '/greet': self._handle_greet() elif self.path == '/runcmd': self._handle_runcmd() elif self.path == '/': self._handle_invoke() else: self._send_error(404, "Endpoint not found") def _handle_greet(self): """Handle greeting requests.""" try: content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) data = json.loads(post_data.decode('utf-8')) name = data.get('name', 'World') language = data.get('language', 'en') greetings = { 'en': f"Hello, {name}!", 'zh': f"你好, {name}!" } greeting = greetings.get(language, greetings['en']) response = { "greeting": greeting, "name": name, "language": language, "agent": "hello-agent" } self._send_json_response(response) except (json.JSONDecodeError, KeyError) as e: self._send_error(400, f"Invalid request: {str(e)}") def _handle_invoke(self): """Handle general invocation requests.""" try: content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) data = json.loads(post_data.decode('utf-8')) # Handle different types of prompts prompt = data.get('prompt', '') if not prompt: self._send_error(400, "Prompt is required") return response_data = { "response": f"Hello Agent received: {prompt}", "agent": "hello-agent", "timestamp": self._get_timestamp(), "original_prompt": prompt } self._send_json_response(response_data) except (json.JSONDecodeError, KeyError) as e: self._send_error(400, f"Invalid request: {str(e)}") def _handle_runcmd(self): """Handle runcmd requests using CodeInterpreterClient.""" try: code_interpreter = CodeInterpreterClient( name=CODEINTERPRETER_NAME, namespace=CODEINTERPRETER_NAMESPACE, workload_manager_url=WORKLOAD_MANAGER_URL, router_url=ROUTER_URL, verbose=True, ) code_interpreter.run_code("python","print('Hello from CodeInterpreterClient Python!')") output = code_interpreter.execute_command("echo 'Hello from CodeInterpreterClient Shell!'") response_data = { "output": output, "agent": "hello-agent", "timestamp": self._get_timestamp() } self._send_json_response(response_data) except (json.JSONDecodeError, KeyError) as e: self._send_error(400, f"Invalid request: {str(e)}") def _send_json_response(self, data: Dict[str, Any], status_code: int = 200): """Send JSON response.""" self.send_response(status_code) self.send_header('Content-type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() response = json.dumps(data, indent=2) self.wfile.write(response.encode('utf-8')) def _send_error(self, status_code: int, message: str): """Send error response.""" self.send_response(status_code) self.send_header('Content-type', 'application/json') self.end_headers() error_response = json.dumps({ "error": message, "status_code": status_code }, indent=2) self.wfile.write(error_response.encode('utf-8')) def _get_timestamp(self) -> str: """Get current timestamp.""" from datetime import datetime return datetime.now().isoformat() def log_message(self, format, *args): """Override log_message to reduce noise.""" pass def main(): """Main function to run the Hello Agent.""" port = int(os.environ.get('PORT', 8080)) print(f"Starting Hello Agent on port {port}") print(f"Health check: http://0.0.0.0:{port}/health") print(f"Greet endpoint: http://0.0.0.0:{port}/greet") print(f"Runcmd endpoint: http://0.0.0.0:{port}/runcmd") print(f"Agent endpoint: http://0.0.0.0:{port}/") server_address = ('', port) httpd = HTTPServer(server_address, HelloAgentHandler) try: print(f"Hello Agent is running on port {port}") httpd.serve_forever() except KeyboardInterrupt: print("\nShutting down Hello Agent") httpd.server_close() if __name__ == '__main__': main() ``` - Dockerfile文件： ``` FROM python:3.11-slim WORKDIR /app # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY agent.py /app/ # 暴露端口 EXPOSE 8080 # 运行应用 CMD ["python", "/app/agent.py"] ``` - requirements.txt文件内容 ``` agentcube_sdk ``` 2. 执行以下命令构建Agent示例镜像，并将其推送到镜像仓库。 ``` docker build -t my-agent-app:latest . ``` 3. 部署AgentRuntime agent-runtime-demo.yaml文件： ``` apiVersion: runtime.agentcube.volcano.sh/v1alpha1 kind: AgentRuntime metadata: name: my-agent-app namespace: default spec: targetPort: - pathPrefix: "/" port: 8080 protocol: "HTTP" podTemplate: labels: app: my-agent-app spec: containers: - name: my-agent-app image: {{agent image}} # 构建好的Agent示例镜像 env: - name: WORKLOAD_MANAGER_URL value: http://workloadmanager.agentcube.svc.cluster.local:8080 - name: ROUTER_URL value: http://agentcube-router.agentcube.svc.cluster.local:8080 - name: CODEINTERPRETER_NAME # 代码解释器的名字 value: my-codeinterpreter - name: CODEINTERPRETER_NAMESPACE value: default readinessProbe: httpGet: path: /health port: 8080 periodSeconds: 5 sessionTimeout: "15m" # 空闲 15 分钟后超时 maxSessionDuration: "8h" # 最大会话时长 8 小时 status: {} ``` 4. 部署并查看AgentRuntime状态 ``` kubectl apply -f agent-runtime-demo.yaml # 查看状态 kubectl get agentruntime my-agent-app -oyaml ``` 5. 创建测试脚本（需本地已安装python\>=3.11）。 1. 创建文件夹test_agentruntime。 在其中创建文件test_agentruntime.py、requirements.txt，文件内容如下： - test_agentruntime.py文件： ``` from agentcube.agent_runtime import AgentRuntimeClient import os import time # 配置环境变量 ROUTER_URL = os.getenv('ROUTER_URL', 'http://agentcube-router.agentcube.svc.cluster.local:8080') AGENT_NAME = os.getenv('AGENT_NAME', 'my-agent-app') AGENT_NAMESPACE = os.getenv('AGENT_NAMESPACE', 'default') saved_session_id = "" def test_basic_invoke(): """测试基本调用功能""" print("=== test basic invoke ===\n") # 创建客户端 agent_client = AgentRuntimeClient( agent_name=AGENT_NAME, namespace=AGENT_NAMESPACE, router_url=ROUTER_URL, verbose=True, timeout=500, connect_timeout=120.0 ) print(f"会话 ID: {agent_client.session_id}\n") # 第一次调用（创建新 Pod） print("first invoke（create new Pod）:") result = agent_client.invoke( payload={"prompt": "hello world!"}, ) print(f"response: {result}\n") # # 保存会话 ID global saved_session_id saved_session_id = agent_client.session_id def test_session_reuse(): """测试会话复用""" print(f"=== test session reuse ===\n") # 使用之前的会话 ID agent_client = AgentRuntimeClient( agent_name=AGENT_NAME, namespace=AGENT_NAMESPACE, router_url=ROUTER_URL, session_id=saved_session_id, # 复用会话 verbose=True, timeout=500, connect_timeout=120.0 ) print(f"reuse session ID: {agent_client.session_id}\n") # 第二次调用（复用现有 Pod） print("second invoke(reuse Pod):") result = agent_client.invoke( payload={"prompt": "How's the weather today?"}, ) print(f"response: {result}\n") def test_conversation(): """测试连续对话""" print("=== test conversation ===\n") agent_client = AgentRuntimeClient( agent_name=AGENT_NAME, namespace=AGENT_NAMESPACE, router_url=ROUTER_URL, verbose=True, timeout=500, connect_timeout=120.0 ) messages = [ "Introduce yourself", "What can you do?", "Write a Python function for me", ] for i, msg in enumerate(messages, 1): print(f"message {i}: {msg}") result = agent_client.invoke( payload={"prompt": msg}, ) print(f"response {i}: {result}\n") time.sleep(1) if __name__ == "__main__": test_basic_invoke() test_session_reuse() test_conversation() ``` - requirements.txt文件： ``` agentcube_sdk ``` 2. 进入test_agentruntime文件夹，执行以下命令创建python虚拟环境。 ``` python -m venv venv ``` 3. 执行以下命令激活虚拟环境（Windows系统请在powershell下执行.\\venv\\Scripts\\activate）。 ``` source venv/bin/activate ``` 4. 执行以下命令安装依赖。 ``` pip install -r requirements.txt ``` 6. 运行测试脚本。 1. 设置环境变量（如果在集群外访问，需要进行设置，可为agentcube-router的service配置ELB Ingress，通过ELB的弹性公网IP访问） - Linux环境执行： ``` export ROUTER_URL="http://addr:agentcube-router-port" ``` - Windows环境在powershell下执行： ``` $env:ROUTER_URL = "http://addr:agentcube-router-port" ``` 其中addr为弹性公网IP地址，agentcube-router-port为agentcube-router Service对应Ingress的端口。 2. 运行测试 ``` python test_agentruntime.py ``` 输出： ![](https://support.huaweicloud.com/usermanual-cce-autopilot/zh-cn_image_0000002569307747.png "点击放大") #### 版本记录 表3AgentCube插件版本记录 | 插件版本 | 支持的集群版本 | 更新特性 | |:---|:---|:---| | 1.0.3 | - v1.35 - v1.34 - v1.32 - v1.31 | - CCE Autopilot支持v1.35集群 - CCE Autopilot支持v1.34集群 | | 1.0.2 | - v1.31 - v1.32 | 新增AgentCube插件 |