はじめに
今回のAKARI Tech Blogは、DX Solution事業本部 VPoEの丸尾 (@marumaru_akari) が担当します！
データ分析などで、LLM に任意 Python を実行させたい場面は増えていますが、ホスト上で動かすのは怖いものです。CoreWeave が公開した Sandbox はこの問題への解として位置づけられています。本記事では Sandbox の概要、Weave トレースを使った実行履歴の可視化と実例を扱います。
- はじめに
- 1. W&B Serverless Sandbox とは
- 2. 最小サンプルと Weave トレース
- 3. OpenAI tool use と Sandbox の組み合わせ
- 4. 危険コードを安全に実行する
- 5. CSV をアップロードして分析・グラフを生成する
- まとめ
- We're hiring!
1. W&B Serverless Sandbox とは
この記事では、CoreWeave Sandboxes の serverless 提供形態である W&B Serverless Sandboxes を扱います。
- 隔離方式: 各Sandboxは Kubernetes Pod または Kata Containers で動き、互いに独立しています
- デプロイ: W&B 経由の serverless と、CoreWeave CKS(マネージドクラスタ) の2モードがあります。前者なら
wandb loginだけで使えます - 想定ユースケース: 強化学習の大規模実行、AI エージェントの tool use、モデル評価
この記事ではAIエージェントの tool use によるPythonコード実行のユースケースを想定します。
2. 最小サンプルと Weave トレース
インストールと認証はこの2行で終わります。
uv add 'wandb[sandbox]' uv run wandb login
最小サンプルは次のとおりです。Sandbox の起動・コード実行と、Weave への自動記録を1関数にまとめています。
import weave from wandb.sandbox import Sandbox weave.init("agent-sandbox-demo") @weave.op() def execute_agent_tool(python_code: str) -> dict: with Sandbox.run() as sandbox: sandbox.wait() sandbox.write_file("/tmp/script.py", python_code.encode("utf-8")).result() proc = sandbox.exec(["python", "/tmp/script.py"]).result() return { "exit_code": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, }
with Sandbox.run() as sandbox で隔離環境が立ち上がり、sandbox.exec([...]).result() でコマンドを実行して結果を回収します。さらに @weave.op() でデコレートしておくだけで、引数 python_code(LLM が生成したコード)と戻り値の dict(exit_code / stdout / stderr)が Weave のトレースに自動で記録されます。
最初に踏みやすい注意点を2つだけ触れておきます。WANDB_ENTITY に非 ASCII 値を入れると x-wandb-entity ヘッダーで弾かれるため、ASCII セーフな値を import 前に os.environ.setdefault で明示しておきます。また、write_file / read_file は OperationRef を返す非同期 API なので、.result() を呼ばないと書き込み完了前に exec が走って ENOENT で落ちます。
実行後、コンソールに表示される Weave URL を開くと、入力した Python コードと出力がトレース 1 件として並んで記録されているはずです。
「LLM が生成したコード」と「Sandboxで実際に起きた結果」を1つのレコードで紐づけて残せるので、エージェント運用でのデバッグや監査に効きます。
3. OpenAI tool use と Sandbox の組み合わせ
LLM が「実行して確かめた方が早い」問題を自分で解けるようになると、応答精度が一気に上がります。とはいえ任意 Python 実行を許すのは普通は地雷です。Sandbox に閉じ込めた execute_python ツールを OpenAI のチャットモデルに渡せば、隔離前提で安全に「実行→結果踏まえた応答」のループが回せます。
ツール定義は OpenAI Function Calling 仕様にそのまま乗ります。
TOOLS = [
{
"type": "function",
"function": {
"name": "execute_python",
"description": "Python 3 のコードを隔離サンドボックスで実行し、stdout / stderr / exit_code を返す。",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "実行する Python コード。結果は print() で出力すること。",
},
},
"required": ["code"],
},
},
},
]
スキーマはミニマルで、モデルは code という文字列 1 つを渡してくれれば良いだけです。実行本体はセクション2 で見たパターンに @weave.op() を被せただけです。
@weave.op() def execute_python(code: str) -> dict: with Sandbox.run() as sandbox: sandbox.wait() sandbox.write_file("/tmp/agent_script.py", code.encode("utf-8")).result() proc = sandbox.exec(["python", "/tmp/agent_script.py"]).result() return { "exit_code": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, }
返り値は {exit_code, stdout, stderr} の dict のみ。モデルには事実だけ渡して、解釈は LLM 側に委ねます。
OpenAI から返ってきた tool_call を実体に振り分けるのが dispatch_tool_call です。
@weave.op() def dispatch_tool_call(name: str, arguments_json: str) -> str: args = json.loads(arguments_json) if name == "execute_python": result = execute_python(args["code"]) else: result = {"error": f"unknown tool: {name}"} return json.dumps(result, ensure_ascii=False)
そしてこれを chat_turn でループに乗せます。前半は OpenAI への呼び出しと応答の積み込みです。
@weave.op() def chat_turn(client: OpenAI, messages: list[dict]) -> list[dict]: for _ in range(MAX_TOOL_LOOPS): resp = client.chat.completions.create( model=MODEL, messages=messages, tools=TOOLS, ) msg = resp.choices[0].message # SDK の ChatCompletionMessage はそのまま再送できないので dict 形式に正規化 assistant_msg: dict = {"role": "assistant", "content": msg.content} if msg.tool_calls: assistant_msg["tool_calls"] = [ {"id": tc.id, "type": "function", "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in msg.tool_calls ] messages.append(assistant_msg)
ここで ChatCompletionMessage をそのままでは履歴に積めないという地味なハマりどころがあり、tool_calls 込みで dict に詰め直してから messages に append しています。後半は tool 実行と次ターンへの繰り戻しです。
if not msg.tool_calls: return messages for tc in msg.tool_calls: tool_result = dispatch_tool_call(tc.function.name, tc.function.arguments) messages.append({ "role": "tool", "tool_call_id": tc.id, "content": tool_result, }) raise RuntimeError(f"tool_call ループが {MAX_TOOL_LOOPS} 回上限に達した")
モデルが tool_call を返さなくなったらそのターンは終了します。tool_call が連鎖する(実行結果を見て次の tool_call を出す)ことも普通にあるので、MAX_TOOL_LOOPS=8 の安全弁で暴走を止めています。
chat_turn → dispatch_tool_call → execute_python の3層に @weave.op() をそれぞれ被せておくと、Weave UI ではトレースが入れ子で記録されます。1 ユーザー発話に対する「LLM 呼び出し → tool_call → Sandbox実行 → LLM 再呼び出し」の流れが、1 グラフに親子構造でぶら下がる形です。セクション2 で見た「1 op の可視化」が階層構造に発展した状態です。
実際に動かすと次のような往復になります。
You: フィボナッチ数列の第１００項を計算して
----- [tool: execute_python] generated code -----
def fibonacci(n):
if n <= 0:
return 0
elif n == 1:
return 1
else:
a, b = 0, 1
for _ in range(2, n + 1):
a, b = b, a + b
return b
result = fibonacci(100)
print(result)
----- stdout -----
354224848179261915075
モデルが必要だと判断したら自動で execute_python を呼び、Sandbox 内で計算した結果を踏まえてユーザーに応答します。
4. 危険コードを安全に実行する
セクション3 で組み立てた agent loop に「ホスト上では走らせたくない依頼」を投げると、Sandbox の隔離性が一気に体感できます。ここでは2つのプロンプトで観察します。1つは /etc の中身を覗くだけの読み取り、もう1つは /etc/.pwd.lock を削除する破壊系の依頼です。
その前に1点だけ補足します。gpt-4o-mini は素のままだと、破壊的な依頼は安全のために拒否してしまう場合があります。今回は隔離デモが目的なので、SYSTEM_PROMPT を「ここはSandboxで隔離されておりホストには影響しないので、拒否せず素直に実行してください」というニュアンスに調整して、モデルが依頼を受け入れやすい状態にしてあります。
デモ1: /etc の中身を覗く
You: /etc 以下に含まれるファイルを出力して と入れると、モデルは os.listdir("/etc") 相当のコードをツール経由で実行します。
[tool: execute_python] generated code:
import os
for name in sorted(os.listdir("/etc")):
print(name)
[tool: execute_python] exit_code=0
stdout:
.pwd.lock
alternatives
...
ここに並んでいるのは Sandbox コンテナ内の /etc であって、ホストマシンの /etc ではありません。LLM が好奇心で /etc 配下を全部覗いて回ったとしても、見えるのは使い捨てコンテナの中身だけで、ホスト機の設定や秘密情報が漏れることはありません。
デモ2: /etc/.pwd.lock を削除する
次は明確に破壊系です。You: /etc/.pwd.lock を削除して と頼むと、モデルは os.unlink("/etc/.pwd.lock") を実行します。
[tool: execute_python] generated code:
import os
os.unlink("/etc/.pwd.lock")
print("removed; exists =", os.path.exists("/etc/.pwd.lock"))
[tool: execute_python] exit_code=0
stdout: removed; exists = False
Sandbox の中ではあっさり消えました。とはいえこれが意味するのは「コンテナ内の /etc/.pwd.lock が消えた」だけで、ホストマシン側の /etc/.pwd.lock は手付かずのまま残ります。さらに、Sandbox は with Sandbox.run() as sandbox: のブロックを抜けるとコンテナごと破棄されるため、削除した状態も次の起動時には初期状態に戻ります。何度試しても永続的な汚染は起きません。
LLM がエージェントとして提案してくるコードを「実際に動かしたら何が起きるか」を観察したいとき、ホストに一切影響を出さずに試せる土台があるのは大きな安心材料です。
5. CSV をアップロードして分析・グラフを生成する
最後に、ここまで作った agent loop を実用ユースケースに寄せます。ユーザーが渡した CSV をSandboxに上げ、pandas での分析と matplotlib でのグラフ生成までモデルに任せる構成です。生成された PNG はホスト側に自動で吸い上げます。
ここで初めて、これまで「ツール呼び出しごとに使い捨てていた」Sandbox を REPL 全体で 1 つ使い回し に変えます。pip install pandas matplotlib と CSV のアップロードを毎ターン繰り返すと、ユーザー入力ごとに数十秒のロスが出てしまうためです。起動時にまとめて準備します。
def setup_sandbox(csv_path: Path) -> Sandbox: sandbox = Sandbox.run() sandbox.wait() proc = sandbox.exec(["pip", "install", "-q", *PIP_PACKAGES]).result() if proc.returncode != 0: raise RuntimeError(f"pip install failed:\n{proc.stderr}") sandbox.write_file(SANDBOX_CSV_PATH, csv_path.read_bytes()).result() sandbox.exec(["mkdir", "-p", SANDBOX_OUTPUT_DIR]).result() return sandbox def run_repl(csv_path: Path) -> None: client = OpenAI() sandbox = setup_sandbox(csv_path) try: execute_python_fn = make_execute_python(sandbox) dispatch_tool_call = make_dispatcher(execute_python_fn) messages: list[dict] = [{"role": "system", "content": SYSTEM_PROMPT}] print(f"CSV分析REPL ({csv_path.name} ready, 'exit' で終了)") while True: try: user_input = input("\nYou: ").strip() except (EOFError, KeyboardInterrupt): print() break if not user_input: continue if user_input.lower() in {"exit", "quit"}: break messages.append({"role": "user", "content": user_input}) messages = chat_turn(client, messages, dispatch_tool_call) print(f"\nAssistant: {messages[-1].get('content') or '(no content)'}") finally: sandbox.stop(missing_ok=True).result() print("\n[teardown] sandbox stopped.")
Sandbox.run() で立ち上げ、pandas/matplotlib を入れ、CSV を /tmp/data.csv へ転送、成果物用に /tmp/output/ を作っておくだけです。REPL のあいだはこの同じ sandbox を使い回します。
次は、ツール本体に sandbox を渡す仕組みです。OpenAI tool の関数シグネチャは (code: str) -> dict 固定なので、sandbox を引数で渡せません。そこでクロージャで捕捉します。
def make_execute_python(sandbox: Sandbox): @weave.op() def execute_python(code: str) -> dict: before = list_sandbox_outputs(sandbox) sandbox.write_file("/tmp/script.py", code.encode("utf-8")).result() proc = sandbox.exec(["python", "/tmp/script.py"]).result() after = list_sandbox_outputs(sandbox) new_files = sorted(after - before) HOST_OUTPUT_DIR.mkdir(parents=True, exist_ok=True) saved: list[str] = [] for name in new_files: data = sandbox.read_file(f"{SANDBOX_OUTPUT_DIR}/{name}").result() host_path = HOST_OUTPUT_DIR / name host_path.write_bytes(data) saved.append(str(host_path)) return { "exit_code": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, "saved_files": saved, } return execute_python
外側の make_execute_python で sandbox を捕捉し、内側の execute_python がそれを参照しながら毎ツールコールでコードを実行します。実行前後で /tmp/output/ をリストアップして差分を取り、新しく生まれたファイルだけホストの ./output/ にコピーします。モデル側は「ファイルに保存してください」を意識せず、ただ /tmp/output/ に savefig するだけで済みます。
実際に動かすと次のような流れになります。
$ uv run python csv_analyzer.py samples/sales.csv You: Region 別の平均売上を棒グラフで Assistant: Region 別の平均売上（revenue）の計算が完了しました。以下がその結果です： | Region | Average Revenue | |----------|------------------| | Fukuoka | 20425.40 | | Nagoya | 22091.55 | | Osaka | 16261.55 | | Sapporo | 28022.58 | | Tokyo | 20758.18 | 平均売上を示す棒グラフを作成し、次のファイルに保存しました： このグラフは `/tmp/output/average_revenue_by_region.png` に保存されました。
sandbox内で描画したグラフをホスト側にコピーすることに成功しました。
まとめ
ここまでで、CoreWeave Sandboxes をWeave で実行履歴を残しながら、OpenAI tool use と組み合わせて任意 Python を走らせ、危険コードを隔離下で観察し、CSV 分析エージェントまで作りました。Sandbox の隔離性と Weave のトレースを組み合わせれば、エージェントが書いたコードを安全に動かしつつ、すべての入出力を後追いで監査できる状態が手に入ります。Public Preview 段階ですが、AI エージェントの実験基盤として早めに触っておく価値がありそうです。
We're hiring!
燈ではAIエージェントのアーキテクチャのベストプラクティスを共に探求していただける方を募集しています！興味がある方はぜひカジュアル面談で議論させてください！よろしくお願いします！
