Skip to content
View as Markdown

Tinker Quickstart

Installation

pip install tinker

Authentication

tinker auth login

Or create an API key in the Tinker Console and set it as the TINKER_API_KEY environment variable.

Billing

Set up payment info in Billing, and see Models & Pricing for current training and sampling rates.

Using Tinker SDK

Create clients

View active sessions in Sessions in the Tinker Console. 

import tinker

service_client = tinker.ServiceClient()

# Perform forward/backward passes and optimization
training_client = service_client.create_lora_training_client(
    base_model="Qwen/Qwen3-8B", rank=16
)

# For generating completions from a base model
sampling_client = service_client.create_sampling_client(base_model="Qwen/Qwen3-8B")

# Convert text to/from tokens
tokenizer = training_client.get_tokenizer()

Sampling

prompt = "The capital of France is"
prompt_tokens = tokenizer.encode("The capital of France is")
model_input = tinker.types.ModelInput.from_ints(prompt_tokens)
params = tinker.types.SamplingParams(max_tokens=50, temperature=0.7, stop=["\n"])

# Decode the returned sample
result = await sampling_client.sample_async(
    prompt=model_input, num_samples=1, sampling_params=params
)
response = tokenizer.decode(result.sequences[0].tokens)
print(response)

# For a conversation-formatted prompt
conversation_tokens = tokenizer.apply_chat_template(
    conversation=[{"role": "user", "content": "What is the capital of France?"}],
    add_generation_prompt=True,
)
model_input = tinker.types.ModelInput.from_ints(conversation_tokens)
result = await sampling_client.sample_async(
    prompt=model_input, num_samples=1, sampling_params=params
)
response = tokenizer.decode(result.sequences[0].tokens)
print(response)

Training

See Loss Functions for details about available loss functions, LoRA Primer for details about training with LoRA, and Prepare Training Data for guidance on constructing Datum objects.

Supervised Fine-tuning (SFT)

completion_tokens = result.sequences[0].tokens
full_sequence = prompt_tokens + completion_tokens
n_prefix = len(prompt_tokens) - 1
datum = tinker.types.Datum(
    input=model_input,
    loss_fn_inputs = {
        # 0 to apply no loss, 1 to apply loss to token
        weights=[0.0] * n_prefix + [1.0] * len(completion_tokens),
        target_tokens=full_sequence[1:],  # shifted by 1 from input
    }
)

# Gradient computation
fwd_bwd_result = training_client.forward_backward_async(
    data=[datum],
    loss_fn="cross_entropy"
)
# Update model parameters
optim_future = await training_client.optim_step_async(
    tinker.types.AdamParams(learning_rate=1e-4)
)

await fwd_bwd_result.result_async()
await optim_future.result_async()

Reinforcement Learning (RL)

rewards = [reward_fn(seq) for seq in result.sequences]
mean = sum(rewards) / len(rewards)
datums = []

for seq, reward in zip(result.sequences, rewards):
    completion_tokens = seq.tokens
    n_prefix = len(prompt_tokens) - 1
    advantage = reward - mean
    full_sequence = prompt_tokens + completion_tokens
    datum = tinker.types.Datum(
        input=model_input,
        loss_fn_inputs={
            "target_tokens": full_sequence[1:],
            # Selected token log probabilities in rollout
            # 0 for prompt tokens
            "logprobs": [0.0] * n_prefix + seq.logprobs,
            # per token advantages, 0 for prompt tokens
            "advantages": [0.0] * n_prefix + [advantage] * len(completion_tokens),
        },
    )
    datums.append(datum)

fwd_bwd_result = await training_client.forward_backward_async(
    data=datums,
    loss_fn="importance_sampling",  # "ppo", "cispo", "dro"
)
optim_future = await training_client.optim_step_async(
    tinker.types.AdamParams(learning_rate=1e-4)
)

await fwd_bwd_result.result_async()
await optim_future.result_async()

Save and Load Checkpoints

To view all saved checkpoints, go to Checkpoints in the Tinker Console.

# Save a checkpoint that can be used to resume training
training_checkpoint = await training_client.save_state_async(
    name="my-checkpoint",
    user_metadata={
        "step": 1000,
    },
)

# Load the training checkpoint
service_client.create_training_client_from_state(training_checkpoint.path)

# Save a checkpoint that can be used for sampling
sampling_checkpoint = await training_client.save_weights_for_sampler_async(
    name="my-sampler-checkpoint"
)

# Load the sampling checkpoint
service_client.create_sampling_client(model_path=sampling_checkpoint.path)


# Open in the Tinker Playground
print(sampling_checkpoint.get_playground_url())

Additional Resources