Code Interpreter原理：沙箱执行、文件处理与多步推理

一、引言

ChatGPT Code Interpreter（现Advanced Data Analysis）是2023年最令人惊叹的AI能力之一：上传CSV→自动分析→生成图表→给出结论，全程无需人工干预。其背后是一个精妙的设计：LLM生成代码→沙箱执行→结果反馈→多步推理的闭环系统。

本文将拆解Code Interpreter的完整技术栈：沙箱安全、Python内核管理、文件系统交互、流式输出和多步推理。

二、架构总览

用户上传文件 + 提问
        │
        ▼
┌──────────────┐
│   LLM 推理    │ ← 理解意图，生成Python代码
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  沙箱执行器   │ ← Docker/K8s隔离环境
│  (Jupyter)   │ ← 执行代码，捕获输出/错误/图片
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  结果解析器   │ ← 解析stdout、matplotlib图表、DataFrame
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  反馈 LLM     │ ← 将结果注入上下文，决定下一步
└──────┬───────┘
       │  (如果未完成)
       ▼ 继续循环

三、沙箱执行引擎

3.1 最小化沙箱

import subprocess
import tempfile
import os
import signal
import resource
from typing import Tuple

class SandboxExecutor:
    """安全沙箱执行器"""
    
    def __init__(self, timeout=30, max_memory_mb=512):
        self.timeout = timeout
        self.max_memory_mb = max_memory_mb
        self.work_dir = tempfile.mkdtemp(prefix="sandbox_")
    
    def execute(self, code: str) -> Tuple[str, str, int]:
        """在沙箱中执行Python代码，返回(stdout, stderr, exit_code)"""
        try:
            proc = subprocess.Popen(
                ["python3", "-c", code],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=self.work_dir,
                preexec_fn=self._set_limits,  # Linux only
                text=True
            )
            
            try:
                stdout, stderr = proc.communicate(timeout=self.timeout)
                exit_code = proc.returncode
            except subprocess.TimeoutExpired:
                proc.kill()
                stdout, stderr = proc.communicate()
                return "", "Execution timeout", -1
            
            return stdout, stderr, exit_code
            
        except Exception as e:
            return "", str(e), -1
    
    def _set_limits(self):
        """设置资源限制"""
        # CPU时间限制
        resource.setrlimit(resource.RLIMIT_CPU, (self.timeout, self.timeout))
        # 内存限制
        mem_bytes = self.max_memory_mb * 1024 * 1024
        resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
        # 禁止fork子进程
        resource.setrlimit(resource.RLIMIT_NPROC, (1, 1))
    
    def write_file(self, name, content):
        """写入文件到工作目录"""
        path = os.path.join(self.work_dir, name)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, 'w') as f:
            f.write(content)
        return path
    
    def read_file(self, name):
        """读取工作目录中的文件"""
        path = os.path.join(self.work_dir, name)
        with open(path, 'r') as f:
            return f.read()
    
    def list_outputs(self):
        """列出所有输出文件"""
        outputs = []
        for root, dirs, files in os.walk(self.work_dir):
            for f in files:
                outputs.append(os.path.relpath(
                    os.path.join(root, f), self.work_dir
                ))
        return outputs
    
    def cleanup(self):
        import shutil
        shutil.rmtree(self.work_dir, ignore_errors=True)

3.2 Docker沙箱

import docker
import tempfile
import os

class DockerSandbox:
    """Docker容器沙箱（更强隔离）"""
    
    def __init__(self, image="python:3.11-slim"):
        self.client = docker.from_env()
        self.image = image
        self.work_dir = tempfile.mkdtemp()
    
    def execute(self, code: str, files: dict = None) -> dict:
        """在Docker容器中执行代码"""
        # 写入代码文件
        code_path = os.path.join(self.work_dir, "script.py")
        with open(code_path, 'w') as f:
            f.write(code)
        
        # 上传用户文件
        if files:
            for name, content in files.items():
                path = os.path.join(self.work_dir, name)
                if isinstance(content, bytes):
                    with open(path, 'wb') as f:
                        f.write(content)
                else:
                    with open(path, 'w') as f:
                        f.write(content)
        
        # 启动容器
        container = self.client.containers.run(
            self.image,
            command=f"timeout 30 python3 /workspace/script.py",
            volumes={self.work_dir: {"bind": "/workspace", "mode": "rw"}},
            mem_limit="512m",
            cpu_period=100000,
            cpu_quota=50000,  # 0.5 CPU
            network_disabled=True,  # 禁止网络
            remove=True,
            detach=True
        )
        
        result = container.wait()
        logs = container.logs(stdout=True, stderr=True).decode()
        
        # 收集输出文件
        outputs = {}
        for f in os.listdir(self.work_dir):
            fpath = os.path.join(self.work_dir, f)
            if f.endswith('.png') or f.endswith('.jpg'):
                with open(fpath, 'rb') as fp:
                    outputs[f] = fp.read()
        
        return {
            "stdout": logs,
            "exit_code": result["StatusCode"],
            "output_files": outputs
        }

四、Jupyter内核管理

import asyncio
from jupyter_client import AsyncKernelManager

class JupyterCodeExecutor:
    """基于Jupyter内核的代码执行器（支持变量持久化）"""
    
    def __init__(self):
        self.km = AsyncKernelManager()
        self.kc = None  # KernelClient
        self.execution_count = 0
    
    async def start(self):
        """启动内核"""
        await self.km.start_kernel()
        self.kc = self.km.client()
        self.kc.start_channels()
        await self.kc.wait_for_ready()
        print("Jupyter kernel ready")
    
    async def execute(self, code: str) -> dict:
        """执行代码并获取所有输出"""
        msg_id = self.kc.execute(code)
        
        outputs = {"text": [], "images": [], "error": None}
        
        while True:
            msg = await self.kc.get_iopub_msg(timeout=30)
            msg_type = msg["header"]["msg_type"]
            content = msg["content"]
            
            if msg_type == "stream":
                outputs["text"].append(content["text"])
            
            elif msg_type == "execute_result":
                if "text/plain" in content["data"]:
                    outputs["text"].append(content["data"]["text/plain"])
                if "image/png" in content["data"]:
                    outputs["images"].append(content["data"]["image/png"])
                if "text/html" in content["data"]:
                    outputs["text"].append(f"[HTML] {content['data']['text/html'][:200]}")
                # DataFrame 表格
                if "text/html" in content["data"] and "dataframe" in str(content["data"].get("text/html", "")):
                    outputs["text"].append("[DataFrame rendered]")
            
            elif msg_type == "display_data":
                if "image/png" in content["data"]:
                    outputs["images"].append(content["data"]["image/png"])
            
            elif msg_type == "error":
                outputs["error"] = "\n".join(content["traceback"])
                break
            
            elif msg_type == "status" and content["execution_state"] == "idle":
                break
        
        return outputs
    
    async def get_variable(self, name: str):
        """获取内核中的变量值"""
        result = await self.execute(f"print(repr({name}))")
        return result["text"][0] if result["text"] else None
    
    async def stop(self):
        """关闭内核"""
        if self.kc:
            self.kc.stop_channels()
        if self.km:
            await self.km.shutdown_kernel()

五、多步推理循环

class CodeInterpreterAgent:
    """完整的多步代码推理Agent"""
    
    SYSTEM_PROMPT = """你是一个数据分析助手，可以执行Python代码。

**规则：**
1. 每次只写一个代码块（用```python包裹）
2. 使用print()输出中间结果
3. 使用matplotlib生成图表时，先`plt.figure()`再`plt.show()`
4. 如果出错，分析错误并修正
5. 完成分析后，给出自然语言结论
6. 不要尝试访问网络、修改系统文件

**可用文件：** {files}
**当前工作目录：** /workspace
"""
    
    def __init__(self, llm, max_steps=10):
        self.llm = llm  # LLM客户端
        self.sandbox = SandboxExecutor()
        self.max_steps = max_steps
        self.conversation = []
    
    async def run(self, user_query: str, files: dict = None) -> dict:
        """运行多步推理循环"""
        # 上传文件到沙箱
        if files:
            for name, content in files.items():
                self.sandbox.write_file(name, content)
        
        # 初始化对话
        file_list = ", ".join(files.keys()) if files else "无"
        system_msg = self.SYSTEM_PROMPT.format(files=file_list)
        
        messages = [
            {"role": "system", "content": system_msg},
            {"role": "user", "content": user_query}
        ]
        
        steps = []
        final_answer = None
        
        for step in range(self.max_steps):
            # 1. LLM生成代码或最终回答
            response = self.llm.chat(messages)
            messages.append({"role": "assistant", "content": response})
            
            # 2. 解析响应：提取代码块
            code_blocks = self._extract_code_blocks(response)
            
            if not code_blocks:
                # 没有代码 → 可能是最终回答
                final_answer = response
                break
            
            # 3. 执行代码
            for code in code_blocks:
                stdout, stderr, exit_code = self.sandbox.execute(code)
                
                step_result = {
                    "step": len(steps) + 1,
                    "code": code[:200],
                    "stdout": stdout[:500],
                    "stderr": stderr[:200],
                    "exit_code": exit_code
                }
                steps.append(step_result)
                
                # 4. 构造执行结果反馈
                feedback = f"**执行结果 (exit_code={exit_code}):**\n"
                if stdout:
                    feedback += f"```\n{stdout[:1000]}\n```\n"
                if stderr:
                    feedback += f"**错误:** {stderr[:500]}\n"
                
                # 5. 检查输出文件（如图表）
                outputs = self.sandbox.list_outputs()
                new_images = [f for f in outputs if f.endswith('.png')]
                if new_images:
                    feedback += f"\n生成了图表: {', '.join(new_images)}"
                
                messages.append({"role": "system", "content": feedback})
        
        # 如果没有自然语言回答，让LLM总结
        if not final_answer:
            messages.append({
                "role": "user",
                "content": "请基于以上分析给出最终结论（不要写代码）。"
            })
            final_answer = self.llm.chat(messages)
        
        # 清理
        self.sandbox.cleanup()
        
        return {
            "final_answer": final_answer,
            "steps": steps,
            "total_steps": len(steps)
        }
    
    def _extract_code_blocks(self, text):
        import re
        # 匹配```python ... ```代码块
        pattern = r'```python\s*\n(.*?)```'
        matches = re.findall(pattern, text, re.DOTALL)
        return matches

六、文件处理

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from io import BytesIO
import base64

class FileProcessor:
    """文件处理器：解析上传文件 → DataFrame → 分析"""
    
    SUPPORTED_FORMATS = {
        '.csv': lambda p: pd.read_csv(p),
        '.xlsx': lambda p: pd.read_excel(p),
        '.json': lambda p: pd.read_json(p),
        '.parquet': lambda p: pd.read_parquet(p),
        '.txt': lambda p: open(p).read(),
        '.png': lambda p: plt.imread(p),
    }
    
    def __init__(self, max_rows=100000):
        self.max_rows = max_rows
        self.dataframes = {}
    
    def load(self, filename, content_bytes):
        """加载文件"""
        ext = os.path.splitext(filename)[1].lower()
        
        if ext in ['.csv', '.xlsx', '.json', '.parquet']:
            # 表格文件
            with tempfile.NamedTemporaryFile(suffix=ext) as tmp:
                tmp.write(content_bytes)
                tmp.flush()
                df = self.SUPPORTED_FORMATS[ext](tmp.name)
                
                # 采样（避免超大文件）
                if len(df) > self.max_rows:
                    df = df.sample(self.max_rows)
                
                name = os.path.splitext(filename)[0]
                self.dataframes[name] = df
                
                # 生成数据摘要
                summary = self._generate_summary(df, name)
                
                return {"type": "dataframe", "name": name,
                        "shape": df.shape, "summary": summary,
                        "dtypes": str(df.dtypes.to_dict())}
        
        elif ext in ['.png', '.jpg', '.jpeg']:
            # 图像文件（用matplotlib读取分析）
            img = plt.imread(BytesIO(content_bytes))
            return {"type": "image", "shape": img.shape}
        
        return {"type": "unknown"}
    
    def _generate_summary(self, df, name):
        """生成DataFrame摘要"""
        summary = f"**{name}** ({df.shape[0]}行 × {df.shape[1]}列)\n\n"
        summary += "**列信息:**\n"
        for col in df.columns:
            dtype = df[col].dtype
            null_pct = df[col].isnull().mean() * 100
            summary += f"- {col}: {dtype}"
            if null_pct > 0:
                summary += f" (缺失{null_pct:.1f}%)"
            if pd.api.types.is_numeric_dtype(dtype):
                summary += f" [范围: {df[col].min():.1f} ~ {df[col].max():.1f}]"
            summary += "\n"
        return summary

七、总结

Code Interpreter 的核心技术栈：

1. 沙箱执行 — Docker/Jupyter内核 + 资源限制
2. 多步推理 — LLM生成→执行→反馈的闭环
3. 文件处理 — 自动解析CSV/Excel/JSON/图像
4. 输出捕获 — stdout + matplotlib图表 + DataFrames
5. 安全防护 — 内存/CPU/网络/进程全面限制