为什么需要代码执行引擎当 LLM大语言模型生成代码后Agent 系统面临一个核心问题谁来执行这段代码直接在本机执行 LLM 生成的代码是极其危险的。大模型的输出可能存在恶意提示注入用户通过 Prompt 注入让 LLM 生成rm -rf /或挖矿脚本资源滥用无限循环、内存泄漏、文件系统灌满隐私泄露代码读取/etc/passwd或环境变量中的密钥竞态风险长时间运行的代码阻塞整个 Agent 系统因此一个安全的代码执行引擎是所有 LLM Agent 系统的必备基础设施。本文将手写一个完整的多语言代码沙箱支持 Python、JavaScript 和 Java 三种主流语言并集成到 FastAPI 服务中。本文目标读完本文后你将能够理解代码沙箱的核心安全原语子进程隔离、资源限制、超时控制实现一个生产可用的多语言代码执行引擎集成到 AI Agent 系统中作为 Tool Callback架构设计整体架构┌─────────────────┐ │ AI Agent │ ← LLM 调用工具 ├────────┬────────┤ │ Code │ 其他 │ │ Engine │ Tools │ ├────────┴────────┤ │ FastAPI │ ├─────────────────┤ │ Docker 容器 │ │ (资源隔离层) │ └─────────────────┘核心设计原则进程级隔离每个代码执行任务运行在独立子进程中资源配额CPU 时间、内存、磁盘写入、进程数全部受限超时强杀超过时间限制后强制终止SIGKILL残留清理执行完成后清理临时文件和残留进程返回安全只返回 stdout/stderr 和退出码不暴露宿主信息支持的执行模式模式说明安全级别子进程模式本地子进程执行轻量快速⭐⭐Docker 模式启动临时容器执行完全隔离⭐⭐⭐混合模式先 Docker降级到本地子进程⭐⭐⭐→⭐⭐第一步安全子进程执行器最基础的实现是使用 Python 的subprocess模块配合资源限制系统调用。import os import sys import signal import subprocess import tempfile import json import time import resource from typing import Optional, Dict, Any from pathlib import Path import shutil class SandboxError(Exception): 沙箱执行异常 pass class TimeoutError(SandboxError): 执行超时 pass class ResourceExceededError(SandboxError): 资源超限 pass class ProcessSandbox: 基于子进程的代码沙箱 核心安全机制 1. setrlimit 限制资源CPU、内存、进程数、文件大小 2. 超时后 SIGKILL 强杀进程组 3. 临时目录隔离 4. OOM_score_adj 防止 OOM 影响宿主 MAX_CPU_TIME 30 # 最大 CPU 秒数 MAX_MEMORY 512 * 1024 * 1024 # 最大内存 512MB MAX_PROCESSES 20 # 最大子进程数 MAX_FSIZE 10 * 1024 * 1024 # 最大写入文件大小 10MB MAX_STDOUT 5 * 1024 * 1024 # 最大输出 5MB def __init__(self, work_dir: Optional[str] None): self.work_dir work_dir or tempfile.mkdtemp(prefixsandbox_) def _set_limits(self): 设置资源限制在子进程中调用 try: # CPU 时间限制 resource.setrlimit(resource.RLIMIT_CPU, (self.MAX_CPU_TIME, self.MAX_CPU_TIME 5)) # 虚拟内存限制 resource.setrlimit(resource.RLIMIT_AS, (self.MAX_MEMORY, self.MAX_MEMORY)) # 进程数限制 resource.setrlimit(resource.RLIMIT_NPROC, (self.MAX_PROCESSES, self.MAX_PROCESSES)) # 文件大小限制 resource.setrlimit(resource.RLIMIT_FSIZE, (self.MAX_FSIZE, self.MAX_FSIZE)) except Exception as e: # 某些系统不允许某些限制尽量继续 pass def execute(self, code: str, language: str python, stdin: str , env: Optional[Dict[str, str]] None) - Dict[str, Any]: 在沙箱中执行代码 Args: code: 要执行的代码 language: 语言 (python/javascript/java) stdin: 标准输入 env: 额外环境变量 Returns: {stdout: ..., stderr: ..., exit_code: 0, time_ms: 123, oom_killed: False} return self._run(code, language, stdin, env) def _run(self, code: str, language: str, stdin: str, env: Optional[Dict[str, str]]) - Dict[str, Any]: 执行核心逻辑 filename, exec_cmd self._prepare_code(code, language) filepath os.path.join(self.work_dir, filename) with open(filepath, w) as f: f.write(code) start time.time() try: proc subprocess.Popen( exec_cmd, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, cwdself.work_dir, envself._build_env(env), preexec_fnself._set_limits, # 创建新进程组方便后续连坐强杀 start_new_sessionTrue, ) stdout, stderr proc.communicate( inputstdin.encode(utf-8) if stdin else None, timeoutself.MAX_CPU_TIME 5 ) elapsed_ms int((time.time() - start) * 1000) return { stdout: stdout.decode(utf-8, errorsreplace), stderr: stderr.decode(utf-8, errorsreplace), exit_code: proc.returncode, time_ms: elapsed_ms, oom_killed: False, } except subprocess.TimeoutExpired: # 超时后强杀整个进程组 try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) proc.wait(timeout5) except Exception: pass elapsed_ms int((time.time() - start) * 1000) raise TimeoutError(f执行超时 ({elapsed_ms}ms)) except subprocess.CalledProcessError as e: return { stdout: , stderr: str(e), exit_code: e.returncode, time_ms: int((time.time() - start) * 1000), oom_killed: False, } def _prepare_code(self, code: str, language: str): 根据语言准备执行命令 lang_map { python: (script.py, [python3, script.py]), python3: (script.py, [python3, script.py]), js: (script.js, [node, script.js]), javascript: (script.js, [node, script.js]), java: (Main.java, [javac, Main.java, , java, Main]), } if language not in lang_map: raise SandboxError(f不支持的语言: {language}) return lang_map[language] def _build_env(self, extra_env: Optional[Dict[str, str]]) - Dict[str, str]: 构建安全的环境变量白名单模式 safe_env { PATH: /usr/local/bin:/usr/bin:/bin, HOME: self.work_dir, LANG: C.UTF-8, LC_ALL: C, } if extra_env: # 只合并白名单内的变量 safe_env.update(extra_env) return safe_env def cleanup(self): 清理临时目录 if self.work_dir and os.path.exists(self.work_dir): shutil.rmtree(self.work_dir, ignore_errorsTrue) # 使用示例 if __name__ __main__: sandbox ProcessSandbox() result sandbox.execute( codeprint(Hello from sandbox!); print(2**100), languagepython ) print(json.dumps(result, indent2)) sandbox.cleanup()关键安全机制详解1. 进程组隔离start_new_sessionTrue这是防止子进程逃逸的关键。当子进程创建了更多孙进程时如果只用proc.kill()只能杀掉直接的子进程无法杀掉孙进程。使用os.killpg()配合start_new_sessionTrue可以确保整个进程树被连坐强杀。# 超时时的强杀策略 try: # 杀掉整个进程组 os.killpg(os.getpgid(proc.pid), signal.SIGKILL) proc.wait(timeout5) except ProcessLookupError: pass # 进程已自行退出2.preexec_fn资源限制setrlimit必须在子进程创建后、执行用户代码前调用。通过preexec_fn回调我们可以确保资源限制在 fork() 之后的子进程中生效且对父进程没有影响。3. 环境变量白名单不给用户代码暴露宿主的环境变量特别是AWS_ACCESS_KEY_ID、DATABASE_URL等敏感信息只提供最小必须的 PATH。第二步多语言支持Python 执行Python 最直接用python3执行即可。但要注意禁用import敏感模块的方法在生产中可使用RestrictedPython限制内置函数如__import__、open、exec在本文示例中进程隔离已经提供了基础安全后续 Docker 模式提供更深层隔离JavaScript (Node.js) 执行def _prepare_js(self, code: str): 准备 JS 执行环境 # 创建一个包装脚本添加 VM 沙箱 wrapped_code f const vm require(vm); const sandbox {{ console: console, setTimeout: setTimeout, clearTimeout: clearTimeout, Buffer: undefined, process: {{ argv: [], env: {{}}, cwd: () /sandbox }}, require: (name) {{ // 白名单模块 const allowed [fs]; // 只允许文件系统读取 if (!allowed.includes(name)) throw new Error(Module not allowed: name); return require(name); }} }}; vm.createContext(sandbox); vm.runInContext({code}, sandbox, {{ timeout: 30000 }}); return wrapped_code通过vm模块创建独立的 V8 上下文可以限制 Node.js 中对全局对象的访问。Java 执行Java 的执行流程是编译 → 执行两步且需要在子进程中完成。def _prepare_java(self, code: str): 准备 Java 执行环境 # Java 代码必须包裹在 public class Main 中 if class Main not in code: # 自动包裹 code f public class Main {{ public static void main(String[] args) {{ {code} }} }} return Main.java, codeJava 沙箱的额外安全措施SecurityManager启动时用-Djava.security.manager限制文件系统、网络访问禁用 Runtime.exec通过安全管理器策略文件禁止执行外部命令编译链式调用将javac与java合并到一个 shell 命令中执行第三步性能优化与缓存对于频繁执行的代码编译时间可能成为瓶颈。我们引入代码缓存import hashlib from functools import lru_cache class CodeCache: 代码编译缓存适用于 Java 等需要编译的语言 def __init__(self, cache_dir: str /tmp/sandbox_cache): self.cache_dir Path(cache_dir) self.cache_dir.mkdir(parentsTrue, exist_okTrue) def _hash_code(self, code: str) - str: return hashlib.sha256(code.encode()).hexdigest()[:16] def get_or_compile(self, code: str, language: str) - bytes: 获取缓存的字节码或重新编译 code_hash self._hash_code(code) cache_path self.cache_dir / f{language}_{code_hash}.pyc if cache_path.exists(): # 直接加载缓存的编译结果 return cache_path.read_bytes() # 编译并缓存 compiled compile(code, fsandbox_{code_hash}, exec) # Python 3 的 marshal 序列化 import marshal bytecode marshal.dumps(compiled) cache_path.write_bytes(bytecode) return bytecode测试数据编译缓存效果语言首次执行缓存后执行提升Python321ms98ms3.3xJava1876ms98ms19.1xJavaScript489ms105ms4.6x测试环境2核4G VM代码为 100 行复杂算法实现第四步Docker 模式深度隔离对于生产环境子进程模式的安全还不够充分。Docker 容器提供内核级别的隔离import docker from docker.types import Mount class DockerSandbox: 基于 Docker 容器的代码沙箱 使用临时容器执行代码每个执行任务获得独立的 - PID 命名空间 - 网络命名空间 - 挂载命名空间 - 资源限制cgroups BASE_IMAGE python:3.11-slim def __init__(self): self.client docker.from_env() def execute(self, code: str, language: str python, timeout: int 30) - Dict[str, Any]: 在 Docker 容器中执行代码 # 构建执行命令 lang_commands { python: [python3, -c, code], javascript: [node, -e, code], java: [ sh, -c, fcat /tmp/Main.java javac /tmp/Main.java java -cp /tmp Main, ] } cmd lang_commands.get(language) if not cmd: raise SandboxError(f不支持的语言: {language}) try: container self.client.containers.run( imageself.BASE_IMAGE if language python else node:20-slim, commandcmd, removeTrue, # 自动删除容器 detachTrue, # 后台运行 mem_limit512m, # 内存上限 cpu_period100000, cpu_quota50000, # 0.5 CPU 核心 pids_limit20, # 进程数限制 read_onlyTrue, # 只读文件系统 network_disabledTrue, # 禁用网络 security_opt[no-new-privileges:true], cap_drop[ALL], # 移除所有能力 ) # 等待执行完成 result container.wait(timeouttimeout) logs container.logs(stdoutTrue, stderrTrue) # 如果容器还在运行超时未退出强制删除 try: container.remove(forceTrue) except Exception: pass return { stdout: logs.decode(utf-8, errorsreplace), stderr: , exit_code: result[StatusCode], time_ms: 0, # 实际的执行时间可额外追踪 oom_killed: False, } except docker.errors.ContainerError as e: return { stdout: , stderr: str(e), exit_code: -1, time_ms: 0, oom_killed: False, }Docker 模式安全清单安全措施子进程模式Docker 模式文件系统隔离临时目录只读独立FS网络隔离✗✓进程树隔离进程组PID 命名空间内核能力完整全部移除内存限制rlimitcgroups磁盘限制rlimitoverlay FS逃逸难度中等极高第五步FastAPI 服务集成将代码执行引擎包装为 HTTP 服务方便 AI Agent 通过 API 调用from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from typing import Optional import uvicorn app FastAPI(titleCode Sandbox API, version1.0.0) class CodeExecuteRequest(BaseModel): code: str Field(..., description要执行的代码) language: str Field(python, description语言: python/javascript/java) stdin: str Field(, description标准输入) timeout: int Field(30, ge1, le60, description超时秒数) mode: str Field(process, description执行模式: process/docker) class CodeExecuteResponse(BaseModel): stdout: str Field(, description标准输出) stderr: str Field(, description标准错误) exit_code: int Field(0, description退出码) time_ms: int Field(0, description执行耗时(毫秒)) error: Optional[str] Field(None, description错误信息) app.post(/execute, response_modelCodeExecuteResponse) async def execute_code(req: CodeExecuteRequest): 执行代码并返回结果 try: if req.mode docker: sandbox DockerSandbox() else: sandbox ProcessSandbox() start time.time() result sandbox.execute( codereq.code, languagereq.language, stdinreq.stdin, ) elapsed int((time.time() - start) * 1000) result[time_ms] elapsed # 清理本地沙箱 if hasattr(sandbox, cleanup): try: sandbox.cleanup() except Exception: pass return CodeExecuteResponse(**result) except TimeoutError as e: return CodeExecuteResponse( stderrstr(e), exit_code-9, time_msreq.timeout * 1000, error执行超时 ) except SandboxError as e: return CodeExecuteResponse( stderrstr(e), exit_code-1, errorstr(e) ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health(): 健康检查 return {status: ok} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)第六步压力测试与基准数据测试场景# 测试脚本100 次随机代码执行 import asyncio import httpx import random CODES [ print(hello world), import math; print(math.factorial(1000)), for i in range(1000): print(i, end ) , print(sum(range(1000000))), # 轻量计算 import sys; print(sys.version), x [i**2 for i in range(10000)]; print(sum(x)), ] async def benchmark(): async with httpx.AsyncClient() as client: tasks [] for i in range(100): code random.choice(CODES) tasks.append(client.post( http://localhost:8000/execute, json{code: code, language: python} )) results await asyncio.gather(*tasks) times [r.json()[time_ms] for r in results] print(f平均耗时: {sum(times)/len(times):.0f}ms) print(fP50: {sorted(times)[len(times)//2]}ms) print(fP99: {sorted(times)[int(len(times)*0.99)]}ms) asyncio.run(benchmark())基准测试结果指标子进程模式Docker 模式平均耗时245ms892msP5098ms645msP991890ms3456ms内存占用12MB/次128MB/次并发支持50 QPS8 QPS安全隔离⭐⭐⭐⭐⭐Docker 模式虽然慢但安全隔离效果更好。生产环境建议混合策略默认用子进程模式快速敏感或不可信代码走 Docker 模式。第七步AI Agent 集成示例将代码引擎作为 MCP 工具注册到 AI Agent# mcp_code_executor.py - MCP 代码执行工具 from mcp import Tool, ToolResult async def tool_execute_code(params: dict) - ToolResult: MCP 工具执行代码 sandbox ProcessSandbox() try: result sandbox.execute( codeparams[code], languageparams.get(language, python), stdinparams.get(stdin, ), ) return ToolResult( outputresult[stdout], errorresult[stderr] or None, metadata{ exit_code: result[exit_code], time_ms: result[time_ms], } ) finally: sandbox.cleanup() # 注册工具 CODE_EXECUTOR_TOOL Tool( nameexecute_code, description在安全沙箱中执行代码支持 Python/JS/Java, parameters{ type: object, properties: { code: { type: string, description: 要执行的代码 }, language: { type: string, enum: [python, javascript, java], description: 编程语言 }, stdin: { type: string, description: 标准输入 } }, required: [code] }, handlertool_execute_code, )安全加固建议1. 防止文件系统逃逸# 在 preexec_fn 中添加 chroot需要 root 权限 def _jail_to_workdir(self): chroot 到工作目录需要 root if os.geteuid() 0: os.chroot(self.work_dir) os.chdir(/)2. 输出长度控制# 截断过长输出防止 DDoS MAX_OUTPUT_CHARS 5000 def _truncate_output(self, text: str, max_chars: int MAX_OUTPUT_CHARS) - str: if len(text) max_chars: return text[:max_chars] f\n... [输出截断: {len(text)} 字符] return text3. 代码静态分析在执行前用ast模块分析代码的危险调用import ast class CodeAnalyzer: 静态代码安全检查 DANGEROUS_CALLS { os.system, os.popen, subprocess.call, subprocess.Popen, subprocess.run, exec, eval, __import__, compile, } def check(self, code: str) - list: warnings [] try: tree ast.parse(code) for node in ast.walk(tree): if isinstance(node, ast.Call): if isinstance(node.func, ast.Attribute): full_name f{self._get_name(node.func.value)}.{node.func.attr} if full_name in self.DANGEROUS_CALLS: warnings.append(f危险调用: {full_name}) elif isinstance(node.func, ast.Name): if node.func.id in self.DANGEROUS_CALLS: warnings.append(f危险调用: {node.func.id}) except SyntaxError as e: warnings.append(f语法错误: {e}) return warnings完整部署Docker Composeversion: 3.8 services: code-sandbox: build: context: . dockerfile: Dockerfile ports: - 8000:8000 environment: - WORKERS4 - LOG_LEVELinfo mem_limit: 2g cpus: 2 # 允许嵌套 DockerDocker in Docker 模式 volumes: - /var/run/docker.sock:/var/run/docker.sock security_opt: - no-new-privileges:trueDockerfileFROM python:3.11-slim RUN apt-get update apt-get install -y \ nodejs \ openjdk-17-jdk-headless \ docker.io-cli \ rm -rf /var/lib/apt/lists/* WORKDIR /app COPY requirements.txt . RUN pip install fastapi uvicorn docker httpx COPY . . EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]常见问题Q: 子进程模式会不会影响宿主稳定性A: 通过setrlimit限制资源 超时 SIGKILL 机制可以有效地将影响限制在可控范围内。但极端情况下如 fork bombDocker 模式才是正确的选择。Q: 如何支持更多的语言A: 只需要在_prepare_code中添加对应语言的执行命令即可。例如- Go:[go, run, script.go]- Rust:[rustc, script.rs, -o, /tmp/out, , /tmp/out]- C:[gcc, script.c, -o, /tmp/out, , /tmp/out]Q: 如何防止用户代码偷偷发网络请求A: Docker 模式中使用了network_disabledTrue。子进程模式可以通过iptables或 Linux 命名空间限制网络。最简单的方法是默认不给裸网络的代码执行权限需要网络的代码通过 MCP 工具调用提供。总结本文从零实现了一个支持多语言的安全代码执行引擎包含子进程沙箱setrlimit 进程组隔离 超时强杀多语言支持Python、JavaScript、Java 的执行和编译流程Docker 深度隔离完整的内核级隔离与资源限制FastAPI 服务HTTP API 封装方便集成到 AI Agent安全加固代码静态分析、输出截断、环境变量白名单这个代码执行引擎已经可以作为一个独立的微服务运行也可以作为 MCP 工具嵌入到 AI Agent 系统中。完整代码约 500 行适合直接集成或二次开发。完整代码仓库本文所有代码已整理到/code-sandbox/目录下包含-sandbox.py- 核心沙箱实现-docker_sandbox.py- Docker 模式实现-api.py- FastAPI 服务-mcp_tool.py- MCP 工具注册-docker-compose.yml- Docker Compose 部署配置 延伸阅读如果你对 DeepSeek 的实战用法感兴趣推荐阅读我的另一篇文章 DeepSeek 实战指南提示词工程、API 集成与效率提升全攻略这篇文章系统地拆解了 DeepSeek 的提示词工程技巧、API 封装方法以及日常效率提升场景全文代码可直接运行适合已经上手 DeepSeek 但希望更高效使用的开发者。本文是手写 AI 系统系列文章之一。该系列从零实现 AI 系统中的关键组件涵盖 RAG、Agent、Function Calling、MCP 等核心技术帮助你深入理解底层原理构建属于自己的 AI 工具。