ACS Agent Sandbox 产品试用申请：https://survey.aliyun.com/apps/zhiliao/RMQSRd7yd
AgentScope与容器计算服务ACS深度集成，通过一键部署 AgentScope 至 ACS 集群，可以统一业务负载的运维体系，同时大幅降低对集群运维管理的复杂度。通过使用 ACS Serverless 容器算力，可以享受默认安全隔离、大规模弹性、普惠易用的特性，加速 AI Agent 应用的创新迭代。
背景介绍
自 ChatGPT 发布以来，大语言模型（Large Language Model，LLM）迅速从“辅助组件”升级为智能系统的核心引擎。系统架构的关注点正从“功能堆叠”转向“智能编排”：LLM 不再只是为应用提供回答，而是作为决策中枢，负责理解目标、拆解任务并驱动工具完成闭环执行。围绕这一能力，正在涌现一类新型智能系统——AI Agent（智能体），标志着人工智能从单纯的“工具增强”加速迈向“具备自主决策与行动能力的智能代理”，并开始在越来越多的真实业务场景中落地生根。
AgentScope和AgentScope Runtime
AgentScope是阿里巴巴推出的多智能体（Multi-Agent）应用框架与运行时，面向开发者和企业客户，帮助在大模型之上快速构建、编排和部署具备工具调用能力的智能代理应用。它提供多 Agent 协同、工具集成、对话状态管理、观测与调试等能力，并支持在云原生环境中以容器化方式运行。
过程透明、可观测：对话状态、消息传递、工具调用和模型交互全程可记录、可追踪，便于排查问题和审计行为。
可控执行：支持 ReAct 等模式，任务执行过程中可随时中断和人工干预，满足高可控场景需求。
工具与知识增强：统一工具管理与调用接口，原生支持 RAG，便于智能体利用企业知识和外部数据。
模型无关与多模态：通过适配层对接多家语言模型及多模态模型，降低模型切换与演进成本。
模块化与多智能体编排：将感知、推理、记忆、执行等环节模块化，支持“总控 + 执行 + 审核”等多智能体流水线编排。
工程化与生产可用：可对接日志、监控、告警、权限等基础设施，支撑从 PoC 到生产级部署的全生命周期。
在提供应用开发框架 AgentScope 之外，还推出了配套的运行时组件AgentScope Runtime，解决智能体在生产环境中的协议、部署、运行与治理问题，使基于 AgentScope 构建的应用能够以服务化方式稳定落地。同时，AgentScope Runtime 也可以与主流开源 Agent 框架或自研智能体实现无缝配合，其目标是为“服务化”的智能体应用提供可观测、可治理、易部署且安全可控的运行底座。
在企业内部，智能体应用一旦走向生产，就离不开稳定的运行环境和统一的运维体系。基于这一共识，AgentScope 与 AgentScope Runtime 在设计之初就面向Kubernetes 进行了优化，原生适配阿里云容器计算服务ACS。目前，将 AgentScope 部署在百炼应用平台、阿里云容器计算服务上已经成为了主流的部署模式，可以保障 Agent 应用的稳定安全运行、支持大规模并发弹性以及满足高性价比的成本诉求。
容器计算服务ACS相关能力
在 Agent Runtime 场景中，ACS每个 Pod 对应一个安全沙箱容器，简单易用。ACS 以 Serverless 方式交付容器算力，无需关注底层节点的运维管理，只需按需申请使用，用完即释放。
大规模弹性
ACS 容器算力结合用户负载特征的沙箱资源预调度优化，通过预测预调度、资源复用等手段，节省了 ACS 的调度、创建时间。同时，基于云盘快照技术提供镜像缓存功能。通过预先将需要使用的镜像制作成镜像缓存，然后基于该镜像缓存来大规模创建 Pod，避免或者减少镜像层的下载，从而进一步提升 Pod 的启动速度。
强安全隔离
使用 ACS 容器算力可以快速创建代码沙箱、浏览器沙箱、桌面沙箱环境。ACS 基于轻量级虚拟机（MicroVM）技术，为每个 Agent 任务提供独立的、硬件级别的计算安全隔离环境。基于 Network Policy 可以有效限制 Agent 之间的网络通信，防止恶意 Agent 访问敏感数据或服务。再结合云盘、NAS、OSS 等存储挂载隔离能力，ACS 实现了 Pod 级别计算、网络、存储的端到端安全运行环境。
普惠易用
ACS 支持 0.5 vCPU 1 GiB 精细化步长递进，同时支持秒级按需热变配，可根据 AI Agent 真实资源需求按需使用 ACS CPU/GPU Pod，降低综合资源成本。ACS 也邀测发布状态保持的能力，支持一键休眠和快速唤醒。在休眠期间，Pod 的 CPU 和内存资源释放，降低休眠期间的资源成本；同时，也可以基于用户请求快速唤醒 Pod，将内存状态保持，完整恢复到休眠前的环境，实现在成本与体验之间找到最佳平衡。
场景一：基于 AgentScope 构建智能体并部署至 ACS 集群
步骤一：准备 ACS 集群环境
创建ACS集群或者使用已有集群，获取集群kubeconfig并通过kubectl工具连接集群。
创建命名空间agentscope-runtime。
kubectl create namespace agentscope-runtime创建拉取智能体镜像使用的Secret。
docker-server：ACR实例地址，如registry-vpc.cn-wulanchabu.aliyuncs.com。docker-username：登录ACR实例的用户名，为阿里云账号全名。docker-password：登录ACR实例的用户名密码，为开通服务时设置的密码。可在访问凭证页面修改凭证密码。
kubectl create secret docker-registry demo-credential \ -n agentscope-runtime \ --docker-server=your-registry \ --docker-username=your-registry-username \ --docker-password=your-registry-password
步骤二：准备本地环境
下载或升级Python。AgentScope需要Python 3.10或更高版本。
下载 agentscope-runtime。
python3 -m venv demo1 source demo1/bin/activate pip install agentscope-runtime==1.0.1设置环境变量。
REGISTRY_URL：ACR实例地址，如registry-vpc.cn-wulanchabu.aliyuncs.com。REGISTRY_NAMESPACE：ACR实例命名空间。REGISTRY_USERNAME：登录ACR实例的用户名，为阿里云账号全名。REGISTRY_PASSWORD：登录ACR实例的用户名密码，为开通服务时设置的密码。可在访问凭证页面修改凭证密码。RUNTIME_SANDBOX_REGISTRY：ACR实例地址，如registry-vpc.cn-wulanchabu.aliyuncs.com。KUBECONFIG_PATH：KubeConfig的本地路径，默认路径为$HOME/.kube/config。DASHSCOPE_API_KEY：请开通阿里云百炼服务，并获取 API Key。
export REGISTRY_URL="your-acr-registry" export REGISTRY_NAMESPACE="your-registry-namespace" export REGISTRY_USERNAME="your-registry-username" export REGISTRY_PASSWORD="your-registry-password" export RUNTIME_SANDBOX_REGISTRY="your-acr-registry" export KUBECONFIG_PATH="/path-to-your-kubeconfig" export DASHSCOPE_API_KEY="your-api-key"安装Docker。用于在本地构建并上传镜像至ACR。
步骤三：构建 AI Agent 应用并部署
完整示例代码：
# -*- coding: utf-8 -*-
import asyncio
import time
import os
from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.tool import Toolkit, execute_python_code
from agentscope.pipeline import stream_printing_messages
from agentscope_runtime.adapters.agentscope.memory import (
AgentScopeSessionHistoryMemory,
)
from agentscope_runtime.engine.app import AgentApp
from agentscope_runtime.engine.deployers.kubernetes_deployer import (
KubernetesDeployManager,
RegistryConfig,
K8sConfig,
)
from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest
from agentscope_runtime.engine.services.agent_state import (
InMemoryStateService,
)
from agentscope_runtime.engine.services.session_history import (
InMemorySessionHistoryService,
)
agent_app = AgentApp(
app_name="Friday",
app_description="A helpful assistant",
)
@agent_app.init
async def init_func(self):
self.state_service = InMemoryStateService()
self.session_service = InMemorySessionHistoryService()
await self.state_service.start()
await self.session_service.start()
@agent_app.shutdown
async def shutdown_func(self):
await self.state_service.stop()
await self.session_service.stop()
@agent_app.query(framework="agentscope")
async def query_func(
self,
msgs,
request: AgentRequest = None,
**kwargs,
):
assert kwargs is not None, "kwargs is Required for query_func"
session_id = request.session_id
user_id = request.user_id
state = await self.state_service.export_state(
session_id=session_id,
user_id=user_id,
)
toolkit = Toolkit()
toolkit.register_tool_function(execute_python_code)
agent = ReActAgent(
name="Friday",
model=DashScopeChatModel(
"qwen-turbo",
api_key=os.getenv("DASHSCOPE_API_KEY"),
enable_thinking=True,
stream=True,
),
sys_prompt="You're a helpful assistant named Friday.",
toolkit=toolkit,
memory=AgentScopeSessionHistoryMemory(
service=self.session_service,
session_id=session_id,
user_id=user_id,
),
formatter=DashScopeChatFormatter(),
)
if state:
agent.load_state_dict(state)
async for msg, last in stream_printing_messages(
agents=[agent],
coroutine_task=agent(msgs),
):
yield msg, last
state = agent.state_dict()
await self.state_service.save_state(
user_id=user_id,
session_id=session_id,
state=state,
)
@agent_app.endpoint("/sync")
def sync_handler(request: AgentRequest):
yield {"status": "ok", "payload": request}
@agent_app.endpoint("/async")
async def async_handler(request: AgentRequest):
yield {"status": "ok", "payload": request}
@agent_app.endpoint("/stream_async")
async def stream_async_handler(request: AgentRequest):
for i in range(5):
yield f"async chunk {i}, with request payload {request}\n"
@agent_app.endpoint("/stream_sync")
def stream_sync_handler(request: AgentRequest):
for i in range(5):
yield f"sync chunk {i}, with request payload {request}\n"
@agent_app.task("/task", queue="celery1")
def task_handler(request: AgentRequest):
time.sleep(30)
yield {"status": "ok", "payload": request}
@agent_app.task("/atask")
async def atask_handler(request: AgentRequest):
await asyncio.sleep(15)
yield {"status": "ok", "payload": request}
# agent_app.run()
async def deploy_app_to_k8s():
"""Deploy AgentApp to Kubernetes"""
# 1. Configure Registry
registry_config = RegistryConfig(
registry_url=os.getenv("REGISTRY_URL"),
namespace=os.getenv("REGISTRY_NAMESPACE"),
username=os.getenv("REGISTRY_USERNAME"),
password=os.getenv("REGISTRY_PASSWORD"),
)
# 2. Configure K8s connection
k8s_config = K8sConfig(
k8s_namespace="agentscope-runtime",
kubeconfig_path=os.getenv("KUBECONFIG_PATH"),
)
port = 8080
# 3. Create KubernetesDeployManager
deployer = KubernetesDeployManager(
kube_config=k8s_config,
registry_config=registry_config,
use_deployment=True, # Use Deployment mode, supports scaling
)
# 4. Runtime configuration
runtime_config = {
# Resource limits
"resources": {
"requests": {"cpu": "200m", "memory": "512Mi"},
"limits": {"cpu": "1000m", "memory": "2Gi"},
},
# Image pull policy
"image_pull_policy": "IfNotPresent",
# Image Pull Secrets
"image_pull_secrets": ["demo-credential"],
}
# 5. Deployment configuration
deployment_config = {
# Basic configuration
"port": str(port),
"replicas": 1, # Deploy 1 replica
"image_tag": "linux-amd64-1",
"image_name": "agent_app",
# Dependencies configuration
"requirements": [
"agentscope",
"fastapi",
"uvicorn",
],
"extra_packages": [
os.path.join(
os.path.dirname(__file__),
"others",
"other_project.py",
),
],
"base_image": "python:3.10-slim-bookworm",
# Environment variables
"environment": {
"PYTHONPATH": "/app",
"LOG_LEVEL": "INFO",
"DASHSCOPE_API_KEY": os.environ.get("DASHSCOPE_API_KEY"),
},
# K8s runtime configuration
"runtime_config": runtime_config,
# Deployment timeout
"deploy_timeout": 300,
"health_check": True,
"platform": "linux/amd64",
"push_to_registry": True,
}
try:
print("Starting AgentApp deployment to Kubernetes...")
# 6. Execute deployment
result = await agent_app.deploy(
deployer,
**deployment_config,
)
print("Deployment successful!")
print(f"Deployment ID: {result['deploy_id']}")
print(f"Service URL: {result['url']}")
print(f"Resource name: {result['resource_name']}")
print(f"Replicas: {result['replicas']}")
# 7. Check deployment status
print("\nChecking deployment status...")
status = deployer.get_status()
print(f"Status: {status}")
return result, deployer
except Exception as e:
print(f"Deployment failed: {e}")
raise
async def deployed_service_run(service_url: str):
"""Test the deployed service"""
import aiohttp
test_request = {
"input": [
{
"role": "user",
"content": [{"type": "text", "text": "Hello, how are you?"}],
},
],
"session_id": "123",
}
try:
async with aiohttp.ClientSession() as session:
# Test sync endpoint
async with session.post(
f"{service_url}/sync",
json=test_request,
headers={"Content-Type": "application/json"},
) as response:
if response.status == 200:
result = await response.text()
print(f"Sync endpoint test successful: {result}")
else:
print(f"Sync endpoint test failed: {response.status}")
# Test async endpoint
async with session.post(
f"{service_url}/async",
json=test_request,
headers={"Content-Type": "application/json"},
) as response:
if response.status == 200:
result = await response.text()
print(f"Async endpoint test successful: {result}")
else:
print(f"Async endpoint test failed: {response.status}")
except Exception as e:
print(f"Service test exception: {e}")
async def main():
"""Main function"""
try:
# Deploy
result, deployer = await deploy_app_to_k8s()
service_url = result["url"]
# Test service
print("\nTesting the deployed service...")
await deployed_service_run(service_url)
# Keep running, you can test manually
print(
f"""
Service deployment completed, you can test with the following commands:
# Health check
curl {service_url}/health
# Test sync endpoint
curl -X POST {service_url}/sync \\
-H "Content-Type: application/json" \\
-d '{{
"input": [
{{
"role": "user",
"content": [
{{
"type": "text",
"text": "Hello, how are you?"
}}
]
}}
],
"session_id": "123"
}}'
# Test async endpoint
curl -X POST {service_url}/async \\
-H "Content-Type: application/json" \\
-d '{{
"input": [
{{
"role": "user",
"content": [
{{
"type": "text",
"text": "Hello, how are you?"
}}
]
}}
],
"session_id": "123"
}}'
# Test streaming endpoint
curl -X POST {service_url}/stream_async \\
-H "Content-Type: application/json" \\
-H "Accept: text/event-stream" \\
--no-buffer \\
-d '{{
"input": [
{{
"role": "user",
"content": [
{{
"type": "text",
"text": "Hello, how are you?"
}}
]
}}
],
"session_id": "123"
}}'
""",
)
print("\nOr use kubectl to check:")
print("kubectl get pods -n agentscope-runtime")
print("kubectl get svc -n agentscope-runtime")
print(
f"kubectl logs -l app={result['resource_name']} "
"-n agentscope-runtime",
)
# Wait for user confirmation before cleanup
input("\nPress Enter to cleanup deployment...")
# Cleanup deployment
print("Cleaning up deployment...")
cleanup_result = await deployer.stop()
if cleanup_result:
print("Cleanup completed")
else:
print("Cleanup failed, please check manually")
except Exception as e:
print(f"Error occurred during execution: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
# Run deployment
asyncio.run(main())
将完整示例代码保存为
main.py文件并运行。代码主要逻辑为镜像构建、上传至镜像仓库、部署到ACS集群，运行完成需约30分钟。
python main.py预期输出：
... Deployment successful! Deployment ID: 81550fdb-ee0c-4595-add2-c9041f90d43a Service URL: http://***.***.***.***:8080 Resource name: agent-81550fdb Replicas: 1 ...其中
Deployment successful!表示已完成部署，可在容器计算服务控制台中agentscope-runtime命名空间下查看创建的无状态工作负载Deployment。通过API请求Agent应用。
其中
your-service-url需替换为实际的公网IP地址。可通过kubectl get svc -n agentscope-runtime命令找到对应服务的EXTERNAL-IP或在容器计算服务控制台中agentscope-runtime命名空间下查看创建的服务Service。curl -N \ -X POST "http://your-service-url:8080/async" \ -H "Content-Type: application/json" \ -d '{ "input": [ { "role": "user", "content": [ { "type": "text", "text": "请帮我使用python生成一段冒泡排序算法，并执行" } ] } ] }'预期输出：
data: {"status": "ok", "payload": {"input": [{"sequence_number": null, "object": "message", "status": "created", "error": null, "id": "msg_3502bc4e-914b-46c2-ab84-f0bd1d886cbf", "type": "message", "role": "user", "content": [{"sequence_number": null, "object": "content", "status": null, "error": null, "type": "text", "index": null, "delta": false, "msg_id": null, "text": "请帮我使用python生成一段冒泡排序算法，并执行"}], "code": null, "message": null, "usage": null, "metadata": null}], "stream": true, "id": null, "model": null, "top_p": null, "temperature": null, "frequency_penalty": null, "presence_penalty": null, "max_tokens": null, "stop": null, "n": 1, "seed": null, "tools": null, "session_id": null, "user_id": null}}其中
"status": "ok"表示请求示例Agent应用成功。
场景二：使用ACS作为 Sandbox 的资源载体
AgentScope-Runtime 内置提供了几类Sandbox：Base、Browser 和 FileSystem。可将 Sandbox 部署至ACS集群中，使用Serverless容器算力。
步骤一：准备 ACS 集群环境
创建ACS集群或者使用已有集群，获取集群kubeconfig并通过kubectl工具连接集群。
创建命名空间agentscope-runtime。
kubectl create namespace agentscope-runtime
步骤二：准备本地环境
下载或升级Python。AgentScope需要Python 3.10或更高版本。
安装依赖。
安装gcc-c++和python3-devel。
# 以CentOS / RHEL / Fedora操作系统为例 yum install gcc-c++ python3-devel -y安装Rust和Cargo。
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
下载 agentscope-runtime。
python3 -m venv demo2 source demo2/bin/activate pip install agentscope-runtime==1.0.1 pip install agentscope-runtime[ext]==1.0.1若安装Rust相关组件失败，可尝试执行
rustup default nightly命令后再下载agentscope-runtime。安装完成需约30分钟。
步骤三：将社区提供的 Base Sandbox 部署至 ACS 集群
创建
custom.env文件，配置相关参数。# .env # 服务设置 HOST="0.0.0.0" PORT=8000 WORKERS=4 DEBUG=False BEARER_TOKEN=demo # 沙盒管理器设置 DEFAULT_SANDBOX_TYPE=base POOL_SIZE=10 AUTO_CLEANUP=True CONTAINER_PREFIX_KEY=agent-runtime-container- CONTAINER_DEPLOYMENT=k8s DEFAULT_MOUNT_DIR=sessions_mount_dir PORT_RANGE=[49152,59152] # Redis设置 REDIS_ENABLED=False # K8S 设置 K8S_NAMESPACE=agentscope-runtime KUBECONFIG_PATH=/path-to-your-kubeconfig # 替换为KubeConfig的本地路径，默认路径为$HOME/.kube/config使用 runtime-sandbox-server 将 BaseSandbox 部署至 ACS 集群。
runtime-sandbox-server --config custom.env预期输出：
可在容器计算服务控制台查看预热创建了 Sandbox Pod。（可选）将 Sandbox 作为一个插件注册到百炼中，进行业务编排。另外也可以使用高代码的方式基于agentscope-runtime 进行 Agent 应用研发，通过 SandboxService 结合会话来使用 Sandbox，如：
... sandbox_service = SandboxService() sandboxes = sandbox_service.connect( session_id=SESSION_ID, user_id=USER_ID, tools=tools, ) ...