YAML Spec API¶

The YAML spec API lets you declare a ROSE workflow as a data file instead of Python code. The spec is schema-validated when loaded — missing slots, unknown keys, and type errors are caught before any infrastructure starts. Task functions are plain Python functions with no decorators required. The same spec file works with any ROSE learner type: Active Learning, Reinforcement Learning, or UQ.

Loading a Spec¶

For the common case — build a learner and run it — LearnerBuilder loads and validates the YAML itself; you never need to touch a config object:

from rose.spec.builder import LearnerBuilder

builder = LearnerBuilder("workflow.yaml", asyncflow)  # validates schema on load
cfg     = builder.config                              # typed WorkflowConfig, if you need it

Reach for load_spec instead when you need WorkflowSpec-level features: spec variants via workflow_with(), import validation, or the .workflow coroutine used by ROSE as a Service:

from rose.spec import load_spec

spec = load_spec("workflow.yaml")  # validates schema on load
cfg  = spec.config                 # typed WorkflowConfig object

Both raise ValueError with a precise message on any schema violation. Neither imports task modules by default — see Import Validation to enable that check.

Resource Specification via task_description¶

Every task — simulation, training, active_learn, environment, update, … — can carry an optional task_description. In the Python API this is a parameter on the decorated function itself, with a default value:

@acl.simulation_task(as_executable=False)
async def simulate(*args, task_description={"process_templates": [(4, {})]}, **kwargs):
    ...

The accepted keys depend entirely on which execution backend your asyncflow session is using — see asyncflow's Execution Backends guide for the full per-backend reference.

The YAML spec exposes the same dict as a task_description: key on the corresponding task slot — see the side-by-side example in Task Types below.

Sequential Active Learner¶

from rose.al import SequentialActiveLearner

acl = SequentialActiveLearner(asyncflow)

@acl.simulation_task(as_executable=False)
async def simulate(*args, **kwargs):
    ...  # return simulation result

@acl.training_task(as_executable=False)
async def train(sim_result, **kwargs):
    ...  # return trained model

@acl.active_learn_task(as_executable=False)
async def active_learn(sim_result, model, **kwargs):
    ...  # return updated dataset

@acl.as_stop_criterion(
    metric_name="mse",
    threshold=0.01,
    operator="<",
)
async def check_mse(*args, **kwargs):
    ...  # return float metric

async for state in acl.start(max_iter=5):
    print(f"iter {state.iteration}  "
          f"mse={state.metric_value:.4f}")

await asyncflow.shutdown()

learner:
  type: sequential_active_learner
  max_iter: 5

simulation:
  type: python
  function: tasks:simulate

training:
  type: python
  function: tasks:train

active_learn:
  type: python
  function: tasks:active_learn

stop_criterion:
  metric: mse
  threshold: 0.01
  operator: "<"
  evaluator:
    type: python
    function: tasks:check_mse

# run it
from rose.spec.builder import LearnerBuilder

builder = LearnerBuilder("workflow.yaml", asyncflow)
cfg     = builder.config
learner = builder.build()

async for state in learner.start(
    max_iter=cfg.learner.max_iter
):
    print(f"iter {state.iteration}  "
          f"mse={state.metric_value:.4f}")

Task functions referenced by function: tasks:simulate are ordinary Python callables in a tasks.py module — no ROSE decorators required:

# tasks.py
def simulate(*args, **kwargs):
    ...  # return simulation result

def train(sim_result, **kwargs):
    ...  # return trained model

def active_learn(sim_result, model, **kwargs):
    ...  # return updated dataset

def check_mse(*args, **kwargs):
    ...  # return float metric value

Parallel Learner — Shared Tasks¶

When all parallel learners run the same task implementations, declare the task slots at the top level exactly as in the sequential case.

from rose.al import ParallelActiveLearner

acl = ParallelActiveLearner(asyncflow)

@acl.simulation_task(as_executable=False)
async def simulate(*args, **kwargs):
    ...

@acl.training_task(as_executable=False)
async def train(sim_result, **kwargs):
    ...

@acl.active_learn_task(as_executable=False)
async def active_learn(sim_result, model, **kwargs):
    ...

@acl.as_stop_criterion(
    metric_name="mse",
    threshold=0.01,
)
async def check_mse(*args, **kwargs):
    ...

async for state in acl.start(
    parallel_learners=2,
    max_iter=5,
):
    print(f"learner {state.learner_id}  "
          f"iter {state.iteration}")

learner:
  type: parallel_active_learner
  max_iter: 5
  parallel_learners: 2

simulation:
  type: python
  function: tasks:simulate

training:
  type: python
  function: tasks:train

active_learn:
  type: python
  function: tasks:active_learn

stop_criterion:
  metric: mse
  threshold: 0.01
  operator: "<"
  evaluator:
    type: python
    function: tasks:check_mse

# run it
builder = LearnerBuilder("workflow.yaml", asyncflow)
cfg     = builder.config
learner = builder.build()

async for state in learner.start(
    max_iter=cfg.learner.max_iter,
    parallel_learners=cfg.learner.parallel_learners,
):
    print(f"learner {state.learner_id}  "
          f"iter {state.iteration}")

Each learner receives a unique learner_id kwarg (0, 1, …) in its task calls. Task functions can use it to write to isolated files or namespaces.

Parallel Learner — Distinct Tasks (learners: block)¶

When each parallel learner needs its own task implementations — different models, different executables, different command-line flags — use the learners: block. Each entry gets a label that is injected as learner_label into every task kwarg.

acl = ParallelActiveLearner(asyncflow)

# Manual routing by learner_id
TRAIN_CMD = {
    0: "python train.py --model linear",
    1: "python train.py --model ridge",
}

@acl.simulation_task
async def simulate(*args, **kwargs):
    return "python sim.py"

@acl.training_task
async def train(*args, **kwargs):
    return TRAIN_CMD[kwargs["learner_id"]]

@acl.active_learn_task
async def active_learn(*args, **kwargs):
    return "python active_learn.py"

@acl.as_stop_criterion(
    metric_name="mse", threshold=0.01
)
async def check_mse(*args, **kwargs):
    ...

# Build per-learner configs manually
from rose.learner import LearnerConfig, TaskConfig
lcs = [
    LearnerConfig(
        simulation={i: TaskConfig(kwargs={"learner_id": i})
                    for i in range(6)},
        ...
    )
    for lid in range(2)
]

async for state in acl.start(
    parallel_learners=2,
    max_iter=5,
    learner_configs=lcs,
):
    ...

learner:
  type: parallel_active_learner
  max_iter: 5

learners:
  - label: linear_regression
    simulation:
      type: shell
      command: python sim.py --label a
    training:
      type: shell
      command: python train.py --label a --model linear
    active_learn:
      type: shell
      command: python active_learn.py --label a

  - label: ridge_regression
    simulation:
      type: shell
      command: python sim.py --label b
    training:
      type: shell
      command: python train.py --label b --model ridge
    active_learn:
      type: shell
      command: python active_learn.py --label b

stop_criterion:
  metric: mse
  threshold: 0.01
  operator: "<"
  evaluator:
    type: python
    function: tasks:check_mse

# run it — builder handles routing + LearnerConfig
builder = LearnerBuilder("workflow.yaml", asyncflow)
cfg     = builder.config
learner = builder.build()
lcs     = builder.build_learner_configs()

async for state in learner.start(
    max_iter=cfg.learner.max_iter,
    parallel_learners=len(lcs),
    learner_configs=lcs,
):
    label = cfg.learners[state.learner_id].label
    print(f"[{label}]  iter {state.iteration}")

The builder creates a single dispatch closure per slot that routes to the correct command or function based on learner_id. The learner_id kwarg is consumed by the routing layer and never reaches the user function.

Task Types¶

simulation:
  type: python
  function: my_package.tasks:simulate

# tasks.py
def simulate(*args, **kwargs):
    # receives positional results from
    # prior tasks as *args
    return {"X": ..., "y": ...}

simulation:
  type: shell
  command: python sim.py

simulation:
  type: shell
  command: python sim.py --dataset {dataset}

task_description example¶

@acl.simulation_task(as_executable=False)
async def simulate(*args, task_description={"process_templates": [(4, {})]}, **kwargs):
    ...

simulation:
  type: python
  function: tasks:simulate
  task_description:
    process_templates:
      - [4, {}]

Stop Criterion¶

@acl.as_stop_criterion(
    metric_name="mse",
    threshold=0.01,
    operator="<",
    as_executable=False,
)
async def check_mse(*args, **kwargs):
    ...  # return float

stop_criterion:
  metric: mse
  threshold: 0.01
  operator: "<"        # default; also: >, ==, <=, >=
  evaluator:
    type: python
    function: tasks:check_mse

The evaluator function follows the same *args, **kwargs convention as task functions. ROSE stops the loop when metric_value <operator> threshold is satisfied, or when max_iter is reached — whichever comes first.

Parameters¶

The parameters: block defines key-value pairs that are injected into every task's **kwargs at every iteration:

parameters:
  dataset: my_dataset
  batch_size: 32
  growing_pool: false

def simulate(*args, **kwargs):
    dataset      = kwargs["dataset"]      # "my_dataset"
    batch_size   = kwargs["batch_size"]   # 32
    growing_pool = kwargs["growing_pool"] # False
    iteration    = kwargs["iteration"]    # 0, 1, 2, ...
    ...

Reserved keys¶

The following keys are injected automatically by the builder and must not appear in parameters::

Remote Config¶

remote:
  pythonpath:
    - /path/to/my/task/modules
    - /path/to/shared/utilities
  backends: [dragon]   # optional — default shown; use [concurrent] for CPU-only runs

remote.pythonpath¶

remote.pythonpath entries are added to sys.path on the remote worker before any task module is imported. They are also injected into every task's kwargs["pythonpath"] (as a list), so task functions can construct file paths without duplicating the value in parameters::

def train(sim_result, **kwargs):
    pythonpath = kwargs.get("pythonpath", [])
    base = pythonpath[0] if pythonpath else ""
    script = Path(base) / "scripts" / "train_model.py"
    ...

remote.pythonpath is the single edit point for the remote worker path — changing it updates both the import path and the kwarg.

remote.backends¶

remote.backends selects which Rhapsody backends are instantiated on the remote orbit endpoint session. The value is a list of backend name strings:

Example — switch to concurrent for a CPU-only endpoint:

remote:
  pythonpath: [...]
  backends: [concurrent]

backends is consumed only at engine-creation time and is not injected into task kwargs.

remote.target — bootstrapping the endpoint for rose run --remote¶

remote.target tells rose run <yaml> --remote how to launch the remote orbit endpoint itself, instead of assuming one is already running. Omit it to keep today's manual-bootstrap behavior (start the endpoint yourself, use --local/the orbit backend directly). See the Command-Line Interface page for the --local/--remote flags themselves, and rose setup for an interactive wizard that walks you through getting this working.

remote:
  backends: [dragon]
  target:
    kind: sfapi            # iri | sfapi | psij
    endpoint: nersc          # iri/sfapi only: nersc | olcf
    resource_id: perlmutter
    account: amsc007
    queue_name: debug
    walltime_min: 30
    n_nodes: 1
    constraint: cpu
    home_dir: /global/u2/m/merzky
    tunnel: none            # none | forward | reverse

Credentials are read from disk/env, never stored in the spec: iri reads a bearer token from ~/.amsc/token_<endpoint>; sfapi reads a client ID from $SFAPI_CLIENT_ID and a private key from ~/.amsc/sfapi_key_<endpoint>.pem.

remote.embedded — run without a standalone broker¶

remote.embedded: true hosts the ORBIT broker inside the rose run --remote process itself (EndpointRuntime/EmbeddedBroker) instead of connecting to one already running elsewhere. No separate radical-orbit-broker.py deployment is needed. Mutually exclusive with remote.broker_url.

remote:
  embedded: true
  target:
    kind: psij
    account: amsc007
    queue_name: debug
    # edge_name omitted — PsiJ runs on the embedded broker itself

The embedded broker is still a real broker: it needs the same operator-placed cert/key/token under ~/.radical/orbit (or their env-var redirects) as a standalone one — see radical.orbit's CLAUDE.md TOKEN_RECIPE if those aren't set up yet.

Combined with target.kind: psij, target.edge_name is not required — PsiJ loads directly on the embedded broker (this only works when the process running rose run --remote itself has batch-scheduler access, e.g. you're on a login node). target.kind: iri/sfapi work the same as without embedded — only where the broker lives changes.

Tracking¶

tracking:
  backend: mlflow      # mlflow | clearml | none (default)
  experiment: my-exp   # experiment/project name
  run_name: run-01     # optional run label

See MLflow integration and ClearML integration for configuration details.

Spec Variants with workflow_with()¶

workflow_with() returns a new WorkflowSpec with selective overrides applied. The original spec is never mutated:

base_spec  = load_spec("workflow.yaml")
test_spec  = base_spec.workflow_with(max_iter=2, parameters={"dataset": "test_ds"})
large_spec = base_spec.workflow_with(max_iter=50, parameters={"batch_size": 128})

Accepted override keys:

• parameters — merged (not replaced) into the existing parameters: block
• Any learner field: max_iter, parallel_learners
• Any other top-level spec field

Unknown keys raise ValueError immediately.

Import Validation¶

By default, load_spec does not import task modules — this is intentional when remote.pythonpath points to paths that only exist on the remote worker. To catch function: typos locally during development:

spec = load_spec("workflow.yaml", validate_imports=True)

With validate_imports=True, every module:callable string is resolved in the current environment. All failures are collected and reported in a single ValueError before any infrastructure starts. Use this during development when task files are locally accessible.

Spec Reference¶

Top-level keys¶

TaskDef fields¶