Choose a verifier, a task, and a base model. Download a ready-to-run training loop.
Loading the builder…
Reward cited tool calls and thrift.
Source: Waldo on Tinker Open recipeReward proofs that verify.
Source: AxiomProver on Tinker Open recipeReward calibrated probabilities.
Source: Mantic on Tinker Open recipe
Tinker…cross_entropy on on-policy rewritestool-agents-nemotron-3-nano-sdft.pytool-agents-nemotron-3-nano-sdft.pyRunnable loop for Specialized Tool Agents on Nemotron-3-Nano.
LoRA r=32, 50 steps, Cross-Entropy (SDFT) loss.
"""Reinforcement.tech compiled Tinker loop H1: Reinforcement: Build Your Own Reward Model INPUT Signal : Demonstrations (Self-distillation (SDFT)) In : Expert demonstrations you want to keep. Task : Specialized Tool Agents — Search / tool-use trajectories Model : nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16 (MOE, 30B-A3B) OUTPUT Loop : Runnable distillation loop. Loss : cross_entropy on on-policy SDFT rewrites LoRA rank : 32 Steps : 50 Tinker primitives used in this file: sample generate on-policy rollouts forward_backward accumulate LoRA gradients optim_step Adam update on the adapter save_state checkpoint weights + optimizer Requires: uv pip install tinker export TINKER_API_KEY=... Docs: https://tinker-docs.thinkingmachines.ai/tinker/quickstart/ """ from __future__ import annotations import asyncio import os from pathlib import Path import tinker from tinker import types BASE_MODEL = "nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16" LORA_RANK = 32 LEARNING_RATE = 5e-4 STEPS = 50 SAVE_EVERY = 10 RUN_NAME = Path(__file__).stem def require_key() -> None: if not os.environ.get("TINKER_API_KEY"): raise SystemExit("Set TINKER_API_KEY before running this loop.") async def connect(): service = tinker.ServiceClient() training = await service.create_lora_training_client_async( base_model=BASE_MODEL, rank=LORA_RANK, user_metadata={"product": "reinforcement.tech", "run": RUN_NAME}, ) tokenizer = training.get_tokenizer() return service, training, tokenizer async def sampling_client(training): """Ephemeral on-policy sampler. Do not pass name= — it is deprecated and ignored.""" return await training.save_weights_and_get_sampling_client_async() async def checkpoint(training, step: int) -> None: if step % SAVE_EVERY != 0 and step != STEPS - 1: return saved = await training.save_state_async(name=f"{RUN_NAME}-step-{step}") await saved.result_async() # Replace DEMO with a demonstration you cannot afford to forget. DEMO = "1) search(query) 2) keep only cited docs 3) answer in 4 bullets." TEACHER_PROMPT = ( "You are the teacher. The expert demonstration is below. " "Rewrite it in your own on-policy style. Keep the same facts and tools.\n\n" f"DEMO:\n{DEMO}" ) STUDENT_PROMPT = "Solve the task. Do not look at a demonstration." def as_sft_datum(tokenizer, prompt: str, completion: str) -> types.Datum: prompt_tokens = tokenizer.encode(prompt) completion_tokens = tokenizer.encode(completion) full = prompt_tokens + completion_tokens n_prefix = max(len(prompt_tokens) - 1, 0) return types.Datum( model_input=types.ModelInput.from_ints(tokens=full[:-1]), loss_fn_inputs=dict( target_tokens=full[1:], weights=[0.0] * n_prefix + [1.0] * len(completion_tokens), ), ) async def train() -> None: require_key() _service, training, tokenizer = await connect() teacher_prompt = types.ModelInput.from_ints(tokenizer.encode(TEACHER_PROMPT)) for step in range(STEPS): sampling = await sampling_client(training) rewrite = await sampling.sample_async( prompt=teacher_prompt, num_samples=1, sampling_params=types.SamplingParams(max_tokens=192, temperature=0.5), ) on_policy = tokenizer.decode(rewrite.sequences[0].tokens) datum = as_sft_datum(tokenizer, STUDENT_PROMPT, on_policy) fwdbwd = await training.forward_backward_async([datum], loss_fn="cross_entropy") optim = await training.optim_step_async(types.AdamParams(learning_rate=LEARNING_RATE)) result = await fwdbwd.result_async() await optim.result_async() await checkpoint(training, step) print(f"step {step:03d} {result.metrics}") sampling = await sampling_client(training) preview = await sampling.sample_async( prompt=types.ModelInput.from_ints(tokenizer.encode(STUDENT_PROMPT)), num_samples=1, sampling_params=types.SamplingParams(max_tokens=128, temperature=0.4), ) print("sample:", tokenizer.decode(preview.sequences[0].tokens)) if __name__ == "__main__": asyncio.run(train())