Training Gemma-3 for Structured Mathematical Reasoning with Tunix GRPO, LoRA Adapters, and GSM8K Rewards

import importlib.util, os, shutil as _sh
if importlib.util.find_spec("tunix") is None:
   print("Installing Tunix + JAX ecosystem — this takes ~5-8 min…")
   %pip install -q ipywidgets tensorboardX transformers grain nest_asyncio
   %pip install -q datasets huggingface_hub "numpy>2"
   %pip install -q tensorflow tensorflow_datasets
   %pip install -q git+https://github.com/jax-ml/jax
   %pip install -q git+https://github.com/google/tunix
   %pip install -q git+https://github.com/google/qwix
   %pip uninstall -q flax -y
   %pip install -q git+https://github.com/google/flax
   %pip uninstall -q wandb -y
   print("\n\n Install done. The runtime will RESTART now.")
   print("  After it restarts, RUN THIS CELL AGAIN to start training.\n")
   os.kill(os.getpid(), 9)
import getpass, functools, json, re
import numpy as np
os.environ["WANDB_MODE"] = "disabled"
os.environ["TOKENIZERS_PARALLELISM"] = "false"
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
HF_TOKEN = os.environ.get("HF_TOKEN")
if not HF_TOKEN:
   try:
       from google.colab import userdata
       HF_TOKEN = userdata.get("HF_TOKEN")
   except Exception:
       HF_TOKEN = getpass.getpass("Hugging Face token (needs Gemma license access): ")
os.environ["HF_TOKEN"] = HF_TOKEN or ""
import nest_asyncio; nest_asyncio.apply()
import tensorflow as tf; tf.config.set_visible_devices([], "GPU")
import jax, jax.numpy as jnp, optax, grain, qwix
from flax import nnx
from orbax import checkpoint as ocp
from huggingface_hub import snapshot_download, login
from datasets import load_dataset
from tunix.generate import sampler as sampler_lib
from tunix.generate import tokenizer_adapter as tokenizer_lib
from tunix.models.gemma3 import model as gemma_lib
from tunix.models.gemma3 import params_safetensors as params_safetensors_lib
from tunix.models.gemma3 import params as gemma_params
from tunix.rl import rl_cluster as rl_cluster_lib
from tunix.rl.grpo.grpo_learner import GRPOConfig, GRPOLearner
from tunix.rl.rollout import base_rollout
from tunix.sft import metrics_logger
if HF_TOKEN:
   login(token=HF_TOKEN)
devices = jax.devices()
print("JAX backend :", jax.default_backend(), "|", len(devices), "device(s):", devices)
IS_GPU = _sh.which("nvidia-smi") is not None
if IS_GPU and jax.default_backend() == "cpu":
   print("  GPU runtime but JAX sees CPU only. Re-run `pip install -U \"jax[cuda12]\"` "
         "and restart, or switch to a TPU runtime.")
MODEL_ID = "google/gemma-3-1b-it"
RANK, ALPHA = 32, 32.0
MAX_PROMPT_LENGTH      = 256
TOTAL_GENERATION_STEPS = 512
NUM_GENERATIONS        = 2
NUM_ITERATIONS         = 1
BETA                   = 0.08
EPSILON                = 0.2
TEMPERATURE, TOP_P, TOP_K = 0.9, 1.0, 50
MAX_STEPS   = 100
TRAIN_LIMIT = MAX_STEPS
NUM_TEST    = 16
LEARNING_RATE, B1, B2, WEIGHT_DECAY = 3e-6, 0.9, 0.99, 0.1
WARMUP_STEPS, MAX_GRAD_NORM = int(0.1 * MAX_STEPS), 0.1
CKPT_DIR, TB_DIR = "/content/ckpts/", "/content/tb/grpo"
N = jax.device_count()
MESH = [(N, 1), ("fsdp", "tp")]
mesh = jax.make_mesh(*MESH, axis_types=(jax.sharding.AxisType.Auto,) * 2)

reasoning_start, reasoning_end = "<reasoning>", "</reasoning>"

solution_start,  solution_end  = "<answer>",   "</answer>"

SYSTEM_PROMPT = (f"You are given a problem. First, think about the problem and provide your "

f"reasoning between {reasoning_start} and {reasoning_end}. Then give the final "

f"answer (just one number) between {solution_start} and {solution_end}.")

TEMPLATE = ("<start_of_turn>user\n{system_prompt}\n\n{question}<end_of_turn>\n<start_of_turn>model\n")

def extract_hash_answer(text):

return text.split("####")[1].strip() if "####" in text else None

gsm = load_dataset("openai/gsm8k", "main")

def build_grain(split, limit):

rows = [{"question": r["question"], "answer": r["answer"]}

for r in split.select(range(min(limit, len(split))))]

return (grain.MapDataset.source(rows).shuffle(seed=42).map(

lambda x: {"prompts":  TEMPLATE.format(system_prompt=SYSTEM_PROMPT, question=x["question"]),

"question": x["question"],

"answer":   extract_hash_answer(x["answer"])}))

train_dataset = build_grain(gsm["train"], TRAIN_LIMIT).batch(1)[:MAX_STEPS]

val_dataset   = None

test_rows     = [{"question": r["question"], "answer": extract_hash_answer(r["answer"])}

for r in gsm["test"].select(range(NUM_TEST))]

print(f"Train batches: {len(train_dataset)} | Test examples: {len(test_rows)}")

match_format = re.compile(

rf"^[\s]{{0,}}{reasoning_start}.+?{reasoning_end}.*?{solution_start}(.+?){solution_end}[\s]{{0,}}$",

flags=re.MULTILINE | re.DOTALL)

match_numbers = re.compile(rf"{solution_start}.*?([\d\.]{{1,}})", flags=re.MULTILINE | re.DOTALL)

def match_format_exactly(prompts, completions, **kw):

return [3.0 if match_format.search(c) else 0.0 for c in completions]

def match_format_approximately(prompts, completions, **kw):

out = []

for c in completions:

s = 0.0

s += 0.5 if c.count(reasoning_start) == 1 else -0.5

s += 0.5 if c.count(reasoning_end)   == 1 else -0.5

s += 0.5 if c.count(solution_start)  == 1 else -0.5

s += 0.5 if c.count(solution_end)    == 1 else -0.5

out.append(s)

return out

def check_answer(prompts, completions, answer, **kw):

guesses = [(m.group(1) if (m := match_format.search(c)) else None) for c in completions]

scores = []

for g, a in zip(guesses, answer):

if g is None:

scores.append(0.0); continue

if g == a:               scores.append(3.0)

elif g.strip() == a.strip(): scores.append(1.5)

else:

try:

r = float(g) / float(a)

scores.append(0.5 if 0.9 <= r <= 1.1 else 0.25 if 0.8 <= r &
Source Read original →
Related reading
NVIDIA AI Introduce SpatialClaw: A Training-Free Agent That Treats Code as the Action Interface for Spatial Reasoning
VibeThinker-3B: A 3B Dense Reasoning Model Built on Qwen2.5-Coder-3B With the Spectrum-to-Signal Post-Training Pipeline
If you use Google, you’re training its AI. Here’s how to opt out.