RL training

このガイドでは、モデルが分離された環境でツール呼び出しを実行する強化学習 (RL) トレーニングワークフロー向けにサーバーレスサンドボックスの使い方を説明します。エージェントトレーニングパイプラインを構築する ML エンジニアおよび研究者を対象としており、モデルが生成した action を安全に評価し、報酬を計算するためのエフェメラルな環境を必要とする方向けです。

エージェントのツール実行にサンドボックスを使う理由

RL でコードエージェントをトレーニングするには、分離された環境でツール呼び出し (bash コマンド、ファイル操作) を実行する必要があります。信頼できないモデル生成コードは、ホストのファイルシステムを変更したり、ネットワークにアクセスしたり、非決定的な結果を生成したりする可能性があります。サンドボックスは、ホストや他のロールアウトに影響を与えずにツール呼び出しを実行できる、分離されたエフェメラルな環境を提供します。トレーニングループでは、モデルが action (ツール呼び出し) を生成し、サンドボックスがそれを実行して、観測値をモデルに返します。サンドボックスは、エピソード内のツール呼び出し間で維持されるため、ファイルの変更やインストール済みパッケージはステップ間で引き継がれます。報酬は、サンドボックスの最終状態 (たとえば、テストの成功) や軌跡の品質から得られます。タグ付け API と一覧取得 API により、数千のサンドボックスを使用するトレーニング run のクリーンアップや監視を実用的に行えます。

前提条件

始める前に、次のセットアップ手順を完了してください。 CWSANDBOX_API_KEY に CoreWeave API アクセストークンを設定します。

export CWSANDBOX_API_KEY="[API-ACCESS-TOKEN]"

Python SDK をインストールします。

uv pip install cwsandbox

このページの例では、SandboxDefaults からサンドボックスを作成し、デフォルトで CoreWeave 認証を使用します。Forge APIキーを使用する場合は、はじめにに記載されている明示的な認証セットアップの手順に従い、生成された auth の値を各 SandboxDefaults(...) に渡してください。

コアパターン

コアのセットアップは、トレーニングインフラストラクチャー上で実行されるエージェントループで、ツール呼び出しはサンドボックス内で実行されます。

import cwsandbox
from cwsandbox import Sandbox

def run_agent_episode(model, task: dict, sandbox: Sandbox) -> tuple[list, float]:
    """Run one agent episode, returning trajectory and reward."""
    messages = [{"role": "user", "content": task["prompt"]}]

    for step in range(task.get("max_steps", 10)):
        # モデルが次の action を生成します
        response = model.generate(messages)
        messages.append({"role": "assistant", "content": response})

        tool_calls = parse_tool_calls(response)
        if not tool_calls:
            break

        # サンドボックスでツール呼び出しを実行します
        for tool in tool_calls:
            if tool.name == "bash":
                result = sandbox.exec(
                    ["bash", "-c", tool.command],
                    timeout_seconds=30.0,
                ).result()
                observation = f"exit={result.returncode}\n{result.stdout}{result.stderr}"
            elif tool.name == "read_file":
                content = sandbox.read_file(tool.path).result()
                observation = content.decode()
            elif tool.name == "write_file":
                sandbox.write_file(tool.path, tool.content.encode()).result()
                observation = "File written successfully"

            messages.append({"role": "tool", "name": tool.name, "content": observation})

    # サンドボックスの最終状態から reward を計算します
    test_result = sandbox.exec(task["test_command"]).result()
    reward = 1.0 if test_result.returncode == 0 else 0.0

    return messages, reward

サンドボックスはエピソード内のツール呼び出しを通じて永続化されるため、エージェントが作業するにつれてファイルの変更が蓄積されます。

並列エピソードを用いたトレーニングステップ

エピソードごとに 1 つのサンドボックスを使用して、タスクのバッチを処理します。

def training_step(model, batch: list[dict], session) -> list[float]:
    """Run agent episodes for a batch of tasks."""

    # すべてのサンドボックスを並列に作成して事前起動します
    sandboxes = [session.sandbox() for _ in batch]
    refs = [sb.start() for sb in sandboxes]
    [r.result() for r in refs]  # すべてのバックエンドが受け入れるのを待ちます

    trajectories = []
    rewards = []

    for task, sandbox in zip(batch, sandboxes):
        trajectory, reward = run_agent_episode(model, task, sandbox)
        trajectories.append(trajectory)
        rewards.append(reward)
        sandbox.stop()  # ノンブロッキングなクリーンアップ

    # 軌跡と rewards はポリシー更新に使用されます
    return rewards

ジョブのメタデータのタグ付け

タグを使用すると、トレーニングジョブによって作成されたサンドボックスをフィルターして検索できます。単一の run から数千ものサンドボックスが生成されることもあるため、タグは欠かせません。デバッグやクリーンアップの際にサンドボックスを特定しやすくなるよう、メタデータを含めてください。

import os
from cwsandbox import SandboxDefaults

def make_defaults(model_name: str) -> SandboxDefaults:
    return SandboxDefaults(
        container_image="python:3.11",
        tags=(
            f"training-run:{os.environ.get('TRAINING_RUN_ID', 'local')}",
            f"slurm-job:{os.environ.get('SLURM_JOB_ID', 'interactive')}",
            f"model:{model_name}",
            "rl-training",
        ),
    )

タグに含めると便利なメタデータ:

タグのパターン	用途
`training-run:{id}`	トレーニング run ID (`TRAINING_RUN_ID` から取得) 。トレーニング run でのフィルタリングに使用します
`slurm-job:{id}`	Slurm ジョブ ID (`SLURM_JOB_ID` 環境変数から取得) 。クラスターのジョブのトラッキングに使用します
`model:{name}`	モデル名またはチェックポイント。複数のモデルを扱う実験に使用します
`env:{name}`	環境 (dev、staging、prod) 。リソース管理に使用します

サンドボックスのタグは Kubernetes の Pod ラベルになり、CoreWeave の可観測性プラットフォームでフィルタリングやダッシュボードに使用されます。

報酬関数の例を試す

コード実行の報酬を並列サンドボックス実行で計算する最小限のインテグレーションです。 実行内容:

算術、文字列操作、構文エラー、ランタイムエラーなどのおもちゃのコード完了のセットを実行します。
隔離のために完了ごとに1つのサンドボックスを作成します。
二値報酬を計算します: 実行成功で1.0、失敗で0.0。
結果が到着するにつれて進捗を表示します (より高速な実行が先に完了します)。

CoreWeave Sandboxes の使用方法: この例では cwsandbox.wait() を使用して、完了した結果を処理します:

# サンドボックスを作成し、すべての完了を並列に実行します
processes = [
    session.sandbox().exec(
        ["python", "-c", code],
        timeout_seconds=EXECUTION_TIMEOUT_SECONDS,
    )
    for code in completions
]

# 完了した結果を収集します
while pending:
    [process], pending = cwsandbox.wait(pending, num_returns=1)
    result = process.result()
    reward = 1.0 if result.returncode == 0 else 0.0

実行してください:

uv run examples/rl_training/reward_function.py

追加の依存関係は必要ありません。GPU は必要ありません。 期待される出力: 結果は実行が完了するたびに到着するため、より高速な問題が先に終了します:

RL Training Reward Function Example (job: cc92116b)
============================================================

Evaluating 5 completions...

Progress (results arrive as executions complete):
------------------------------------------------------------
  [1/5] Problem 1 (string-ops): PASS
  [2/5] Problem 3 (syntax-error): FAIL
  [3/5] Problem 2 (delayed-error): FAIL
  [4/5] Problem 0 (slow-sum): PASS
  [5/5] Problem 4 (slow-list): PASS
------------------------------------------------------------

Final summary (original order):
------------------------------------------------------------
  Problem 0 (slow-sum): reward=1.0 [PASS] OK
  Problem 1 (string-ops): reward=1.0 [PASS] OK
  Problem 2 (delayed-error): reward=0.0 [FAIL] OK
  Problem 3 (syntax-error): reward=0.0 [FAIL] OK
  Problem 4 (slow-list): reward=1.0 [PASS] OK
------------------------------------------------------------
Total reward: 3.0/5
Pass rate: 3/5 (60%)

TRL GRPOTrainer インテグレーション

TRL では、完了が直接報酬にマッピングされる報酬関数インターフェースを使用しています。エージェントが完了を出力し、報酬関数がそれをサンドボックスで実行します。標準的なパターンでは、コード抽出に <answer> XML タグを使用します (GRPO の数学例で使用される \boxed{} 形式に一致) ：

import cwsandbox
from cwsandbox import SandboxDefaults

session = cwsandbox.Session(defaults=SandboxDefaults(
    container_image="python:3.11",
    tags=("trl-grpo",),
))

def extract_xml_answer(text: str) -> str:
    if "<answer>" not in text:
        return ""
    return text.split("<answer>")[-1].split("</answer>")[0].strip()

def reward_fn(completions, **kwargs) -> list[float]:
    # 会話データセットはメッセージのdictを渡し、標準データセットはstringsを渡します
    texts = [c[0]["content"] if isinstance(c, list) else c for c in completions]
    codes = [extract_xml_answer(t) for t in texts]
    code_indices = [(i, code) for i, code in enumerate(codes) if code]

    processes = [
        (i, session.sandbox().exec(
            ["python", "-c", code],
            timeout_seconds=30.0,
        ))
        for i, code in code_indices
    ]

    rewards = [0.0] * len(codes)
    for i, process in processes:
        try:
            rewards[i] = 1.0 if process.result().returncode == 0 else 0.0
        except Exception:
            pass

    return rewards

このパターンは、1回のターンで正しいコードを生成するようにモデルをトレーニングする場合に適しています。

TRL GRPO インテグレーションの例を試す

CoreWeave Sandboxes を TRL の GRPOTrainer と組み合わせて使用し、コード実行の報酬を計算します。 内容:

小規模なモデル (Qwen/Qwen2.5-0.5B-Instruct) を読み込みます。
シンプルなコーディング問題で構成されるトイデータセットを作成します。
サンドボックスベースの報酬計算を用いて、GRPO でモデルをトレーニングします。
インテグレーションの動作を示すために、10 トレーニングステップを実行します。

CoreWeave Sandboxes の使用方法: 報酬関数は、<answer> タグからコードを抽出し (標準的な GRPO パターン)、Session を通じてサンドボックスを並列に作成して各完了を実行し、二値の報酬を返します。

def extract_xml_answer(text: str) -> str:
    """Extract answer from XML-style <answer> tags."""
    if "<answer>" not in text:
        return ""
    answer = text.split("<answer>")[-1]
    answer = answer.split("</answer>")[0]
    return answer.strip()

def code_execution_reward(completions, **kwargs) -> list[float]:
    # 会話型データセットはメッセージの dict を渡し、標準データセットは strings を渡します
    texts = [c[0]["content"] if isinstance(c, list) else c for c in completions]
    codes = [extract_xml_answer(t) for t in texts]

    # サンドボックスを作成し、非空のコードを並列に実行します
    processes = [
        (i, session.sandbox().exec(
            ["python", "-c", code],
            timeout_seconds=EXECUTION_TIMEOUT_SECONDS,
        ))
        for i, code in enumerate(codes) if code
    ]

    # 報酬を収集し、デフォルトで 0.0 にします
    rewards = [0.0] * len(codes)
    for i, process in processes:
        try:
            result = process.result()
            rewards[i] = 1.0 if result.returncode == 0 else 0.0
        except Exception:
            rewards[i] = 0.0

    return rewards

プロンプトでは、コードを <answer> タグで囲んで出力するようモデルに指示するシステムメッセージを使用します:

SYSTEM_PROMPT = """You solve coding problems by writing Python code.
Put your code inside <answer> tags like this: <answer>print("hello")</answer>
Only include the code, no explanations."""

Session はサンドボックスをトラッキングし、閉じるときにそれらをクリーンアップします。 実行:

uv pip install trl==0.27.1 transformers==5.0.0 datasets==4.5.0 torch==2.10.0
uv run examples/rl_training/trl_grpo_integration.py

合理的なパフォーマンスを得るには GPU を使用することを推奨します。GPU がない場合、トレーニングは動作しますが遅くなります。 期待される出力:

TRL GRPO Integration Example (job: def67890)
============================================================

Loading model: Qwen/Qwen2.5-0.5B-Instruct
Creating toy dataset...
Dataset size: 5 problems

Setting up GRPOTrainer...

Starting training (10 steps)...
------------------------------------------------------------
  [CWSandbox] Reward call 1: 2 sandboxes, 0/2 passed
  [CWSandbox] Reward call 2: 1 sandboxes, 0/1 passed, 1 skipped (no code)
  [CWSandbox] Reward call 3: 0 sandboxes, 0/0 passed, 2 skipped (no code)
  [CWSandbox] Reward call 4: 2 sandboxes, 0/2 passed
  ...
  [CWSandbox] Reward call 8: 2 sandboxes, 1/2 passed
  [CWSandbox] Reward call 9: 2 sandboxes, 1/2 passed
  [CWSandbox] Reward call 10: 1 sandboxes, 0/1 passed, 1 skipped (no code)
[training logs]
------------------------------------------------------------

Training completed successfully!

出力の理解: サンドボックスの数はステップごとに異なります。これは、スクリプトが extract_xml_answer() がモデルの完了から抽出可能なコードを検索した場合にのみサンドボックスを作成するためです。モデルが期待される <answer>...</answer> タグなしでテキストを生成した場合、その完了はスキップされ、報酬 0.0 を受け取ります。

2 sandboxes, 0/2 passed: モデルが 2 個のコードブロックを生成し、どちらも実行に失敗しました。
1 sandboxes, 0/1 passed, 1 skipped (no code): モデルが 1 個のコードブロック (失敗) と 1 個のテキストのみの完了を生成しました。
0 sandboxes, 0/0 passed, 2 skipped (no code): モデルがどちらの完了からも抽出可能なコードを生成しませんでした。

小さな未トレーニングのモデルではこのような結果が予想されます。トレーニングが進むにつれて、スキップされる完了が減り、パスが増えるはずです。

エージェントエピソードにおけるエラーの処理

サンドボックス操作は、タイムアウト、ファイルの欠落、サンドボックスの終了などによって失敗することがあります。こうした失敗をどのようにエージェントに伝えるかによって、エージェントが何を学習するかが決まります。何が問題だったのかを説明する観測値を返してください。

from cwsandbox import SandboxTimeoutError, SandboxFileError

def execute_tool(sandbox, tool) -> str:
    """Execute a tool call, returning an observation string."""
    try:
        if tool.name == "bash":
            result = sandbox.exec(
                ["bash", "-c", tool.command],
                timeout_seconds=30.0,
            ).result()
            return f"exit={result.returncode}\n{result.stdout}{result.stderr}"
        elif tool.name == "read_file":
            content = sandbox.read_file(tool.path).result()
            return content.decode()
        elif tool.name == "write_file":
            sandbox.write_file(tool.path, tool.content.encode()).result()
            return "File written successfully"
    except SandboxTimeoutError:
        return "Error: command timed out after 30 seconds"
    except SandboxFileError as e:
        return f"Error: {e}"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

報酬の計算では、例外をキャッチして、トレーニングループに伝播させる代わりにフォールバック報酬を返します。

Weights & Biases メトリクスインテグレーション

Weights & Biases をトレーニングに使用すると、cwsandbox Sessions はサンドボックスの使用状況メトリクスを実行中の wandb run に自動的にログします。これにより、サンドボックスの健全性 (起動時間、実行の失敗、ロールアウトごとのツール呼び出し) を同じダッシュボード内のトレーニングメトリクスと関連付けることができます。以下のセクションでは、自動検出の仕組み、report を明示的に制御する方法、および追跡されるメトリクスについて説明します。

自動検出

wandb パッケージがインストールされていて、Weights & Biases の run が実行中 (wandb.run が存在する) の場合、メトリクスをログする機能が自動的に有効になります。実験管理を行うには、省略可能な wandb パッケージをインストールし、Forge の APIキーを設定してください：

uv pip install wandb
export FORGE_API_KEY="[FORGE-API-KEY]"

キーは wandb.login() に明示的に渡してください。FORGE_API_KEY はこの例のコードで読み取っているもので、SDK が自動的に検出するわけではありません。サンドボックスの認証には、引き続き CWSANDBOX_API_KEY を使用します：

import os

import wandb
from cwsandbox import Session, SandboxDefaults

wandb.login(key=os.environ["FORGE_API_KEY"])
wandb.init(project="my-rl-training")
defaults = SandboxDefaults(container_image="python:3.11")
code = "print('Training tool call')"

# メトリクスをログする機能が自動的に有効になります
with Session(defaults) as session:
    for step in range(3):
        sandbox = session.sandbox()
        # Exec 結果は自動的に追跡されます - 手動の Call は不要です
        result = sandbox.exec(["python", "-c", code]).result()
        # このトレーニングステップでメトリクスをログして相関を調べます
        session.log_metrics(step=step)
# セッション終了時に最終メトリクスがログされます

明示的な制御

report_to パラメーターで、メトリクスのレポート先を制御します：

# 明示的なオプトイン（reporter を作成; wandb.run が存在する場合にメトリクスをログする）
session = Session(defaults, report_to=["wandb"])

# report を無効にする（wandb run が存在する場合でも）
session = Session(defaults, report_to=[])

# 自動検出（デフォルトの動作）
session = Session(defaults, report_to=None)

メトリクス

exec() が完了すると、実行メトリクスが自動的に追跡されます：

メトリクス	説明
`cwsandbox/sandboxes_created`	セッションを通じて作成されたサンドボックスの総数
`cwsandbox/executions`	exec() の呼び出し総数
`cwsandbox/exec_completed_ok`	完了した実行 (returncode=0)
`cwsandbox/exec_completed_nonzero`	完了した実行 (returncode!=0)
`cwsandbox/exec_failures`	失敗した実行 (タイムアウト、通信障害)
`cwsandbox/exec_completion_rate`	returncode=0 で完了した exec() の割合
`cwsandbox/exec_failure_rate`	完了できなかった exec() の割合
`cwsandbox/startup_count`	記録されたサンドボックス起動時間の件数
`cwsandbox/avg_startup_seconds`	サンドボックスの平均起動時間
`cwsandbox/min_startup_seconds`	サンドボックスの最短起動時間
`cwsandbox/max_startup_seconds`	サンドボックスの最長起動時間

追跡は自動的に行われます。セッションに関連付けられた任意のサンドボックスで exec() を呼び出します。特定のトレーニングステップでログするには、session.log_metrics(step=N) を呼び出します：

def training_step(session, model, batch, step: int) -> list[float]:
    rewards = []
    for task in batch:
        sandbox = session.sandbox()
        # exec() の完了時にメトリクスが自動的に追跡されます
        result = sandbox.exec(["python", "-c", task["code"]]).result()
        reward = 1.0 if result.returncode == 0 else 0.0
        rewards.append(reward)
        sandbox.stop()

    # 相関を取るため、このトレーニングステップでメトリクスをログします
    session.log_metrics(step=step)
    return rewards

exec_stats プロパティを使用して、サンドボックスごとの統計にアクセスすることもできます：

sandbox = session.sandbox()
result = sandbox.exec(["echo", "hello"]).result()
print(sandbox.exec_stats)  # {"exec_count": 1, "completed_ok": 1, "completed_nonzero": 0, "failures": 0}

サンドボックスごとの exec メトリクス

Weights & Biases インテグレーションを使用するセッションでは、サンドボックスごとのメトリクスもトラッキングします。

メトリクス	説明
`cwsandbox/avg_execs_per_sandbox`	サンドボックスあたりの平均 exec() 呼び出し数 (「ロールアウトあたりのツール呼び出し数」の把握に役立ちます)
`cwsandbox/min_execs_per_sandbox`	全サンドボックス中の最小 exec() 呼び出し数
`cwsandbox/max_execs_per_sandbox`	全サンドボックス中の最大 exec() 呼び出し数

これらのメトリクスからわかるエージェントの動作:

avg_execs_per_sandbox の値が高い場合、エージェントが 1 エピソードあたりに多数のツール呼び出しを行う冗長な動作をしている可能性があります。
ばらつき (max-min の差) が大きい場合、エピソード間でロールアウトの動作に一貫性がない可能性があります。
トレーニングステップごとの傾向から、ポリシーの改善に伴ってエージェントの動作がどのように変化するかを確認できます。

ダッシュボードの使用例:

avg_execs_per_sandbox をトレーニングステップに対してプロットし、トレーニングを通じたツールの使用傾向を確認します。
max_execs_per_sandbox がしきい値を超えた場合にアラートを発行します (過剰なツール呼び出しを行う暴走エージェントの検知)。
min/max の幅を比較し、エージェントがループに陥ったエピソードと短時間で完了したエピソードを見分けます。

デフォルトでは、log_metrics() はログした後にカウンターをリセットします。累積を継続するには、reset=False を設定します。

session.log_metrics(step=step, reset=False)  # 累積を継続します

メトリクスはセッションの終了時にも自動的にログされるため、明示的にログしなくても最終サマリーメトリクスを取得できます。

モニタリングとデバッグ

以下のパターンを使用すると、トレーニング run の動作をリアルタイムで確認し、障害の診断に役立つ詳細情報を取得できます。

実行中のサンドボックスのカウント

トレーニング中のサンドボックス使用状況を監視します：

from cwsandbox import Sandbox, SandboxStatus

def count_active_sandboxes(run_id: str) -> dict:
    sandboxes = Sandbox.list(
        tags=[f"training-run:{run_id}"],
        status=[SandboxStatus.RUNNING, SandboxStatus.PENDING],
    ).result()

    return {
        "running": sum(1 for s in sandboxes if s.status == SandboxStatus.RUNNING),
        "pending": sum(1 for s in sandboxes if s.status == SandboxStatus.PENDING),
        "total": len(sandboxes),
    }

execution の詳細をログする

報酬計算のデバッグのために execution の詳細を取得します：

import logging

logger = logging.getLogger(__name__)

# モジュールレベルで `session` が作成されていることを前提としています

def logged_reward(completion: str, step: int) -> float:
    code = extract_xml_answer(completion)  # <answer> タグからコードを抽出します

    sandbox = session.sandbox()
    sandbox.wait()
    result = sandbox.exec(
        ["python", "-c", code],
        timeout_seconds=30.0,
    ).result()

    logger.debug(
        "Reward computation",
        extra={
            "step": step,
            "sandbox_id": sandbox.sandbox_id,
            "returncode": result.returncode,
            "stdout_len": len(result.stdout),
            "stderr_len": len(result.stderr),
        },
    )

    sandbox.stop()
    return 1.0 if result.returncode == 0 else 0.0

ART を使用したマルチステップロールアウト

前の TRL の例では、single-shot 実行にサンドボックスを使用しています: 完了ごとに 1 つのサンドボックスで、1 回実行して報酬を返します。これは、モデルをトレーニングして 1 回の試行で正しいコードを生成するのに適しています。ステートフルなマルチステップロールアウトは異なります: エージェントは単一のサンドボックス内で複数の action を実行し、サンドボックスは action 間で状態を維持します。エージェントはファイルを書き込み、実行し、エラーを観察し、ファイルを編集して再試行できます。すべて同じサンドボックス内で行われます。 examples/rl_training/art/ ディレクトリは、MBPP ベンチマークでこのパターンを示しています。ソリューションが失敗した場合、エージェントはエラーのフィードバックを受け取り、アプローチを反復できます。

概要

このセクションの以降の内容では、ART の概要、この例で使用するトレーニング手法、前提条件、インストール、例の実行方法、および各コンポーネントの連携について説明します。 ART (Agent Reinforcement Trainer) は、GRPO を使用して複数ステップのエージェントをトレーニングするための、OpenPipe によるオープンソースの RL フレームワークです。この例では、CoreWeave Sandboxes を ART と統合します。

art パッケージ (openpipe-art) を使用して、軌跡の収集とトレーニングを行います。
LocalBackend (GPU が必要) と TinkerBackend (GPU 不要) の 2 つのバックエンドをサポートします。
CoreWeave Sandboxes でツール呼び出しを通じてコードを実行します。
MBPP テストケースの結果に基づいて二値の報酬を計算します。

GRPO と蒸留を用いたトレーニングアプローチ

この例では、強化学習を用いた蒸留を使用します。より強力なモデルがデモンストレーションを生成し、より小さなモデルが成功したものを複製することを学習します。 2 つのモデルが関与します：

Inference model (--model, default: gpt-5.1-codex-mini): ロールアウト中に軌跡を生成します。このモデルはツール呼び出しを行い、サンドボックスの結果を確認し、エラーを反復し、ソリューションを送信します。トレーニングされません。
Base model (--base-model, default: Qwen/Qwen3-8B): トレーニングされるモデルです。inference model が生成した軌跡を受け取り、GRPO (Group Relative Policy Optimization) を通じてそれらから学習します。

動作の仕組み：

inference model は問題ごとに複数の軌跡を生成し、それぞれのツール呼び出しは CoreWeave サンドボックスで実行されます。
各軌跡には二値報酬が与えられます。テストに合格した場合は 1.0、不合格の場合は 0.0 です。
同じ問題の軌跡はグループを形成し、GRPO は各グループ内の軌跡を比較します。
base model (Qwen3-8B) は、より高い報酬の軌跡を低い報酬の軌跡よりも優先するようにトレーニングされます。

トレーニング後、Qwen3-8B を同じツール定義とともにデプロイします。inference model の成功した軌跡を模倣することで、同様のツール呼び出しを行うことを学習しています。

前提条件

モード	要件
ドライラン	CPU のみ
TinkerBackend	CPU のみ (API 経由でトレーニング)
LocalBackend	GPU 必須

環境変数:

export OPENAI_API_KEY="your-openai-key"
export ART_TINKER_API_KEY="your-tinker-key"  # --backend=tinker に必須
export CWSANDBOX_API_KEY="[API-ACCESS-TOKEN]"  # サンドボックスの認証

Weights & Biases にログする場合 (省略可能) は、Forge APIキーを設定します:

export FORGE_API_KEY="[FORGE-API-KEY]"

ART 0.5.7 は、ログのインテグレーションを有効にする前に、従来の WANDB_API_KEY 変数を確認します。バックエンドを作成する前に、次の互換性マッピングを train.py に追加してください。このキーには Forge の APIキーを使用します。

import os

if api_key := os.environ.get("FORGE_API_KEY"):
    os.environ["WANDB_API_KEY"] = api_key

インストール

uv pip install -r examples/rl_training/art/requirements.txt

これにより、以下がインストールされます：

openpipe-art==0.5.7       # ART フレームワーク
openai==2.15.0            # LLM 推論
datasets==4.5.0           # MBPP 読み込み
wandb==0.24.0             # オプションでログする

GPU サポート付きの LocalBackend を使用する場合は、以下もインストールしてください。

uv pip install "openpipe-art[backend]==0.5.7"

例の実行

# Dry run - セットアップを検証し、トレーニングせずに実行
uv run examples/rl_training/art/train.py --dry-run

# TinkerBackend でトレーニングする (ローカル GPU は不要)
uv run examples/rl_training/art/train.py --backend tinker --num-problems 10

# LocalBackend でトレーニングする (GPU が必要)
uv run examples/rl_training/art/train.py --backend local --num-problems 10

期待される出力:

ART Training with CWSandbox
========================================
Backend: tinker
Model: gpt-5.1-codex-mini
Base model: Qwen/Qwen3-8B
Problems: 10
Steps: 5
Trajectories per problem: 2
Project: cwsandbox-mbpp
Run name: train-001

Loading MBPP problems...
Loaded 10 problems

Creating tinker backend...
Creating trainable model...
Registering model with backend...

Starting training...

=== Step 1 ===
Collecting trajectories for 10 problems...
step 1: 100%|██████████| 10/10 [00:45<00:00]
Collected 20 trajectories, avg reward: 0.35
Training...
Training complete: step=1, metrics={'loss': 0.42}
...

設定オプション

フラグ	デフォルト	説明
`--backend`	`local`	トレーニングのバックエンド: `local` (GPU) または `tinker` (GPU なし)
`--model`	`gpt-5.1-codex-mini`	推論用モデル
`--base-model`	`Qwen/Qwen3-8B`	トレーニング用ベースモデル
`--num-problems`	`10`	MBPP の問題数
`--num-steps`	`5`	トレーニングステップ数
`--trajectories-per-problem`	`2`	各ステップで問題ごとに収集する軌跡数
`--base-url`	`None`	OpenAI 互換 API のベース URL
`--project`	`cwsandbox-mbpp`	Weights & Biases プロジェクト名
`--run-name`	`train-001`	トレーニング run 名
`--learning-rate`	`1e-5`	学習率
`--dry-run`	`false`	トレーニングせずにセットアップを検証します

アーキテクチャ

art/
├── train.py         # トレーニングループ、CLI、ART バックエンドの設定
├── rollout.py       # 複数ステップのサンドボックス実行、Trajectory の構築
├── tools.py         # execute_code と submit_solution のツールスキーマ
└── __init__.py

主な ART のインポート:

import art
from art.local import LocalBackend
from art.tinker import TinkerBackend

# トレーニング可能なモデルを作成します
model = art.TrainableModel(
    name="train-001",
    project="cwsandbox-mbpp",
    base_model="Qwen/Qwen3-8B",
)

# 軌跡を収集します
groups = await art.gather_trajectory_groups(
    (collect_trajectories(problem) for problem in problems),
    pbar_desc="collecting",
)

# トレーニングします
result = await backend.train(model, groups, learning_rate=1e-5)

ロールアウトは art.Trajectory を返します:

from openai.types.chat import ChatCompletionToolParam

trajectory = art.Trajectory(
    messages_and_choices=messages_and_choices,  # 会話履歴
    tools=ROLLOUT_TOOLS,                        # ツール定義
    reward=1.0 if passed else 0.0,              # 二値報酬
    metadata={"task_id": problem.task_id},
)
return trajectory.finish()

ツール呼び出しパターン: ロールアウトでは、次の2つのツールを使用して OpenAI 互換のツール呼び出しを行います:

execute_code: サンドボックスでコードをテストし、標準出力/標準エラー出力を返します。
submit_solution: 最終的な解答を送信し、すべてのテストケースを実行します。

ROLLOUT_TOOLS: list[ChatCompletionToolParam] = [
    {
        "type": "function",
        "function": {
            "name": "execute_code",
            "description": "Execute Python code in an isolated sandbox...",
            "parameters": {
                "type": "object",
                "properties": {"code": {"type": "string"}},
                "required": ["code"],
            },
        },
    },
    # submit_solution ツール...
]

サンドボックスには、トラッキング用のタグが付けられます。

sandbox_defaults = SandboxDefaults(
    container_image="python:3.11",
    tags=("art-training", args.project, args.run_name),
)

データフロー

トレーニングパイプラインは、いくつかのコンポーネントで構成されます。

ローカルマシン: トレーニングスクリプト (train.py) は、マシンまたはトレーニングサーバー上で実行されます。MBPP データセットから問題を読み込み、トレーニングループを制御します。
OpenAI API (推論): 軌跡の収集中、ロールアウトコードは OpenAI API (または互換エンドポイント) を呼び出してモデルの応答を生成します。モデルはツール定義を受け取り、ロールアウトが実行するツール呼び出しを返します。
CoreWeave サンドボックス (コード実行): 各ロールアウトは、すべてのツール呼び出しにわたって維持される単一の CoreWeave サンドボックスを使用します。モデルが execute_code または submit_solution を呼び出すと、コードはそのサンドボックス内で実行されます。そのため、エージェントが処理を繰り返すにつれて、ファイルの変更と状態が蓄積されます。ファイルを書き込み、実行し、エラーを確認して修正できます。サンドボックスは隔離環境を提供するため、モデルが生成した信頼できないコードがホストに影響を及ぼすことはありません。結果 (stdout、stderr、終了コード) はロールアウトに返されます。
軌跡の収集: ロールアウトは、会話の履歴 (メッセージとツール結果) を最終報酬とともに art.Trajectory オブジェクトに蓄積します。同じ問題に対する複数の軌跡が、1 つの art.TrajectoryGroup を形成します。
トレーニングバックエンド: 収集された軌跡グループは、トレーニングバックエンドに送信されます。LocalBackend では、ローカル GPU 上でトレーニングが行われます。TinkerBackend では、軌跡が Thinking Machines の Tinker サービスにアップロードされ、リモートでトレーニングが行われます。ローカル GPU は必須ではありません。

サンドボックス

サンドボックスを使用する

サンドボックスの管理

リファレンス

エージェントのツール実行にサンドボックスを使う理由

前提条件

コアパターン

並列エピソードを用いたトレーニングステップ

ジョブのメタデータのタグ付け

報酬関数の例を試す

TRL GRPOTrainer インテグレーション

TRL GRPO インテグレーションの例を試す

エージェントエピソードにおけるエラーの処理

Weights & Biases メトリクスインテグレーション

自動検出

明示的な制御

メトリクス

サンドボックスごとの exec メトリクス

モニタリングとデバッグ

実行中のサンドボックスのカウント

execution の詳細をログする

ART を使用したマルチステップロールアウト

概要

GRPO と蒸留を用いたトレーニングアプローチ

前提条件

インストール

例の実行

設定オプション

アーキテクチャ

データフロー

​エージェントのツール実行にサンドボックスを使う理由

​前提条件

​コアパターン

​並列エピソードを用いたトレーニングステップ

​ジョブのメタデータのタグ付け

​報酬関数の例を試す

​TRL GRPOTrainer インテグレーション

​TRL GRPO インテグレーションの例を試す

​エージェントエピソードにおけるエラーの処理

​Weights & Biases メトリクス インテグレーション

​自動検出

​明示的な制御

​メトリクス

​サンドボックスごとの exec メトリクス

​モニタリングとデバッグ

​実行中のサンドボックスのカウント

​execution の詳細をログする

​ART を使用したマルチステップ ロールアウト

​概要

​GRPO と蒸留を用いたトレーニングアプローチ

​前提条件

​インストール

​例の実行

​設定オプション

​アーキテクチャ

​データフロー

エージェントのツール実行にサンドボックスを使う理由

前提条件

コアパターン

並列エピソードを用いたトレーニングステップ

ジョブのメタデータのタグ付け

報酬関数の例を試す

TRL GRPOTrainer インテグレーション

TRL GRPO インテグレーションの例を試す

エージェントエピソードにおけるエラーの処理

Weights & Biases メトリクスインテグレーション

自動検出

明示的な制御

メトリクス

サンドボックスごとの exec メトリクス

モニタリングとデバッグ

実行中のサンドボックスのカウント

execution の詳細をログする

ART を使用したマルチステップロールアウト

概要

GRPO と蒸留を用いたトレーニングアプローチ

前提条件

インストール

例の実行

設定オプション

アーキテクチャ

データフロー