Skip to content

OpenAI Agents Example 源码级讲解

源码路径:CubeSandbox/examples/openai-agents-example/

概述

本例演示 CubeSandbox 在 SWE-bench 场景下的应用:在隔离沙箱中让 LLM Agent 自主修复真实开源项目的 bug。包含三个子 demo:

  1. pause_resume_demo.py — Sandbox 生命周期管理(暂停/恢复)
  2. simple_demo.py — 基础对话式 Agent
  3. main.py — SWE-bench Django 任务自动化

整体架构

Sandbox 生命周期 — 暂停/恢复

pause_resume_demo.py 展示了 MicroVM 的快照能力:

python
sandbox = Sandbox("base", timeout=300)
sandbox.write_file("test.txt", "hello")

# 暂停 → MicroVM 内存快照
sandbox.pause()
print(sandbox.is_alive())  # False

# 恢复 → 从快照还原
sandbox.resume()
print(sandbox.is_alive())  # True
print(sandbox.read_file("test.txt"))  # "hello" 保留

SWE-bench 任务执行流程

python
# 加载任务
tasks = [json.loads(line) for line in open("tasks.jsonl")]
task = tasks[0]  # Django bug

# 创建沙箱(预装 Django 环境)
sandbox = Sandbox(template="swe-bench-django")

# 写入仓库到 /tmp
sandbox.write_file(f"/tmp/{task['repo']}", repo_content)

# Agent 修复循环
for turn in range(max_turns):
    # 1. 构建 prompt(包含错误信息)
    prompt = f"Fix this bug: {task['problem_statement']}\nError: {error}"

    # 2. LLM 生成 patch
    response = llm.chat(prompt)
    patch = extract_patch(response)

    # 3. 应用 patch
    sandbox.write_file(patch_path, patch)

    # 4. 运行测试验证
    result = sandbox.process.start_and_wait(
        f"python -m pytest {task['test_file']}"
    )

    if result.exit_code == 0:
        break  # 修复成功

环境变量与 SSL 补丁

python
# 1. 代理设置
agent_env["http_proxy"] = "http://127.0.0.1:7890"
agent_env["REQUESTS_CA_BUNDLE"] = "/etc/ssl/certs/ca-certificates.crt"

# 2. SSL 证书注入
ca_cert = open("~/.local/share/mkcert/rootCA.pem").read()
sandbox.process.start_and_wait("echo '...' >> /etc/ssl/certs/ca-certificates.crt")

# 3. Sandbox.create 透传环境变量
sandbox = Sandbox(template=template, envs=agent_env)

这让沙箱内的 LLM API 调用走本地代理,解决网络访问和证书信任问题。

定时器管理

python
# 每 60 秒打印沙箱状态(监控任务)
timer = RepeatingTimer(60, print_sandbox_status, sandbox, end_time)
timer.start()

try:
    # 执行 Agent 任务
    agent.run(task)
finally:
    timer.cancel()  # 确保清理

推荐阅读

  1. simple_demo.py — 基础 Agent 对话流程
  2. pause_resume_demo.py — 暂停/恢复生命周期
  3. main.py — SWE-bench 完整流水线

源码参考:main.py