概述
在AI Agent系統(tǒng)(比如我們的龍蝦系統(tǒng))中,安全的代碼執(zhí)行能力是核心組件之一。本文將介紹如何使用 OpenSandbox 構(gòu)建本地代碼執(zhí)行環(huán)境,并展示從基礎(chǔ)連接到高性能執(zhí)行的完整實(shí)現(xiàn)方案。
快速體驗(yàn):啟動和運(yùn)行效果
1. 啟動沙箱服務(wù)
使用 uvx 命令一鍵啟動 OpenSandbox 服務(wù)器:
uvx opensandbox-server
啟動成功后,終端將顯示服務(wù)就緒信息如下:
2. 基礎(chǔ)命令測試
首先測試基礎(chǔ)的 Linux 命令執(zhí)行能力。以下代碼調(diào)用 echo命令驗(yàn)證沙箱連通性:
sandbox.commands.run("echo '您好,郭秀志的本地Sandbox!'")
執(zhí)行結(jié)果如下,表明命令執(zhí)行正常:
3. Python 代碼異步執(zhí)行測試
接下來測試沙箱對 Python 代碼的支持,包括標(biāo)準(zhǔn)輸出、系統(tǒng)信息獲取及表達(dá)式求值:
# 測試1:標(biāo)準(zhǔn)打印
result1 = await executor.execute("print('Hello World')")
print(result1)
# 測試2:獲取Python版本
print("\n=== 執(zhí)行代碼 2 ===")
result2 = await executor.execute("import sys\nprint(sys.version)")
print(result2)
# 測試3:表達(dá)式計(jì)算與返回值
print("\n=== 執(zhí)行代碼 3 ===")
result3 = await executor.execute("result = 100 + 200\nresult")
print(result3)
最終執(zhí)行效果如下圖所示,沙箱正確返回了各項(xiàng)執(zhí)行結(jié)果:
一、OpenSandbox
OpenSandbox 是一個(gè)開源的沙箱系統(tǒng),提供了安全的代碼執(zhí)行環(huán)境。它支持:
- 容器隔離:基于 Docker 的安全執(zhí)行環(huán)境
- 多語言支持:Python、JavaScript、Java、Go 等
- API 接口:通過 HTTP API 進(jìn)行遠(yuǎn)程管理
- 靈活配置:可自定義鏡像、環(huán)境變量和入口點(diǎn)
二、基礎(chǔ)連接測試
在開始使用沙箱之前,首先需要驗(yàn)證服務(wù)連接的可用性。我們提供了 local_sandbox.py 作為連接測試工具。
2.1 連接測試實(shí)現(xiàn)
async def test_connection(domain: str):
"""測試與沙箱服務(wù)的連接"""
test_urls = [
f"{domain.rstrip('/')}/health"
]
for url in test_urls:
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
print(f"{'?' if response.status == 200 else '?'} {url} - 狀態(tài)碼: {response.status}")
except Exception as e:
print(f"? {url} - 錯(cuò)誤: {e}")
2.2 沙箱創(chuàng)建流程
config = ConnectionConfig(
domain="http://localhost:8080",
api_key="" # 可選:認(rèn)證密鑰
)
sandbox = await Sandbox.create(
image_name,
connection_config=config
)
async with sandbox:
# 執(zhí)行命令
execution = await sandbox.commands.run("echo 'Hello Sandbox!'")
# 獲取輸出
if execution.logs.stdout:
for line in execution.logs.stdout:
print(line.text)
# 清理資源
await sandbox.kill()
三、代碼執(zhí)行器
基礎(chǔ)實(shí)現(xiàn)雖然功能完整,但每次執(zhí)行都創(chuàng)建新沙箱會導(dǎo)致性能問題。codeExecutor.py 展示了如何通過沙箱復(fù)用實(shí)現(xiàn)高效執(zhí)行。
3.1 核心設(shè)計(jì)思想
FastCodeExecutor 類的核心優(yōu)勢:
- 沙箱復(fù)用:避免重復(fù)創(chuàng)建容器
- 上下文管理:為每種語言維護(hù)獨(dú)立執(zhí)行上下文
- 按需初始化:首次使用時(shí)創(chuàng)建,后續(xù)快速執(zhí)行
- 統(tǒng)一接口:簡潔的 execute() 方法
3.2 架構(gòu)設(shè)計(jì)
class FastCodeExecutor:
"""快速代碼執(zhí)行器 - 復(fù)用沙箱實(shí)例以加速執(zhí)行"""
def __init__(self):
self.config = ConnectionConfig(
domain="http://10.7.11.21:8080",
api_key="",
request_timeout=timedelta(seconds=60),
)
self.sandbox: Optional[Sandbox] = None
self.interpreter: Optional[CodeInterpreter] = None
self.contexts = {} # 按語言緩存上下文
3.3 初始化流程
async def initialize(self):
"""初始化沙箱(只需調(diào)用一次)"""
if self.sandbox is None:
self.sandbox = await Sandbox.create(
"sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/code-interpreter:v1.0.2",
connection_config=self.config,
entrypoint=["/opt/opensandbox/code-interpreter.sh"],
env={
"PYTHON_VERSION": "3.11",
"JAVA_VERSION": "17",
"NODE_VERSION": "20",
"GO_VERSION": "1.24",
},
)
await self.sandbox.__aenter__() # 激活上下文管理器
self.interpreter = await CodeInterpreter.create(sandbox=self.sandbox)
關(guān)鍵配置說明:
- 鏡像選擇:使用專用的 code-interpreter 鏡像,預(yù)裝多語言環(huán)境
-
入口點(diǎn):指定
/opt/opensandbox/code-interpreter.sh啟動代碼解釋器 - 環(huán)境變量:配置各語言版本,確保執(zhí)行環(huán)境一致
3.4 代碼執(zhí)行優(yōu)化
async def execute(self, code: str, language: SupportedLanguage = SupportedLanguage.PYTHON) -> str:
"""快速執(zhí)行代碼(復(fù)用已創(chuàng)建的沙箱)"""
if self.interpreter is None:
await self.initialize()
# 獲取或創(chuàng)建上下文(支持多語言)
if language not in self.contexts:
self.contexts[language] = await self.interpreter.codes.create_context(language)
result = await self.interpreter.codes.run(
code,
context=self.contexts[language],
)
# 合并標(biāo)準(zhǔn)輸出和返回值
outputs = []
# 獲取 print() 等標(biāo)準(zhǔn)輸出
if hasattr(result, 'logs') and result.logs and result.logs.stdout:
for log_entry in result.logs.stdout:
if hasattr(log_entry, 'text') and log_entry.text:
outputs.append(log_entry.text)
# 獲取表達(dá)式返回值
if result.result:
for res_entry in result.result:
if hasattr(res_entry, 'text') and res_entry.text:
outputs.append(res_entry.text)
return "\n".join(outputs) if outputs else ""
輸出處理策略:
-
標(biāo)準(zhǔn)輸出捕獲:收集
print()等 stdout 輸出 - 返回值提取:獲取最后一個(gè)表達(dá)式的計(jì)算結(jié)果
- 結(jié)果合并:將所有輸出按順序拼接
3.5 使用示例
async def main():
executor = FastCodeExecutor()
try:
# 首次初始化(較慢,只需一次)
await executor.initialize()
# 快速執(zhí)行多次代碼
result1 = await executor.execute("print('Hello World')")
result2 = await executor.execute("import sys\nprint(sys.version)")
result3 = await executor.execute("result = 100 + 200\nresult")
# 執(zhí)行復(fù)雜代碼
py_code = """
import math
def calculate_circle_area(radius):
return math.pi * radius ** 2
area = calculate_circle_area(5)
print(f"圓面積: {area:.2f}")
"""
result4 = await executor.execute(py_code)
finally:
await executor.cleanup()
四、性能對比
4.1 單次執(zhí)行 vs 復(fù)用執(zhí)行
| 指標(biāo) | 單次執(zhí)行(main_single) | 復(fù)用執(zhí)行(FastCodeExecutor) |
|---|---|---|
| 首次創(chuàng)建 | ~10-30秒 | ~10-30秒 |
| 單次執(zhí)行 | ~2-5秒 | ~0.1-0.5秒 |
| 10次執(zhí)行 | ~20-50秒 | ~1-5秒 |
| 資源消耗 | 每次創(chuàng)建新容器 | 復(fù)用單一容器 |
4.2 性能優(yōu)化關(guān)鍵點(diǎn)
- 避免重復(fù)創(chuàng)建:容器創(chuàng)建是最耗時(shí)的操作
- 上下文緩存:每種語言只創(chuàng)建一次執(zhí)行上下文
- 連接復(fù)用:保持與沙箱服務(wù)的持久連接
- 超時(shí)配置:合理設(shè)置請求超時(shí)(60秒)
五、最佳實(shí)踐
5.1 資源管理
# 使用 try-finally 確保資源釋放
async def main():
executor = FastCodeExecutor()
try:
await executor.initialize()
# ... 執(zhí)行代碼
finally:
await executor.cleanup() # 必須調(diào)用!
5.2 錯(cuò)誤處理
try:
result = await executor.execute(code)
except Exception as e:
# 記錄錯(cuò)誤但不清理沙箱(可繼續(xù)執(zhí)行)
print(f"執(zhí)行失敗: {e}")
# 如果是嚴(yán)重錯(cuò)誤,才清理并重新初始化
5.3 多語言支持
# Python
await executor.execute("print('Hello')", SupportedLanguage.PYTHON)
# JavaScript
await executor.execute("console.log('Hello')", SupportedLanguage.JAVASCRIPT)
# 自動創(chuàng)建對應(yīng)語言的上下文
5.4 環(huán)境變量配置
env={
"PYTHON_VERSION": "3.11", # 根據(jù)需求調(diào)整
"JAVA_VERSION": "17",
"NODE_VERSION": "20",
"GO_VERSION": "1.24",
}
六、應(yīng)用場景
6.1 AI Agent 代碼執(zhí)行
在 AI Agent 系統(tǒng)中,代碼執(zhí)行器可用于:
- 代碼生成驗(yàn)證:AI 生成的代碼即時(shí)運(yùn)行驗(yàn)證
- 數(shù)據(jù)分析:執(zhí)行數(shù)據(jù)分析和可視化代碼
- 算法測試:快速測試算法實(shí)現(xiàn)
- 教育場景:在線編程練習(xí)和評估
6.2 CI/CD 集成
- 自動化測試執(zhí)行
- 代碼質(zhì)量檢查
- 構(gòu)建驗(yàn)證
七、配置與部署
7.1 本地部署
# 啟動 OpenSandbox 服務(wù)
docker run -d -p 8080:8080 opensandbox/sandbox
# 驗(yàn)證服務(wù)
curl http://localhost:8080/health
7.2 遠(yuǎn)程部署
config = ConnectionConfig(
domain="http://your-server:8080",
api_key="your-api-key", # 生產(chǎn)環(huán)境必須配置
request_timeout=timedelta(seconds=60),
)
7.3 鏡像選擇
基礎(chǔ)測試:
image_name = "ubuntu" # 簡單命令執(zhí)行
代碼執(zhí)行:
image_name = "sandbox-registry.cn-zhangjiakou.cr.aliyuncs.com/opensandbox/code-interpreter:v1.0.2"
# 預(yù)裝 Python、Java、Node.js、Go 環(huán)境
八、總結(jié)
本文介紹了基于 OpenSandbox 構(gòu)建代碼執(zhí)行環(huán)境的兩種方案:
-
基礎(chǔ)方案(local_sandbox.py):
- 適合連接測試和簡單場景
- 每次創(chuàng)建新沙箱,資源隔離性好
- 執(zhí)行效率較低
-
優(yōu)化方案(codeExecutor.py):
- 適合頻繁代碼執(zhí)行場景
- 沙箱復(fù)用,性能提升 10-50 倍
- 支持多語言上下文管理
推薦使用 FastCodeExecutor 作為生產(chǎn)環(huán)境的代碼執(zhí)行方案,它在保證安全性的同時(shí),提供了優(yōu)異的執(zhí)行性能。