Programmatic training

Use the Loops SDK to train a model from Python, evaluate its answers, and save checkpoints. Your application chooses the training data and decides when to run another round. This example trains Qwen/Qwen3.5-2B to label support tickets as access or billing issues. It saves a report of the model’s answers so your application can find mistakes and choose data for the next round.

Before you begin

You need Loops access for your workspace, a Baseten API key, and uv. The script uses Python 3.12. uv installs the dependencies listed in the script, including baseten-loops==0.23.2. Use the exact model ID from the supported models list. Check the list for the Base or Instruct version you need. The script also checks which models your workspace supports before creating a trainer. You pay for the GPUs used by the trainer and sampler. The script requests deactivation after each round, including if training or evaluation fails, and records whether cleanup succeeded. Start with one round on a small model before running multiple rounds at once. Create a working directory and set your API key. These commands use macOS or Linux shell syntax:

mkdir loops-campaign
cd loops-campaign
export BASETEN_API_KEY="YOUR_API_KEY"

This example creates independent runs using the default API endpoint and your API key’s workspace. It rejects the SDK overrides LOOPS_REUSE_FROM_RUN_ID, LOOPS_REUSE_FROM_SESSION_ID, LOOPS_BASE_URL, and LOOPS_TEAM. Unset these variables if they’re present in your environment.

Prepare training and evaluation data

Create train.jsonl with prompts and expected answers for training:

train.jsonl

{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: I cannot sign in to my account.\nLabel:","answer":" access"}
{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: My card was charged twice.\nLabel:","answer":" billing"}
{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: Please reset my account password.\nLabel:","answer":" access"}
{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: The invoice shows the wrong price.\nLabel:","answer":" billing"}

Create eval.jsonl with different prompts to evaluate the model:

eval.jsonl

{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: I forgot my login credentials.\nLabel:","answer":" access"}
{"prompt":"Classify the support ticket as access or billing. Respond with only the label.\nTicket: I need a refund for a duplicate payment.\nLabel:","answer":" billing"}

Use the same evaluation examples in each round so you can compare results. Keep a separate test set for the final assessment; don’t use it for training or choosing new training data. The script rejects identical prompts in the training and evaluation files. Also check for examples that repeat the same question in different words.

Build a training round

These snippets show the training and evaluation code inside run_round(). The complete script after the steps includes the imports, data-loading helpers, and cleanup needed to run it.

Create a trainer

Create a session with ServiceClient, then call create_lora_training_client() to start training from the selected base_model. The script reads key from BASETEN_API_KEY:

service = ServiceClient(api_key=key)
trainer = service.create_lora_training_client(
    base_model=base_model, rank=16, seed=42, max_seq_len=MAX_TOKENS,
    ready_timeout=1200, sampler_timeout=120, name=output_path.stem,
)

rank controls the size of the LoRA adapter; this example uses 16. max_seq_len configures a 1,024-token sequence limit, and the script checks each example against that limit. ready_timeout sets how long the client waits for the trainer or sampler to become ready, in seconds. sampler_timeout limits how long it waits for a sampling request.

Measure evaluation loss

Convert the training and evaluation rows into token inputs with the script’s to_datum() helper. Call forward() on the evaluation data to measure loss before training. This call doesn’t update the model:

tokenizer = trainer.get_tokenizer()
train_data = [to_datum(tokenizer, row) for row in train_rows]
eval_data = [to_datum(tokenizer, row) for row in eval_rows]
record["baseline_loss"] = trainer.forward(data=eval_data).result(timeout=600).loss

to_datum() pairs each input token with the next token to predict. It uses -100 to exclude prompt tokens from the loss, so training focuses on the answer. Each Datum contains the input tokens as a ModelInput and the targets as TensorData..result(timeout=600) waits up to 600 seconds for the operation to finish. The report stores the result as baseline_loss for comparison after training.

Train on batches and measure loss again

Call forward_backward() to compute gradients for a batch, then optim_step() to update the model. AdamParams sets the optimizer’s learning_rate. Train on batches of up to eight examples, then measure loss on the same evaluation data:

for _ in range(epochs):
    for start in range(0, len(train_data), 8):
        trainer.forward_backward(data=train_data[start:start + 8]).result(timeout=600)
        trainer.optim_step(AdamParams(learning_rate=4e-5)).result(timeout=600)
record["trained_loss"] = trainer.forward(data=eval_data).result(timeout=600).loss

epochs controls how many times the model trains on the full dataset. The script defaults to two. Compare trained_loss with baseline_loss to check how training changed performance on these examples.

Save a training checkpoint

Call save_state() to save the model weights and optimizer state so you can resume training. name identifies the checkpoint within the run:

record["trainer_checkpoint"] = trainer.save_state(name="final").result(timeout=600).path

The report stores the checkpoint path in trainer_checkpoint.

Prepare the trained weights for sampling

Call save_weights_for_sampler() to save weights for generating answers. Pass the returned path to create_sampling_client() to select those weights:

weights = trainer.save_weights_for_sampler(name="final").result(timeout=600)
record["sampler_checkpoint"] = weights.path
sampler = trainer.create_sampling_client(model_path=weights.path)

The first create_sampling_client() call creates the run’s sampler. The sampler_checkpoint path also lets you deploy the saved weights for inference.

Generate and score answers

Call sample() for each evaluation prompt. num_samples=1 requests one answer. SamplingParams limits the answer to 16 tokens and stops at a newline:

record["evaluations"] = []
for row in eval_rows:
    sample = sampler.sample(
        prompt=ModelInput.from_ints(tokenizer.encode(row["prompt"], add_special_tokens=False)),
        num_samples=1,
        sampling_params=SamplingParams(max_tokens=16, temperature=0, stop=["\n"]),
    )
    completion = tokenizer.decode(sample.sequences[0].tokens).strip()
    record["evaluations"].append({
        "prompt": row["prompt"], "expected": row["answer"].strip(),
        "completion": completion,
        "match": completion.casefold() == row["answer"].strip().casefold(),
    })

The report records each generated answer and whether it matches the expected label. Your application can use incorrect answers to choose training data for the next round.

Save the complete script

Expand the code block, copy the script, and save it as campaign.py. It combines the preceding steps with input validation, report saving, and run deactivation.

Complete campaign.py script

campaign.py

# /// script
# requires-python = ">=3.12"
# dependencies = ["baseten-loops==0.23.2", "requests==2.34.2"]
# ///

"""Runs one train/evaluate round and releases its Loops compute."""

import argparse
import hashlib
import json
import os
from contextlib import ExitStack
from pathlib import Path

import requests
from baseten.loops import (
    AdamParams, Datum, ModelInput, SamplingParams, ServiceClient, TensorData,
)

BASE_MODEL = "Qwen/Qwen3.5-2B"
API_URL = "https://api.baseten.co/v1/loops"
MAX_TOKENS = 1024


def read_examples(path):
    rows = [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()]
    if not rows:
        raise ValueError(f"{path} contains no examples")
    for row in rows:
        if not isinstance(row, dict) or any(
            not isinstance(row.get(key), str) or not row[key].strip()
            for key in ("prompt", "answer")
        ):
            raise ValueError("Each example needs a nonempty prompt and answer")
    return rows


def to_datum(tokenizer, row):
    prompt = tokenizer.encode(row["prompt"], add_special_tokens=False)
    answer = tokenizer.encode(row["answer"], add_special_tokens=False)
    if not prompt or not answer or len(prompt) + len(answer) > MAX_TOKENS:
        raise ValueError("Use nonempty prompts and answers totaling at most 1024 tokens")
    tokens = (prompt + answer)[:-1]
    targets = [-100] * (len(prompt) - 1) + answer
    return Datum(
        model_input=ModelInput.from_ints(tokens),
        loss_fn_inputs={
            "target_tokens": TensorData(data=targets, dtype="int64", shape=[len(targets)])
        },
    )


def api(method, path, key, **kwargs):
    response = requests.request(
        method, f"{API_URL}{path}",
        headers={"Authorization": f"Bearer {key}"}, timeout=30, **kwargs,
    )
    response.raise_for_status()
    return response.json()


def release_session(session_id, key, record):
    # Also find runs created before a provisioning timeout returned a client.
    record["deactivated_run_ids"] = []
    try:
        runs = api("GET", "/runs", key)["runs"]
        owned = [run["id"] for run in runs if run["session_id"] == session_id]
        for run_id in owned:
            api("POST", f"/runs/{run_id}/deactivate", key)
            record["deactivated_run_ids"].append(run_id)
        record["cleanup_status"] = "completed"
    except BaseException as error:
        record["cleanup_status"] = "failed"
        record["cleanup_error_type"] = type(error).__name__
        raise


def write_report(path, record):
    temporary = path.with_suffix(".tmp")
    temporary.write_text(json.dumps(record, indent=2))
    temporary.replace(path)


def run_round(train_path, eval_path, output_path, *, epochs=2, base_model=BASE_MODEL):
    overrides = [name for name in (
        "LOOPS_REUSE_FROM_RUN_ID", "LOOPS_REUSE_FROM_SESSION_ID",
        "LOOPS_BASE_URL", "LOOPS_TEAM",
    ) if name in os.environ]
    if overrides:
        raise ValueError(f"Unset SDK overrides before running this example: {', '.join(overrides)}")
    train_rows, eval_rows = read_examples(train_path), read_examples(eval_path)
    if {row["prompt"] for row in train_rows} & {row["prompt"] for row in eval_rows}:
        raise ValueError("Keep evaluation prompts out of the training data")
    if epochs < 1:
        raise ValueError("epochs must be positive")
    output_path = Path(output_path)
    if output_path.exists():
        raise FileExistsError(f"Choose a new report path: {output_path}")
    output_path.parent.mkdir(parents=True, exist_ok=True)
    key = os.environ["BASETEN_API_KEY"]
    record = {
        "base_model": base_model, "epochs": epochs, "rank": 16, "seed": 42,
        "learning_rate": 4e-5, "status": "starting", "cleanup_status": "pending",
        "train_sha256": hashlib.sha256(Path(train_path).read_bytes()).hexdigest(),
        "eval_sha256": hashlib.sha256(Path(eval_path).read_bytes()).hexdigest(),
    }
    service = ServiceClient(api_key=key)
    record["session_id"] = service.session_id
    write_report(output_path, record)
    print(f"session_id={service.session_id}", flush=True)

    with ExitStack() as cleanup:
        cleanup.callback(write_report, output_path, record)
        cleanup.callback(release_session, service.session_id, key, record)
        try:
            supported = {model.model_name for model in service.get_server_capabilities().supported_models}
            if base_model not in supported:
                raise ValueError(f"Model isn't available through Loops: {base_model}")
            trainer = service.create_lora_training_client(
                base_model=base_model, rank=16, seed=42, max_seq_len=MAX_TOKENS,
                ready_timeout=1200, sampler_timeout=120, name=output_path.stem,
            )
            cleanup.callback(trainer.close)
            record["run_id"] = trainer.run_id
            record["status"] = "training"
            write_report(output_path, record)
            print(f"run_id={trainer.run_id}", flush=True)
            tokenizer = trainer.get_tokenizer()
            train_data = [to_datum(tokenizer, row) for row in train_rows]
            eval_data = [to_datum(tokenizer, row) for row in eval_rows]
            record["baseline_loss"] = trainer.forward(data=eval_data).result(timeout=600).loss
            for _ in range(epochs):
                for start in range(0, len(train_data), 8):
                    trainer.forward_backward(data=train_data[start:start + 8]).result(timeout=600)
                    trainer.optim_step(AdamParams(learning_rate=4e-5)).result(timeout=600)
            record["trained_loss"] = trainer.forward(data=eval_data).result(timeout=600).loss
            record["trainer_checkpoint"] = trainer.save_state(name="final").result(timeout=600).path
            weights = trainer.save_weights_for_sampler(name="final").result(timeout=600)
            record["sampler_checkpoint"] = weights.path
            write_report(output_path, record)

            sampler = trainer.create_sampling_client(model_path=weights.path)
            cleanup.callback(sampler.close)
            record["evaluations"] = []
            for row in eval_rows:
                sample = sampler.sample(
                    prompt=ModelInput.from_ints(tokenizer.encode(row["prompt"], add_special_tokens=False)),
                    num_samples=1,
                    sampling_params=SamplingParams(max_tokens=16, temperature=0, stop=["\n"]),
                )
                completion = tokenizer.decode(sample.sequences[0].tokens).strip()
                record["evaluations"].append({
                    "prompt": row["prompt"], "expected": row["answer"].strip(),
                    "completion": completion,
                    "match": completion.casefold() == row["answer"].strip().casefold(),
                })
            record["status"] = "completed"
        except BaseException as error:
            record["status"] = "failed"
            record["error_type"] = type(error).__name__
            raise
    print(f"saved report to {output_path}", flush=True)
    return record


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--train", default="train.jsonl")
    parser.add_argument("--eval", default="eval.jsonl")
    parser.add_argument("--output", default="round-1.json")
    args = parser.parse_args()
    run_round(args.train, args.eval, args.output)

The script saves the session ID before creating the trainer and the run ID when the trainer is ready. During cleanup, it closes the SDK clients and calls Deactivate a run for runs in that session.

Run the script and check results

Run the script with the example data:

uv run --python 3.12 campaign.py

A successful run prints its session ID, run ID, and report path. The IDs vary by run:

session_id=<SESSION_ID>
run_id=<RUN_ID>
saved report to round-1.json

Open round-1.json to check the results:

Report field	How to use it
`session_id`, `run_id`	Find the run in the dashboard or Loops API.
`train_sha256`, `eval_sha256`	Identify the exact training and evaluation files used.
`baseline_loss`, `trained_loss`	Compare the same evaluation examples before and after training.
`evaluations`	Compare each generated answer with the expected answer.
`trainer_checkpoint`	Resume training with weights and optimizer state.
`sampler_checkpoint`	Generate outputs from the saved weights or deploy them for inference.
`status`, `error_type`	Check whether training and evaluation completed or which error stopped them.
`cleanup_status`, `cleanup_error_type`	Check whether cleanup completed or which error stopped it.
`deactivated_run_ids`	Find the runs whose deactivation requests succeeded.

If cleanup_status isn’t completed, use List runs to find runs with the saved session_id, then deactivate each run. Network failures or a terminated process can prevent cleanup from finishing. Check this before retrying so you don’t leave GPUs running. The example data tests that the code works; it’s too small to measure model quality. The script scores an answer as correct when it matches the expected label, ignoring case and surrounding whitespace. Use evaluation criteria that fit your task.

Train with updated data

Your application can import run_round() from campaign.py and use its returned report to find incorrect answers. Use those mistakes to guide your data selection, then start another round. Keep the evaluation file the same so you can compare results. Save the updated training data as train-v2.jsonl and choose a new report path:

uv run --python 3.12 campaign.py \
  --train train-v2.jsonl \
  --eval eval.jsonl \
  --output round-2.json

Each call starts a new LoRA training run from the selected base model. To continue training from a saved checkpoint, use create_training_client_from_state_with_optimizer() with trainer_checkpoint. Record which starting point you use when comparing rounds. Set limits in your application for spending, total rounds, and rounds running at once. Stop when the model meets your evaluation target or reaches a limit. Save each round’s results before starting the next one. Before retrying a failed worker, check its saved session and run IDs to avoid starting the same round twice.

Next steps

Train on your data covers chat-format data and training in batches.
Loops SDK reference documents the classes and parameters used by the script.
Deploy a checkpoint serves saved sampler weights for inference.

Get started

Model APIs

Inference

Training

Development

Deployment

Engines

Model Labs

Organization

Hosting options

Observability

Troubleshooting

Programmatic training

Before you begin

Prepare training and evaluation data

Build a training round

Save the complete script

Run the script and check results

Train with updated data

Next steps

​Before you begin

​Prepare training and evaluation data

​Build a training round

​Save the complete script

​Run the script and check results

​Train with updated data

​Next steps

Before you begin

Prepare training and evaluation data

Build a training round

Save the complete script

Run the script and check results

Train with updated data

Next steps