"""OpenSleuth GRPO trainer. Trains a small Qwen2.5 model with TRL's GRPOTrainer to do in-context program synthesis — given the public signature of a hidden function plus a handful of (input, output) probe examples, emit a Python function that reproduces it. Reward comes from the live OpenSleuth env Space: the agent's code is executed against the hidden reference under domain-aware fuzzing, and the verifier returns an `execution_reward - complexity_penalty` score that we hand back to GRPO as the per-completion reward (plus a tiny formatting shaping reward). """ from __future__ import annotations import argparse import logging import os import sys import time import torch from peft import LoraConfig from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig from trl import GRPOConfig, GRPOTrainer from opensleuth_train import ( EnvClient, SYSTEM_PROMPT, build_synthesis_dataset, discover_functions, ) from opensleuth_train.reward import format_reward, make_env_reward logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", stream=sys.stdout, ) log = logging.getLogger("opensleuth.train") def _split_csv(s: str) -> list[str]: return [x.strip() for x in s.split(",") if x.strip()] def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser() p.add_argument("--env-url", default=os.environ.get("ENV_URL", "https://anugrah55-opensleuth-env-gemini-cli.hf.space")) # v0.4 default: switch to Qwen2.5-3B-Instruct for the open-ended task pool. # The 0.5B baseline saturated easy tasks but couldn't solve the hard / # Hub-driven ones. 3B + LoRA + 4-bit fits T4-small (16GB). p.add_argument("--model-name", default=os.environ.get("MODEL_NAME", "Qwen/Qwen2.5-3B-Instruct")) p.add_argument("--output-dir", default=os.environ.get("OUTPUT_DIR", "/data/opensleuth-grpo")) p.add_argument( "--push-to-hub", default=os.environ.get( "PUSH_TO_HUB", "anugrah55/opensleuth-qwen2.5-3b-grpo-v2" ), ) # Task selection / curriculum knobs (v0.4). p.add_argument( "--functions", default=os.environ.get("FUNCTIONS_INCLUDE", ""), help="Comma-separated subset of task names to train on. Empty = all " "tasks the env exposes (builtin + Hub).", ) p.add_argument( "--difficulty", default=os.environ.get("DIFFICULTY_FILTER", "all"), choices=["easy", "medium", "hard", "all"], help="Curriculum filter: only sample tasks at this difficulty level.", ) # Difficulty-weighted rollout counts. Replaces the v0.3 single # n-per-function knob (kept as an optional override). p.add_argument("--n-easy", type=int, default=int(os.environ.get("N_EASY", "8"))) p.add_argument("--n-medium", type=int, default=int(os.environ.get("N_MEDIUM", "16"))) p.add_argument("--n-hard", type=int, default=int(os.environ.get("N_HARD", "24"))) p.add_argument( "--n-per-function", type=int, default=int(os.environ.get("N_PER_FUNCTION", "0")), help="If >0, overrides per-difficulty rollout counts with a uniform N.", ) p.add_argument("--n-probes", type=int, default=int(os.environ.get("N_PROBES", "6"))) # GRPO/optim defaults sized for T4-small (16GB) + Qwen2.5-3B-4bit + LoRA. p.add_argument("--num-generations", type=int, default=int(os.environ.get("NUM_GENERATIONS", "2"))) p.add_argument("--max-completion-length", type=int, default=int(os.environ.get("MAX_COMPLETION_LENGTH", "384"))) p.add_argument("--max-prompt-length", type=int, default=int(os.environ.get("MAX_PROMPT_LENGTH", "1024"))) p.add_argument("--learning-rate", type=float, default=float(os.environ.get("LEARNING_RATE", "1e-5"))) p.add_argument("--num-train-epochs", type=float, default=float(os.environ.get("NUM_TRAIN_EPOCHS", "1"))) # GRPO requires per_device_train_batch_size to be a multiple of num_generations # (one prompt is repeated num_generations times, all in the same forward pass). p.add_argument("--per-device-batch-size", type=int, default=int(os.environ.get("PER_DEVICE_BATCH_SIZE", "2"))) p.add_argument("--gradient-accumulation-steps", type=int, default=int(os.environ.get("GRAD_ACCUM", "4"))) p.add_argument("--no-4bit", action="store_true", default=os.environ.get("NO_4BIT", "0") == "1") p.add_argument("--seed", type=int, default=int(os.environ.get("SEED", "42"))) return p.parse_args() def wait_for_env(client: EnvClient, max_wait_s: float = 300.0) -> None: log.info("waiting for env at %s ...", client.base_url) start = time.time() last_err = "" while time.time() - start < max_wait_s: try: h = client.health() log.info("env healthy: %s", h) return except Exception as e: # noqa: BLE001 last_err = str(e) time.sleep(5) raise RuntimeError(f"env never became healthy after {max_wait_s}s. Last error: {last_err}") def main() -> int: args = parse_args() log.info("args: %s", vars(args)) client = EnvClient(base_url=args.env_url, timeout=60.0, retries=4) wait_for_env(client) include = _split_csv(args.functions) if args.functions else None difficulty = None if args.difficulty == "all" else args.difficulty tasks = discover_functions(client, include=include, difficulty=difficulty) log.info( "env catalog: %d task(s) after filter (include=%s, difficulty=%s):", len(tasks), include, difficulty, ) for t in tasks: log.info( " - %-22s difficulty=%-8s source=%s", t["name"], t.get("difficulty"), t.get("source"), ) n_per_function_override = args.n_per_function if args.n_per_function > 0 else None log.info( "building synthesis dataset (n_easy=%d n_medium=%d n_hard=%d override=%s n_probes=%d)", args.n_easy, args.n_medium, args.n_hard, n_per_function_override, args.n_probes, ) dataset = build_synthesis_dataset( client, tasks=tasks, n_per_function=n_per_function_override, n_easy=args.n_easy, n_medium=args.n_medium, n_hard=args.n_hard, n_probes=args.n_probes, seed=args.seed, ) log.info("dataset size: %d rows", len(dataset)) # GRPO with chat-templated prompts: each row needs a "prompt" field, which # we re-format as a chat message list so the trainer applies the chat # template under the hood. def to_chat(row): return { "prompt": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": row["prompt"]}, ], "target_function_name": row["target_function_name"], "row_seed": row["row_seed"], } # Drop the human-readable difficulty column from the GRPO-visible map so # the trainer doesn't try to forward it as a reward-fn kwarg. drop_cols = [c for c in ("prompt", "difficulty") if c in dataset.column_names] dataset = dataset.map(to_chat, remove_columns=drop_cols) # ---- Model + LoRA ---- log.info("loading model %s (4bit=%s)", args.model_name, not args.no_4bit) bnb_config = None if not args.no_4bit: bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16, bnb_4bit_use_double_quant=True, bnb_4bit_quant_type="nf4", ) tokenizer = AutoTokenizer.from_pretrained(args.model_name, trust_remote_code=True) if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token peft_config = LoraConfig( r=16, lora_alpha=32, lora_dropout=0.05, target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], task_type="CAUSAL_LM", bias="none", ) # GRPO requires per_device_train_batch_size to be a multiple of num_generations. # If the caller didn't pin one explicitly, default to one prompt per device. per_device_bs = args.per_device_batch_size or args.num_generations if per_device_bs % args.num_generations != 0: raise ValueError( f"per_device_batch_size ({per_device_bs}) must be a multiple of " f"num_generations ({args.num_generations})." ) log.info( "GRPO batching: per_device_batch_size=%d (= %d prompt(s) × %d generations), grad_accum=%d", per_device_bs, per_device_bs // args.num_generations, args.num_generations, args.gradient_accumulation_steps, ) grpo_config = GRPOConfig( output_dir=args.output_dir, per_device_train_batch_size=per_device_bs, gradient_accumulation_steps=args.gradient_accumulation_steps, learning_rate=args.learning_rate, num_train_epochs=args.num_train_epochs, max_prompt_length=args.max_prompt_length, max_completion_length=args.max_completion_length, num_generations=args.num_generations, beta=0.04, bf16=torch.cuda.is_bf16_supported() if torch.cuda.is_available() else False, fp16=False, logging_steps=1, save_steps=50, save_total_limit=2, report_to=[], seed=args.seed, push_to_hub=bool(args.push_to_hub) and bool(os.environ.get("HF_TOKEN")), hub_model_id=args.push_to_hub or None, hub_strategy="end", gradient_checkpointing=True, ) env_reward_fn = make_env_reward(client) env_reward_fn.__name__ = "env_verifier_reward" format_reward.__name__ = "format_reward" # Load the model ourselves so we control quantization + dtype precisely. # GRPOTrainer in 0.16 takes model objects and passes them through to its # internal ref-model copy + LoRA wrapping. log.info("loading base model with quantization=%s", bnb_config is not None) model_kwargs = {"trust_remote_code": True, "torch_dtype": torch.bfloat16} if bnb_config is not None: model_kwargs["quantization_config"] = bnb_config model = AutoModelForCausalLM.from_pretrained(args.model_name, **model_kwargs) log.info("instantiating GRPOTrainer") trainer = GRPOTrainer( model=model, reward_funcs=[env_reward_fn, format_reward], args=grpo_config, train_dataset=dataset, peft_config=peft_config, processing_class=tokenizer, ) log.info("starting GRPO training") trainer.train() log.info("training complete; saving to %s", args.output_dir) trainer.save_model(args.output_dir) if grpo_config.push_to_hub: log.info("pushing to hub: %s", args.push_to_hub) trainer.push_to_hub() return 0 if __name__ == "__main__": sys.exit(main())