OpenAI Agents Code Interpreter 源码级讲解
概述
本例是 CubeSandbox 与 LLM Agent 深度集成的典型案例：让 LLM 在隔离的 Python 沙箱中执行数据分析、绘图、科学计算。
提供两种执行后端：
- E2B 模式：
write script.py → exec python script.py（无状态） - Code Interpreter 模式：
run_code()via Jupyter kernel（有状态，变量跨轮保留）
整体架构
两种模式源码对比
E2B 模式（write + exec）
python
# 写入脚本文件
await session.write("script.py", generated_code)
# 执行 Python 解释器
result = await session.exec("python -I -B script.py")
# 手动捕获图表
if os.path.exists("output/chart.png"):
with open("output/chart.png", "rb") as f:
image_b64 = base64.b64encode(f.read())Code Interpreter 模式（run_code）
python
# 直接执行代码（Jupyter kernel）
result = await sandbox.run_code(generated_code)
# 图表自动从 Execution.results 解码
for r in result.results:
if r.png:
# 自动 base64 图片
display(Image(data=base64.b64decode(r.png)))
# 错误结构化回传
if result.error:
print(f"{result.error.name}: {result.error.value}")
print(result.error.traceback)执行流程对比
envd 兼容性补丁
源码中有一段重要的 monkey patch：
python
# envd 只支持 root 用户，E2B SDK 默认用 "user"
_e2b_rpc.default_username = "root"
# 强制所有文件操作用 root
for _name in ("read", "write", "write_files", ...):
_orig = getattr(_AsyncFS, _name)
def _wrapper(self, *a, **kw):
kw.setdefault("user", "root") # 强制 root
return await _orig(self, *a, **kw)
setattr(_AsyncFS, _name, _wrapper)流式输出实现
python
_stream_label: ContextVar[str] = ContextVar("_stream_label", default=None)
def _make_stream_handler(label, stream):
def _handler(data):
print(f"[{label}] {line}", file=stream, flush=True)
return _handler
# commands.run 自动注入流式回调
async def _patched_commands_run(self, *a, **kw):
label = _stream_label.get()
if label:
kw.setdefault("on_stdout", _make_stream_handler(label, sys.stdout))
return await _orig(self, *a, **kw)这让 LLM 生成的代码在执行时，输出实时流式显示在本地终端。
推荐阅读
code_interpreter_demo.py— E2B 模式 + Shell/PythonRunner Capabilitycode_interpreter_demo_ci.py— Code Interpreter 模式 + run_code- 理解 envd 兼容性补丁的必要性