Python API

konduktor.Task(
    name: str,                                              # required
    run: str | Callable[[int, list[str]], str | None] | None = None,  # required
    envs: dict[str, str] | None = None,                     # optional
    workdir: str | None = None,                             # optional
    num_nodes: int | None = None,                           # optional; defaults to 1
)

konduktor.Resources(
    cpus: int | float | str | None = None,      # required;
    memory: int | float | str | None = None,    # required;
    accelerators: str | None = None,            # optional;
    image_id: str | None = None,                # required;
    labels: dict[str, str] | None = None,       # required: kueue.x-k8s.io/queue-name
    job_config: dict[str, int | str] | None = None, # optional; only max_restarts and completions
)

konduktor.Serving(
    min_replicas: int | None = None,    # required
    max_replicas: int | None = None,    # optional; defaults to min_replicas
    ports: int | None = 8000,           # optional; defaults to 8000
    probe: str | None = '/health',      # optional; defaults to probe=None (general) or /health (vLLM)
                                        # EXCLUDE PROBE COMPLETELY w VLLM DEPLOYMENTS
)

konduktor.launch(
    task: konduktor.Task,
    dryrun: bool = False,
    detach_run: bool = False,
) -> str | None

import konduktor

# Describe the work to run on each node.
task = konduktor.Task(
    name="python-api-demo",
    run="python train.py --epochs 10",
    workdir=".",
    envs={"myenv": "foo"},
)

# Specify compute requirements and image
resources = konduktor.Resources(
    cpus=4,
    memory=12,
    accelerators="H100",
    image_id="docker.io/ryanattrainy/pytorch-mnist:cpu",
    labels={"mylabel": "bar"},
    job_config={
        "max_restarts": 3,
        "completions": 2,
    },
)

task.set_resources(resources)

job_id = konduktor.launch(task)
print(f"Submitted job: {job_id}")

python launch_from_python.py