Managed fine-tuning

Prepare a chat dataset

Training data is a JSONL file: one complete JSON object per line, each carrying a messages array in the familiar chat format. Roles are system, user and assistant, and the final message in each example is the assistant turn you want the model to learn to produce.

{"messages": [{"role": "system", "content": "You are the support assistant for Acme Billing."}, {"role": "user", "content": "Why was I charged twice this month?"}, {"role": "assistant", "content": "You changed plans mid-cycle, so this month shows a pro-rated charge for the old plan and a full charge for the new one. From next month you will see a single charge."}]}
{"messages": [{"role": "user", "content": "How do I download an invoice?"}, {"role": "assistant", "content": "Open Settings, choose Billing, select the invoice month and press Download PDF."}]}
{"messages": [{"role": "user", "content": "Can I pay by bank transfer?"}, {"role": "assistant", "content": "Yes, annual plans can be paid by bank transfer. Email billing support and we will issue an invoice with transfer details."}]}

A few hundred high-quality examples beat tens of thousands of noisy ones. Hold out a slice of the data as val.jsonl so the job can report validation loss, and keep a separate test set aside for the evaluation in step 5.

Upload and validate

Upload each file with POST /v1/files. The request is multipart/form-data, and purpose is fine-tune for training and validation files. Validation runs asynchronously: a new file starts in validating and moves to ready or error, so poll GET /v1/files/{file_id} before submitting a job.

import time
import requests

def upload(path, purpose):
    resp = requests.post(
        'https://api.nucleus-ai.io/v1/files',
        headers={'Authorization': f'Bearer {API_KEY}'},
        data={'purpose': purpose},
        files={'file': open(path, 'rb')},
    )
    return resp.json()['id']

def wait_ready(file_id):
    while True:
        file = requests.get(
            f'https://api.nucleus-ai.io/v1/files/{file_id}',
            headers={'Authorization': f'Bearer {API_KEY}'},
        ).json()
        if file['status'] != 'validating':
            return file
        time.sleep(5)

train_file = upload('train.jsonl', 'fine-tune')
val_file = upload('val.jsonl', 'fine-tune')
assert wait_ready(train_file)['status'] == 'ready'
assert wait_ready(val_file)['status'] == 'ready'

If validation fails, the file lands in error and errors lists every rejected row with its line number and the reason:

{
  "id": "file_9fQ2x",
  "status": "error",
  "errors": [
    {
      "line": 812,
      "message": "messages[2].role must be assistant"
    }
  ]
}

Fix the export script and re-upload. Catching a malformed dataset here costs seconds; catching it after a training run costs GPU hours.

Submit the job

One call to POST /v1/fine-tuning/jobs replaces the entire training loop:

resp = requests.post(
    'https://api.nucleus-ai.io/v1/fine-tuning/jobs',
    headers={'Authorization': f'Bearer {API_KEY}'},
    json={
        'base_model': 'Qwen/Qwen3-8B',
        'training_file': train_file,
        'validation_file': val_file,
        'method': 'sft',
        'hyperparameters': {'epochs': 3, 'lora_rank': 32},
        'suffix': 'support-assistant',
    },
)
job_id = resp.json()['id']

The parameters, briefly:

• method is sft (supervised fine-tuning on the chat format above, the default) or dpo (direct preference optimisation over preference pairs).
• hyperparameters accepts epochs (passes over the dataset), learning_rate, batch_size (examples per optimiser step) and lora_rank (adapter capacity; higher learns more but costs more memory). Every field is optional. Omit anything you do not have an opinion on and the platform picks sensible defaults; learning_rate defaults to auto, chosen from the base model and dataset size.
• suffix names the output model so you can recognise it in listings later.
• validation_file is optional but worth it: it is what powers the val_loss metric and the overfit warning in the next step.

The response confirms what will run. result_model is null until the job succeeds:

{
  "id": "ftjob_7Kd1p",
  "status": "queued",
  "base_model": "Qwen/Qwen3-8B",
  "training_file": "file_9fQ2x",
  "hyperparameters": {
    "epochs": 3,
    "lora_rank": 32,
    "learning_rate": "auto"
  },
  "result_model": null
}

Submitting a job allocates billable work, so if your client retries automatically, send an Idempotency-Key header to avoid launching the same job twice. See Errors and retries.

Monitor training

The simplest monitor is a poll on GET /v1/fine-tuning/jobs/{job_id}. The job moves through queued and running to succeeded, failed or cancelled, with progress reported as a fraction:

while True:
    job = requests.get(
        f'https://api.nucleus-ai.io/v1/fine-tuning/jobs/{job_id}',
        headers={'Authorization': f'Bearer {API_KEY}'},
    ).json()
    print(job['status'], job.get('progress'))
    if job['status'] in ('succeeded', 'failed', 'cancelled'):
        break
    time.sleep(30)

For a live view while the job runs, stream the events feed instead. GET /v1/fine-tuning/jobs/{job_id}/events with stream=true returns text/event-stream, emitting the loss curve step by step along with any warnings:

import json

resp = requests.get(
    f'https://api.nucleus-ai.io/v1/fine-tuning/jobs/{job_id}/events',
    headers={'Authorization': f'Bearer {API_KEY}'},
    params={'stream': 'true'},
    stream=True,
)
for line in resp.iter_lines():
    if not line.startswith(b'data: '):
        continue
    event = json.loads(line[len(b'data: '):])
    if 'loss' in event:
        print(f"step {event['step']}  loss {event['loss']}")
    if event['level'] == 'warn':
        print('warning:', event['message'])

{ "id": "evt_04", "level": "info", "step": 120, "loss": 0.94, "message": "epoch 2/3" }
{ "id": "evt_05", "level": "warn", "message": "val loss rising, possible overfit" }

Watch for warn events. A rising validation loss means the model has started memorising the training set; if you see it early and often, fewer epochs or more data is usually the fix. If the stream drops, resume from where you left off by passing the last event ID as after.

For long jobs, do not hold a connection open for hours. Register a webhook and let the platform call you when the job finishes; see Register a webhook.

When the job succeeds, the final poll carries everything you need for the next two steps:

{
  "id": "ftjob_7Kd1p",
  "status": "succeeded",
  "progress": 1.0,
  "trained_tokens": 8420117,
  "training_run_id": "run_3Bz",
  "result_model": "nucleus://run_3Bz:train:0/sampler_weights/000240",
  "metrics": { "final_loss": 0.72, "val_loss": 0.81 }
}

result_model is the trained model's path. Everything downstream, from evaluation to serving to export, takes this value.

Evaluate the result

A fine-tune without a measurement is a coin flip. Upload the held-out test set you kept aside in step 1 with purpose=evaluation, wrap it in a benchmark so the cases and the bar travel together, then create an evaluation with compare_to pointing at the base model. That one field is what turns a single score into a before-and-after number:

test_file = upload('test.jsonl', 'evaluation')
wait_ready(test_file)

benchmark = requests.post(
    'https://api.nucleus-ai.io/v1/benchmarks',
    headers={'Authorization': f'Bearer {API_KEY}'},
    json={'name': 'Held-out set', 'dataset_file': test_file, 'threshold': 0.8},
).json()

resp = requests.post(
    'https://api.nucleus-ai.io/v1/evaluations',
    headers={'Authorization': f'Bearer {API_KEY}'},
    json={
        'model': job['result_model'],
        'benchmark': benchmark['id'],
        'metric': 'exact_match',
        'compare_to': 'Qwen/Qwen3-8B',
    },
)
eval_id = resp.json()['id']

Pick the metric that matches the task: exact_match for closed answers, rouge for summaries, pass_rate for code, grader for open-ended responses. The evaluation runs asynchronously (the API returns 202 Accepted), so poll GET /v1/evaluations/{evaluation_id} until it succeeds:

{
  "id": "eval_2Wp",
  "status": "succeeded",
  "metric": "exact_match",
  "score": 0.813,
  "baseline_score": 0.642,
  "n": 500
}

score against baseline_score is the delta the fine-tune bought you, over n held-out examples. If the delta is small or negative, revisit the dataset before touching hyperparameters; data quality moves this number far more often than tuning does.

Serve it and chat

To put the model behind an endpoint, create a sampler from result_model:

requests.post(
    'https://api.nucleus-ai.io/v1/samplers',
    headers={'Authorization': f'Bearer {API_KEY}'},
    json={'model_path': job['result_model']},
)

Once the sampler is up, your model answers on the OpenAI-compatible chat route. Pass the result_model path as the model name:

resp = requests.post(
    'https://api.nucleus-ai.io/v1/chat/completions',
    headers={'Authorization': f'Bearer {API_KEY}'},
    json={
        'model': 'nucleus://run_3Bz:train:0/sampler_weights/000240',
        'messages': [
            {'role': 'user', 'content': 'Why was I charged twice this month?'},
        ],
    },
)
print(resp.json()['choices'][0]['message']['content'])

Because the route speaks the OpenAI wire format, any OpenAI SDK works unchanged with the base URL set to https://api.nucleus-ai.io/v1. See OpenAI compatibility for drop-in SDK setup.

Optional: take the weights with you

The trained model is yours. If you want to run it in your own infrastructure rather than (or as well as) behind a Nucleus sampler, export the checkpoint's weights with Export weights and download the archive. Serving your model covers both options, hosted and self-hosted, end to end.