AgentCube插件

apiVersion: runtime.agentcube.volcano.sh/v1alpha1
kind: CodeInterpreter
metadata:
  name: my-codeinterpreter
  namespace: default
spec:
  template:
    # runtimeClassName: kata    # 若要使用安全容器，则可配置runtimeClassName为kata或kuasar-vmm（需要有支持安全运行时的节点）
    # 使用 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（近零延迟）

kubectl apply -f code-interpreter-demo.yaml

创建完成后，使用以下命令查看状态。

kubectl get codeinterpreter my-codeinterpreter -o yaml

# 查看创建的 SandboxWarmPool
kubectl get sandboxwarmpool -n default
# 查看预热池的 Pod
kubectl get pods |grep my-codeinterpreter

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

1. 设置环境变量（如果在集群外访问，需要进行设置，可将workloadmanager和agentcube-router的service配置为NodePort类型，并给CCE集群节点绑定弹性公网IP，通过弹性公网IP+NodePort访问）
   • Linux环境执行：

     export WORKLOAD_MANAGER_URL="http://addr:workloadmanager-service-nodeport" 
     export ROUTER_URL="http://addr:agentcube-router-service-nodeport"

   • Windows环境在powershell下执行：

     $env:ROUTER_URL = "http://addr:agentcube-router-service-nodeport"
     $env:WORKLOAD_MANAGER_URL = "http://addr:workloadmanager-service-nodeport"

   其中addr为弹性公网IP地址，agentcube-router-service-nodeport为agentcube-router Service的NodePort端口，workloadmanager-service-nodeport为workloadmanager Service的NodePort端口

2. 运行测试

   python test_codeinterpreter.py

   输出：

   

docker build  -t my-agent-app:latest .

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:
      # runtimeClassName: kata    # 若要使用安全容器，则可配置runtimeClassName为kata或kuasar-vmm（需要有支持安全运行时的节点）
      schedulerName: default-scheduler    # 如果开启安装了volcano agent-scheduler调度器，可配置为agent-scheduler
      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: {}

kubectl apply -f agent-runtime-demo.yaml
#  查看状态
kubectl get agentruntime my-agent-app -oyaml

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

1. 设置环境变量（如果在集群外访问，需要进行设置，可将agentcube-router的service配置为NodePort类型，并给CCE集群节点绑定弹性公网IP，通过弹性公网IP+NodePort访问）
   • Linux环境执行：

     export ROUTER_URL="http://addr:agentcube-router-service-nodeport"

   • Windows环境在powershell下执行：

     $env:ROUTER_URL = "http://addr:agentcube-router-service-nodeport"

   其中addr为弹性公网IP地址，agentcube-router-service-nodeport为agentcube-router Service的NodePort端口。

2. 运行测试

   python test_agentruntime.py

   输出：