RL training

이 가이드에서는 모델이 격리된 환경에서 도구 Call을 실행하는 강화 학습(RL) 트레이닝 워크플로에 서버리스 샌드박스를 사용하는 방법을 설명합니다. 에이전트 트레이닝 파이프라인을 구축하는 ML 엔지니어와 연구자를 대상으로 하며, 모델이 생성한 액션을 평가하고 보상을 계산할 수 있는 안전한 임시 환경이 필요한 경우에 적합합니다.

에이전트 도구 실행에 샌드박스를 사용하는 이유

RL로 코드 에이전트를 트레이닝하려면 격리된 환경에서 도구 Call(bash 명령어, 파일 오퍼레이션)을 실행해야 합니다. 신뢰할 수 없는 모델 생성 코드는 호스트 파일 시스템을 수정하거나 네트워크에 접근하거나 비결정적 결과를 생성할 수 있습니다. 샌드박스는 호스트나 다른 롤아웃에 영향을 주지 않고 도구 Call을 실행할 수 있는 격리된 임시 환경을 제공합니다. 트레이닝 루프에서 모델은 액션(도구 Call)을 생성하고, 샌드박스가 이를 실행하며, 관찰 결과가 모델로 다시 흐릅니다. 샌드박스는 에피소드 내 도구 Call 간에 지속되므로 파일 변경 사항과 설치된 패키지가 단계 간에 유지됩니다. 보상은 최종 샌드박스 상태(예: 테스트 통과) 또는 트래젝터리 품질에서 옵니다. 태깅 및 목록 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(...)에 전달하세요.

핵심 패턴

핵심 설정은 트레이닝 인프라에서 실행되는 에이전트 루프이며, 도구 Call은 샌드박스에서 실행됩니다.

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)):
        # 모델이 다음 액션을 생성합니다
        response = model.generate(messages)
        messages.append({"role": "assistant", "content": response})

        tool_calls = parse_tool_calls(response)
        if not tool_calls:
            break

        # 샌드박스에서 도구 Call을 실행합니다
        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})

    # 샌드박스의 최종 상태를 기반으로 보상을 계산합니다
    test_result = sandbox.exec(task["test_command"]).result()
    reward = 1.0 if test_result.returncode == 0 else 0.0

    return messages, reward

샌드박스는 에피소드 내 도구 Call 간에 유지되므로, 에이전트가 작업하는 동안 파일 변경 사항이 누적됩니다.

에피소드를 병렬로 실행하는 트레이닝 단계

에피소드마다 샌드박스를 하나씩 사용하여 작업 배치를 처리하세요:

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()  # 논블로킹 정리

    # 트래젝터리와 보상을 정책 업데이트에 전달
    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별 필터링을 위한 트레이닝 run ID (`TRAINING_RUN_ID`에서 가져옴)
`slurm-job:{id}`	클러스터 작업 추적을 위한 Slurm 작업 ID (`SLURM_JOB_ID` 환경 변수에서 가져옴)
`model:{name}`	다중 모델 실험을 위한 모델 이름 또는 체크포인트
`env:{name}`	리소스 관리를 위한 환경(dev, staging, prod)

샌드박스 태그는 Kubernetes 파드 레이블로 변환되며, CoreWeave 관측성 플랫폼은 이를 필터링과 대시보드에 사용합니다.

보상 함수 예시 사용해 보기

샌드박스 병렬 실행으로 코드 실행 보상을 계산하는 최소한의 인테그레이션입니다. 수행하는 작업:

간단한 코드 완료 결과 모음(산술 연산, 문자열 오퍼레이션, 구문 오류, 런타임 오류)을 실행합니다.
격리를 위해 각 완료 결과마다 샌드박스를 하나씩 생성합니다.
이진 보상을 계산합니다. 실행 성공 시 1.0, 실패 시 0.0입니다.
결과가 도착할 때마다 진행 상황을 표시합니다(실행 속도가 빠른 작업이 먼저 완료됩니다).

CoreWeave 샌드박스 사용 방식: 이 예시는 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 트레이닝 보상 함수 예시 (작업: cc92116b)
============================================================

생성 결과 5개 평가 중...

진행 상황 (실행이 완료되는 순서대로 결과 도착):
------------------------------------------------------------
  [1/5] 문제 1 (string-ops): 통과
  [2/5] 문제 3 (syntax-error): 실패
  [3/5] 문제 2 (delayed-error): 실패
  [4/5] 문제 0 (slow-sum): 통과
  [5/5] 문제 4 (slow-list): 통과
------------------------------------------------------------

최종 요약 (원래 순서):
------------------------------------------------------------
  문제 0 (slow-sum): reward=1.0 [통과] 정상
  문제 1 (string-ops): reward=1.0 [통과] 정상
  문제 2 (delayed-error): reward=0.0 [실패] 정상
  문제 3 (syntax-error): reward=0.0 [실패] 정상
  문제 4 (slow-list): reward=1.0 [통과] 정상
------------------------------------------------------------
총 보상: 3.0/5
통과율: 3/5 (60%)

TRL GRPOTrainer 인테그레이션

TRL은 생성된 완료가 보상에 직접 매핑되는 보상 함수 인터페이스를 사용합니다. 에이전트가 완료를 생성하면 보상 함수가 이를 샌드박스에서 실행합니다. 표준 패턴에서는 코드 추출에 <answer> XML 태그를 사용합니다(\boxed{}를 사용하는 GRPO 수학 예시의 형식과 일치):

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]:
    # 대화형 데이터셋은 메시지 딕셔너리를 전달하고, 일반 데이터셋은 문자열을 전달합니다
    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

이 패턴은 단일 턴에서 올바른 코드를 생성하도록 모델을 트레이닝하는 데 적합합니다.

TRL GRPO 인테그레이션 예시 사용해 보기

TRL’s GRPOTrainer와 CoreWeave 샌드박스를 사용하여 코드 실행에 대한 보상을 계산합니다. 수행 작업:

소형 모델(Qwen/Qwen2.5-0.5B-Instruct)을 로드합니다.
단순한 코딩 문제로 구성된 예제 데이터셋을 생성합니다.
샌드박스 기반 보상 계산과 GRPO를 사용하여 모델을 트레이닝합니다.
트레이닝을 10단계 실행하여 인테그레이션을 시연합니다.

CoreWeave 샌드박스 활용 방식: 보상 함수는 <answer> 태그(표준 GRPO 패턴)에서 코드를 추출하고, 세션을 통해 샌드박스를 병렬로 생성한 다음, 각 완료를 실행하여 이진 보상을 반환합니다:

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를 전달하고, 일반 데이터셋은 문자열을 전달합니다
    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."""

세션은 샌드박스를 추적하고 종료 시 정리합니다. 실행하세요:

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 인테그레이션 예시 (작업: def67890)
============================================================

모델 로딩 중: Qwen/Qwen2.5-0.5B-Instruct
간단한 예제 데이터셋 생성 중...
데이터셋 크기: 문제 5개

GRPOTrainer 설정 중...

트레이닝 시작 중 (10단계)...
------------------------------------------------------------
  [CWSandbox] 보상 호출 1: 샌드박스 2개, 0/2 성공
  [CWSandbox] 보상 호출 2: 샌드박스 1개, 0/1 성공, 1개 건너뜀 (코드 없음)
  [CWSandbox] 보상 호출 3: 샌드박스 0개, 0/0 성공, 2개 건너뜀 (코드 없음)
  [CWSandbox] 보상 호출 4: 샌드박스 2개, 0/2 성공
  ...
  [CWSandbox] 보상 호출 8: 샌드박스 2개, 1/2 성공
  [CWSandbox] 보상 호출 9: 샌드박스 2개, 1/2 성공
  [CWSandbox] 보상 호출 10: 샌드박스 1개, 0/1 성공, 1개 건너뜀 (코드 없음)
[트레이닝 로그]
------------------------------------------------------------

트레이닝이 성공적으로 완료되었습니다!

출력 이해하기: 스크립트는 extract_xml_answer()가 모델의 완료에서 추출 가능한 코드를 찾을 때만 샌드박스를 생성하므로 step마다 샌드박스 수가 달라집니다. 모델이 예상되는 <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): 모델이 두 완료 모두에서 추출 가능한 코드를 생성하지 않았습니다.

이는 트레이닝되지 않은 소형 모델에서 예상되는 결과입니다. 트레이닝이 진행되면 건너뛰는 완료는 줄어들고 통과하는 응답은 늘어납니다.

에이전트 에피소드의 오류 처리

샌드박스 오퍼레이션은 timeout, 파일 누락, 샌드박스 종료 등으로 실패할 수 있으며, 이러한 실패를 어떻게 드러내느냐에 따라 에이전트가 학습하는 내용이 달라집니다. 무엇이 잘못되었는지 설명하는 관찰 결과(observation)를 반환하세요.

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 세션이 샌드박스 사용 메트릭을 활성 wandb run에 자동으로 로깅합니다. 이를 통해 동일한 대시보드에서 샌드박스 상태(시작 시간, 실행 실패, 롤아웃당 도구 Call 수)와 트레이닝 메트릭의 상관관계를 파악할 수 있습니다. 다음 섹션에서는 자동 감지의 작동 방식, 보고를 명시적으로 제어하는 방법, 추적되는 메트릭을 설명합니다.

자동 감지

wandb 패키지가 설치되어 있고 Weights & Biases run이 활성 상태이면(wandb.run이 존재하는 경우) 메트릭 로깅이 자동으로 활성화됩니다. 실험 추적을 사용하려면 선택 패키지를 설치하고 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 결과는 자동으로 추적되므로 수동으로 호출할 필요 없음
        result = sandbox.exec(["python", "-c", code]).result()
        # 상관관계 분석을 위해 현재 트레이닝 step의 메트릭을 로깅
        session.log_metrics(step=step)
# 세션이 종료되면 최종 메트릭이 로깅됨

명시적 제어

report_to 매개변수로 메트릭 보고를 제어하세요:

# 명시적 활성화 (리포터 생성; wandb.run이 존재하면 메트릭 로깅)
session = Session(defaults, report_to=["wandb"])

# 보고 비활성화 (wandb run이 존재해도 비활성화)
session = Session(defaults, report_to=[])

# 자동 감지 (기본 동작)
session = Session(defaults, report_to=None)

메트릭

exec()가 완료되면 실행 메트릭이 자동으로 추적됩니다:

메트릭	설명
`cwsandbox/sandboxes_created`	세션을 통해 생성된 총 샌드박스 수
`cwsandbox/executions`	총 exec() Call 수
`cwsandbox/exec_completed_ok`	완료된 실행 (returncode=0)
`cwsandbox/exec_completed_nonzero`	완료된 실행 (returncode!=0)
`cwsandbox/exec_failures`	실패한 실행 (timeouts, transport 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()

    # 이 트레이닝 step에서 메트릭을 로깅하여 상관관계 분석
    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() Call 수(“롤아웃당 도구 Call 수” 파악에 유용)
`cwsandbox/min_execs_per_sandbox`	모든 샌드박스 중 최소 exec() Call 수
`cwsandbox/max_execs_per_sandbox`	모든 샌드박스 중 최대 exec() Call 수

이 메트릭으로 파악할 수 있는 에이전트 동작:

높은 avg_execs_per_sandbox는 에피소드당 도구 Call이 많은 장황한 에이전트를 나타낼 수 있습니다.
큰 편차(max-min)는 에피소드 간 롤아웃 동작이 일관되지 않음을 나타낼 수 있습니다.
학습 단계에 따른 추세는 정책이 개선되면서 에이전트 동작이 어떻게 변하는지 보여줍니다.

대시보드 사용 예시:

학습 단계에 따른 avg_execs_per_sandbox를 그려 트레이닝 중 도구 사용 추세를 확인하세요.
max_execs_per_sandbox가 임곗값을 초과하면 알림을 보내세요(통제되지 않는 에이전트가 과도하게 도구 Call을 수행하는 경우).
최소/최대 범위 차이를 비교하여 에이전트가 반복문에 갇히는 에피소드와 빠르게 완료하는 에피소드를 구분하세요.

기본적으로 log_metrics()는 로깅한 후 카운터를 재설정합니다. 계속 누적하려면 reset=False를 설정하세요:

session.log_metrics(step=step, reset=False)  # 계속 누적

세션이 종료될 때도 메트릭이 자동으로 로깅되므로, 명시적으로 로깅하지 않아도 최종 summary 메트릭을 조회할 수 있습니다.

모니터링 및 디버깅

다음 패턴을 사용하면 트레이닝 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),
    }

실행 세부 정보 로깅하기

보상 계산을 디버깅하기 위해 실행 세부 정보를 캡처하세요:

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 예제는 단일 샷 실행을 위해 샌드박스를 사용합니다: 완료당 하나의 샌드박스, 한 번 실행, 보상 반환. 이는 한 번의 시도로 올바른 코드를 생성하도록 모델을 트레이닝하는 데 적합합니다. 상태 저장 다단계 롤아웃은 다릅니다: 에이전트는 단일 샌드박스 내에서 여러 액션을 수행하며, 샌드박스는 액션 간 상태를 유지합니다. 에이전트는 파일을 작성하고, 실행하고, 오류를 관찰하고, 파일을 편집하고, 다시 사용해 볼 수 있습니다. 이 모든 것이 동일한 샌드박스 내에서 이루어집니다. examples/rl_training/art/ 디렉터리는 MBPP 벤치마크에서 이 패턴을 보여줍니다. 솔루션이 실패하면 에이전트는 오류 피드백을 받고 접근 방식을 반복할 수 있습니다.

개요

이 섹션의 나머지 부분에서는 ART의 개요, 이 예제에서 사용하는 트레이닝 방식, 사전 요구 사항, 설치, 예제 실행 방법, 그리고 각 컴포넌트가 어떻게 맞물려 동작하는지 설명합니다. ART (Agent Reinforcement Trainer)는 OpenPipe가 개발한 오픈 소스 RL 프레임워크로, GRPO를 사용해 다단계 에이전트를 트레이닝합니다. 이 예제는 CoreWeave 샌드박스를 ART와 통합합니다.

트래젝터리 수집 및 트레이닝에 art 패키지(openpipe-art)를 사용합니다.
LocalBackend(GPU 필요)와 TinkerBackend(GPU 불필요)의 두 가지 백엔드를 지원합니다.
CoreWeave 샌드박스에서 도구 호출을 통해 코드를 실행합니다.
MBPP 테스트 케이스 결과를 기반으로 이진 보상을 계산합니다.

GRPO와 증류를 사용한 트레이닝 방식

이 예제는 강화 학습과 증류를 함께 사용합니다. 더 강력한 모델이 시연 데이터를 생성하면, 더 작은 모델이 그중 성공한 사례를 재현하도록 학습합니다. 이 예제에는 두 가지 모델이 사용됩니다:

추론 모델 (--model, 기본값: gpt-5.1-codex-mini): 롤아웃 중에 트래젝터리를 생성합니다. 이 모델은 도구 Call을 수행하고, 샌드박스 실행 결과를 확인하며, 오류를 반복적으로 수정한 뒤 솔루션을 제출합니다. 이 모델 자체는 트레이닝되지 않습니다.
기본 모델 (--base-model, 기본값: Qwen/Qwen3-8B): 트레이닝 대상 모델입니다. 추론 모델이 생성한 트래젝터리를 받아 GRPO(Group Relative Policy Optimization)로 학습합니다.

작동 방식:

추론 모델은 문제마다 여러 트래젝터리를 생성하며, 각 트래젝터리의 도구 Call은 CoreWeave 샌드박스에서 실행됩니다.
각 트래젝터리는 이진 보상을 받습니다. 테스트를 통과하면 1.0, 그렇지 않으면 0.0입니다.
같은 문제에 대한 트래젝터리들은 하나의 그룹을 이루며, GRPO는 각 그룹 안에서 트래젝터리를 서로 비교합니다.
기본 모델(Qwen3-8B)은 보상이 낮은 트래젝터리보다 보상이 높은 트래젝터리를 선호하도록 트레이닝됩니다.

트레이닝이 끝나면 동일한 도구 정의로 Qwen3-8B를 배포합니다. 이 모델은 추론 모델의 성공한 트래젝터리를 모방하면서 유사한 도구 Call을 수행하는 법을 학습한 상태입니다.

사전 요구 사항

Mode	Requirements
Dry run	CPU only
TinkerBackend	CPU only (API를 통한 트레이닝)
LocalBackend	GPU required

환경 변수:

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}
...

설정 옵션

Flag	Default	설명
`--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-compatible API 기본 URL
`--project`	`cwsandbox-mbpp`	Weights & Biases 프로젝트 이름
`--run-name`	`train-001`	트레이닝 run 이름
`--learning-rate`	`1e-5`	학습률
`--dry-run`	`false`	트레이닝 없이 설정 검증

아키텍처

art/
├── train.py         # 트레이닝 루프, CLI, ART backend 설정
├── 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,              # Binary reward
    metadata={"task_id": problem.task_id},
)
return trajectory.finish()

도구 호출 패턴: 롤아웃은 OpenAI-compatible 도구 호출을 사용하여 두 가지 도구를 사용합니다:

execute_code: 샌드박스에서 코드를 테스트하고 stdout/stderr를 반환합니다.
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 도구...
]

샌드박스에는 tracking을 위한 태그가 지정됩니다:

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

데이터 흐름

트레이닝 파이프라인에는 여러 컴포넌트가 있습니다:

로컬 머신: 트레이닝 스크립트(train.py)가 사용자 머신이나 트레이닝 서버에서 실행됩니다. MBPP 데이터셋에서 문제를 로드하고 트레이닝 루프를 조율합니다.
OpenAI API (추론): 트래젝터리 수집 중에, 롤아웃 코드가 OpenAI API(또는 호환 endpoint)를 호출하여 모델 응답을 생성합니다. 모델은 도구 정의를 받고 롤아웃이 실행하는 도구 Call을 반환합니다.
CoreWeave 샌드박스 (코드 실행): 각 롤아웃은 모든 도구 Call에 걸쳐 유지되는 단일 CoreWeave 샌드박스를 사용합니다. 모델이 execute_code 또는 submit_solution을 호출하면 해당 샌드박스에서 코드가 실행됩니다. 이는 에이전트가 반복하면서 파일 변경과 상태가 누적된다는 의미입니다. 파일을 작성하고, 실행하고, 오류를 관찰한 후 수정할 수 있습니다. 샌드박스는 격리를 제공하여 신뢰할 수 없는 모델 생성 코드가 호스트에 영향을 미치지 못하게 합니다. 결과(stdout, stderr, exit code)는 롤아웃으로 다시 전달됩니다.
트래젝터리 수집: 롤아웃은 대화 이력(메시지와 도구 결과)과 최종 보상을 art.Trajectory 객체에 누적합니다. 동일한 문제에 대한 여러 트래젝터리는 art.TrajectoryGroup을 형성합니다.
트레이닝 백엔드: 수집된 트래젝터리 그룹은 트레이닝 백엔드로 전송됩니다. LocalBackend를 사용하면 로컬 GPU에서 트레이닝이 수행됩니다. TinkerBackend를 사용하면 트래젝터리가 Thinking Machines의 Tinker 서비스로 업로드되어 원격으로 트레이닝을 처리합니다. 로컬 GPU가 필요하지 않습니다.

샌드박스

샌드박스 사용하기

샌드박스 관리

레퍼런스

에이전트 도구 실행에 샌드박스를 사용하는 이유

사전 요구 사항

핵심 패턴

에피소드를 병렬로 실행하는 트레이닝 단계

작업 메타데이터를 위한 태그 지정

보상 함수 예시 사용해 보기

TRL GRPOTrainer 인테그레이션

TRL GRPO 인테그레이션 예시 사용해 보기

에이전트 에피소드의 오류 처리

Weights & Biases 메트릭 인테그레이션

자동 감지

명시적 제어

메트릭

샌드박스별 exec 메트릭

모니터링 및 디버깅

활성 샌드박스 수 세기

실행 세부 정보 로깅하기

ART를 사용한 다단계 롤아웃

개요

GRPO와 증류를 사용한 트레이닝 방식

사전 요구 사항

설치

예시 실행

설정 옵션

아키텍처

데이터 흐름

​에이전트 도구 실행에 샌드박스를 사용하는 이유

​사전 요구 사항

​핵심 패턴

​에피소드를 병렬로 실행하는 트레이닝 단계

​작업 메타데이터를 위한 태그 지정

​보상 함수 예시 사용해 보기

​TRL GRPOTrainer 인테그레이션

​TRL GRPO 인테그레이션 예시 사용해 보기

​에이전트 에피소드의 오류 처리

​Weights & Biases 메트릭 인테그레이션

​자동 감지

​명시적 제어

​메트릭

​샌드박스별 exec 메트릭

​모니터링 및 디버깅

​활성 샌드박스 수 세기

​실행 세부 정보 로깅하기

​ART를 사용한 다단계 롤아웃

​개요

​GRPO와 증류를 사용한 트레이닝 방식

​사전 요구 사항

​설치

​예시 실행

​설정 옵션

​아키텍처

​데이터 흐름

에이전트 도구 실행에 샌드박스를 사용하는 이유

사전 요구 사항

핵심 패턴

에피소드를 병렬로 실행하는 트레이닝 단계

작업 메타데이터를 위한 태그 지정

보상 함수 예시 사용해 보기

TRL GRPOTrainer 인테그레이션

TRL GRPO 인테그레이션 예시 사용해 보기

에이전트 에피소드의 오류 처리

Weights & Biases 메트릭 인테그레이션

자동 감지

명시적 제어

메트릭

샌드박스별 exec 메트릭

모니터링 및 디버깅

활성 샌드박스 수 세기

실행 세부 정보 로깅하기

ART를 사용한 다단계 롤아웃

개요

GRPO와 증류를 사용한 트레이닝 방식

사전 요구 사항

설치

예시 실행

설정 옵션

아키텍처

데이터 흐름