1
import json
2
import os
3
import re
4
import time
5
import uuid
6
from pathlib import Path
7
from nemo_platform import NeMoPlatform, ConflictError
8
from nemo_platform.types.secrets import PlatformSecretResponse
9
from nemo_platform.types.files import HuggingfaceStorageConfigParam
10
11
12
def sanitize_name(prefix: str, name: str):
13
    """Sanitize model_name for deployment/config naming. Compatible with platform naming rules."""
14
    name = name.split("/")[-1]
15
    sanitized = re.sub(r"[^a-z0-9@.+_-]", "-", name.lower())
16
    sanitized = re.sub(r"-+", "-", sanitized).strip("-")
17
    return f"{prefix}-{sanitized}"[:59].rstrip("-")
18
19
20
def max_wait_time_checker(seconds: int, job_name: str = ""):
21
    """Return a check() that raises TimeoutError if called after `seconds` have elapsed."""
22
    start_time = time.time()
23
24
    def check():
25
        if time.time() - start_time > seconds:
26
            raise TimeoutError(f"{job_name} took longer than {seconds} seconds")
27
28
    return check
29
30
31
NMP_BASE_URL = os.environ.get("NMP_BASE_URL", "http://localhost:8080")
32
client = NeMoPlatform(
33
    base_url=NMP_BASE_URL,
34
    workspace="default"
35
)

1
from datasets import load_dataset, DatasetDict
2
3
print("Loading dataset rajpurkar/squad")
4
raw_dataset = load_dataset("rajpurkar/squad")
5
if not isinstance(raw_dataset, DatasetDict):
6
    raise ValueError("Dataset does not contain expected splits")
7
print("Loaded dataset")
8
9
VALIDATION_PROPORTION = 0.05
10
SEED = 1234
11
training_size = 3000
12
validation_size = 300
13
DATASET_NAME = "sft-dataset"
14
DATASET_PATH = Path("sft-dataset").absolute()
15
16
os.makedirs(DATASET_PATH, exist_ok=True)
17
train_set = raw_dataset.get("train")
18
split_dataset = train_set.train_test_split(test_size=VALIDATION_PROPORTION, seed=SEED)
19
train_ds = split_dataset["train"].select(range(min(training_size, len(split_dataset["train"]))))
20
validation_ds = split_dataset["test"].select(range(min(validation_size, len(split_dataset["test"]))))
21
22
def convert_squad_to_sft_format(example):
23
    prompt = f"Context: {example['context']} Question: {example['question']} Answer:"
24
    completion = example["answers"]["text"][0]
25
    return {"prompt": prompt, "completion": completion}
26
27
with open(f"{DATASET_PATH}/training.jsonl", "w", encoding="utf-8") as f:
28
    for example in train_ds:
29
        f.write(json.dumps(convert_squad_to_sft_format(example)) + "\n")
30
with open(f"{DATASET_PATH}/validation.jsonl", "w", encoding="utf-8") as f:
31
    for example in validation_ds:
32
        f.write(json.dumps(convert_squad_to_sft_format(example)) + "\n")
33
34
print(f"Saved training.jsonl with {len(train_ds)} rows")
35
print(f"Saved validation.jsonl with {len(validation_ds)} rows")
36
with open(f"{DATASET_PATH}/training.jsonl", "r") as f:
37
    sample = json.loads(f.readline())
38
print("Sample prompt (first 200 chars):", sample["prompt"][:200] + "...")
39
print("Sample completion:", sample["completion"])

1
try:
2
    client.files.filesets.create(
3
        workspace="default",
4
        name=DATASET_NAME,
5
        description="SFT training data",
6
        cache=True,
7
    )
8
    print(f"Created fileset: {DATASET_NAME}")
9
except ConflictError:
10
    print(f"Fileset '{DATASET_NAME}' already exists, continuing...")
11
12
client.files.fsspec.put(
13
    lpath=DATASET_PATH,
14
    rpath=f"default/{DATASET_NAME}/",
15
    recursive=True
16
)
17
print("Training data:")
18
print(client.files.list(fileset=DATASET_NAME, workspace="default"))

1
HF_TOKEN = os.getenv("HF_TOKEN")
2
3
def create_or_get_secret(name: str, value: str | None, label: str) -> PlatformSecretResponse | None:
4
    if not value:
5
        print(f"{label} is not set - skipping setting secret")
6
        return None
7
    try:
8
        secret = client.secrets.create(name=name, workspace="default", value=value)
9
        print(f"Created secret: {name}")
10
        print(secret.model_dump_json(indent=2))
11
        return secret
12
    except ConflictError:
13
        print(f"Secret '{name}' already exists, continuing...")
14
        secret = client.secrets.retrieve(name=name, workspace="default")
15
        print(secret.model_dump_json(indent=2))
16
        return secret
17
18
19
hf_secret = create_or_get_secret("hf-token", HF_TOKEN, "HF_TOKEN")

1
HF_REPO_ID = "Qwen/Qwen3-0.6B"
2
MODEL_NAME = "qwen3-0.6b"
3
4
try:
5
    storage = HuggingfaceStorageConfigParam(
6
        type="huggingface",
7
        repo_id=HF_REPO_ID,
8
        repo_type="model",
9
    )
10
    if hf_secret:
11
        storage["token_secret"] = hf_secret.name
12
    base_model_fs = client.files.filesets.create(
13
        workspace="default",
14
        name=MODEL_NAME,
15
        description="Qwen3 0.6b base model from Huggingface",
16
        storage=storage,
17
        cache=True,
18
    )
19
except ConflictError:
20
    base_model_fs = client.files.filesets.retrieve(workspace="default", name=MODEL_NAME)
21
22
try:
23
    base_model = client.models.create(
24
        workspace="default",
25
        name=MODEL_NAME,
26
        fileset=f"default/{MODEL_NAME}",
27
        trust_remote_code=False,
28
    )
29
except ConflictError:
30
    client.models.update(
31
        workspace="default",
32
        name=MODEL_NAME,
33
        fileset=f"default/{MODEL_NAME}",
34
        trust_remote_code=False,
35
    )
36
    base_model = client.models.retrieve(workspace="default", name=MODEL_NAME)
37
38
print(f"Base model fileset: fileset://default/{base_model.name}")
39
print(client.files.list(fileset=MODEL_NAME, workspace="default"))
40
41
time_check = max_wait_time_checker(600, "Model Spec")
42
while not base_model.spec:
43
    time_check()
44
    time.sleep(10)
45
    base_model = client.models.retrieve(workspace="default", name=MODEL_NAME)
46
47
# Clear verbose linear_layers list for cleaner output
48
base_model.spec.linear_layers = None
49
print(f"ModelSpec: {base_model.spec}")

1
from nemo_automodel_plugin.schema import AutomodelJobInput
2
3
job_suffix = uuid.uuid4().hex[:4]
4
JOB_NAME = f"my-sft-job-{job_suffix}"
5
OUTPUT_NAME = f"lora-adapter-{job_suffix}"
6
7
spec = AutomodelJobInput(
8
    model=f"default/{base_model.name}",
9
    dataset={"training": f"default/{DATASET_NAME}"},
10
    training={
11
        "training_type": "sft",
12
        "finetuning_type": "lora",
13
        "max_seq_length": 2048,
14
    },
15
    schedule={"epochs": 2},
16
    batch={"global_batch_size": 64, "micro_batch_size": 1},
17
    optimizer={"learning_rate": 5e-5},
18
    parallelism={
19
        "num_gpus_per_node": 1,
20
        "num_nodes": 1,
21
        "tensor_parallel_size": 1,
22
        "pipeline_parallel_size": 1,
23
        "context_parallel_size": 1,
24
        "expert_parallel_size": 1,
25
    },
26
    output={"name": OUTPUT_NAME},
27
)
28
29
job = client.customization.automodel.jobs.create(
30
    spec=spec, workspace="default", name=JOB_NAME
31
)
32
print(f"Submitted job: {job.job.name}")
33
print(f"Output adapter: {OUTPUT_NAME}")

1
from IPython.display import clear_output
2
3
time_check = max_wait_time_checker(3600, "Customization Job")
4
while True:
5
    time_check()
6
    status = client.jobs.get_status(name=job.job.name, workspace="default")
7
    clear_output(wait=True)
8
    print(f"Job Status: {status.status}")
9
    step = max_steps = training_phase = None
10
    for job_step in status.steps or []:
11
        if job_step.name == "training":
12
            for task in job_step.tasks or []:
13
                d = task.status_details or {}
14
                step, max_steps = d.get("step"), d.get("max_steps")
15
                training_phase = d.get("phase")
16
                break
17
            break
18
    if step is not None and max_steps is not None:
19
        print(f"Training: Step {step}/{max_steps} ({100 * step / max_steps:.1f}%)")
20
        if training_phase:
21
            print(f"Phase: {training_phase}")
22
    if status.status in ("completed", "failed", "cancelled", "error"):
23
        print(f"\nJob finished: {status.status}")
24
        break
25
    time.sleep(10)
26
27
assert status.status == "completed"

1
deploy_suffix = uuid.uuid4().hex[:4]
2
DEPLOYMENT_CONFIG_NAME = f"lora-deploy-cfg-{deploy_suffix}"
3
deployment_name = f"lora-deploy-{deploy_suffix}"
4
5
deployment_config = client.inference.deployment_configs.create(
6
    workspace="default",
7
    name=DEPLOYMENT_CONFIG_NAME,
8
    engine="vllm",
9
    model_spec={
10
        "model_namespace": "default",
11
        "model_name": MODEL_NAME,
12
        "lora_enabled": True,
13
    },
14
    executor_config={
15
        "gpu": 1,
16
        "image_name": "vllm/vllm-openai",
17
        "image_tag": "v0.22.1",
18
        "additional_args": ["--max-lora-rank", "32"],
19
    },
20
)
21
22
deployment = client.inference.deployments.create(
23
    workspace="default",
24
    name=deployment_name,
25
    config=deployment_config.name,
26
)
27
28
model_entity = client.models.retrieve(workspace="default", name=MODEL_NAME)
29
if model_entity.spec:
30
    model_entity.spec.linear_layers = None
31
print(model_entity.model_dump_json(indent=2))
32
print(f"Deployment status: {deployment.status}")

1
TIMEOUT_MINUTES = 30
2
start_time = time.time()
3
time_check = max_wait_time_checker(TIMEOUT_MINUTES * 60, "Deployment")
4
print(f"Monitoring deployment '{deployment_name}'... (timeout {TIMEOUT_MINUTES} min)\n")
5
6
while True:
7
    time.sleep(15)
8
    time_check()
9
    deployment_status = client.inference.deployments.retrieve(name=deployment_name, workspace="default")
10
    elapsed = time.time() - start_time
11
    clear_output(wait=True)
12
    print(f"Deployment: {deployment_name}")
13
    print(f"Status: {deployment_status.status}")
14
    print(f"Elapsed: {int(elapsed // 60)}m {int(elapsed % 60)}s")
15
    if deployment_status.status in ("RUNNING", "READY"):
16
        print("\nDeployment is ready!")
17
        if not client.models.wait_for_gateway(deployment_name, workspace="default", timeout=60):
18
            raise RuntimeError("Inference gateway did not become ready")
19
        break
20
    if deployment_status.status in ("FAILED", "ERROR", "TERMINATED"):
21
        raise RuntimeError(f"Deployment failed with status: {deployment_status.status}")
22
23
assert deployment_status.status in ("RUNNING", "READY")

1
context = "The Apollo 11 mission was the first manned mission to land on the Moon. It was launched on July 16, 1969, and Neil Armstrong became the first person to walk on the lunar surface on July 20, 1969. Buzz Aldrin joined him shortly after, while Michael Collins remained in lunar orbit."
2
question = "Who was the first person to walk on the Moon?"
3
messages = [
4
    {"role": "user", "content": f"Based on the following context, answer the question.\n\nContext: {context}\n\nQuestion: {question}"}
5
]
6
response = client.inference.gateway.provider.post(
7
    "v1/chat/completions",
8
    name=deployment_name,
9
    workspace="default",
10
    body={
11
        "model": OUTPUT_NAME,
12
        "messages": messages,
13
        "temperature": 0,
14
        "max_tokens": 256,
15
    }
16
)
17
print("=" * 60)
18
print("MODEL INFERENCE")
19
print("=" * 60)
20
print(f"Question: {question}")
21
print(f"Expected: Neil Armstrong")
22
print(f"Model output: {response['choices'][0]['message']['content']}")