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.
Open recipeReward proofs that verify.
Open recipeReward calibrated probabilities.
Open recipe
Tinker…importance_samplingformal-reasoning-deepseek-v3-1-environment.pyformal-reasoning-deepseek-v3-1-environment.pyRunnable loop for Formal Reasoning Engines on DeepSeek-V3.1.
LoRA r=32, 50 steps, Importance Sampling loss.
"""Reinforcement.tech compiled Tinker loop H1: Reinforcement: Build Your Own Reward Model INPUT Signal : Verifiable outcome (Environment RL) In : A verifier — Lean, pytest, compiler, retrieval set. Task : Formal Reasoning Engines — Checkable traces Model : deepseek-ai/DeepSeek-V3.1 (MOE, large MoE) OUTPUT Loop : Runnable environment loop. Loss : importance_sampling 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 = "deepseek-ai/DeepSeek-V3.1" 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() GROUP_SIZE = 8 PROMPT = "Prove the statement in Lean 4. Return only a complete proof." # Pattern: Checkable traces. Renderer family hint: deepseek. def verify_environment(completion: str) -> float: """Binary environment reward. Swap in `lean --make` or your kernel.""" text = completion.lower() has_qed = "qed" in text or "sorry" not in text has_structure = "theorem" in text or "by" in text return 1.0 if has_qed and has_structure else 0.0 def pack_rl_datum( tokenizer, prompt: str, completion_tokens: list[int], sampling_logprobs: list[float], advantage: float, ) -> types.Datum: prompt_tokens = tokenizer.encode(prompt) 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:], logprobs=[0.0] * n_prefix + list(sampling_logprobs), advantages=[0.0] * n_prefix + [advantage] * len(completion_tokens), ), ) def rollout_logprobs(sequence) -> list[float]: logprobs = sequence.logprobs if getattr(sequence, "logprobs", None) else None if logprobs is None: return [0.0] * len(sequence.tokens) return list(logprobs) async def train() -> None: require_key() _service, training, tokenizer = await connect() params = types.SamplingParams(max_tokens=256, temperature=0.8) prompt = types.ModelInput.from_ints(tokenizer.encode(PROMPT)) for step in range(STEPS): sampling = await sampling_client(training) rollout = await sampling.sample_async( prompt=prompt, num_samples=GROUP_SIZE, sampling_params=params, ) rewards: list[float] = [] datums: list[types.Datum] = [] for sequence in rollout.sequences: text = tokenizer.decode(sequence.tokens) rewards.append(verify_environment(text)) baseline = sum(rewards) / max(len(rewards), 1) if all(reward == rewards[0] for reward in rewards): print(f"step {step:03d} skip degenerate group R={baseline:.3f}") continue for sequence, reward in zip(rollout.sequences, rewards): advantage = reward - baseline datums.append( pack_rl_datum( tokenizer, PROMPT, sequence.tokens, rollout_logprobs(sequence), advantage, ) ) fwdbwd = await training.forward_backward_async(datums, loss_fn="importance_sampling") optim = await training.optim_step_async(types.AdamParams(learning_rate=LEARNING_RATE)) await fwdbwd.result_async() await optim.result_async() await checkpoint(training, step) print(f"step {step:03d} R={baseline:.3f} datums={len(datums)}") sampling = await sampling_client(training) preview = await sampling.sample_async(prompt=prompt, num_samples=1, sampling_params=params) print("sample:", tokenizer.decode(preview.sequences[0].tokens)) if __name__ == "__main__": asyncio.run(train())