The SwarmJob lifecycle#

This page describes how domyn-swarm job submit produces a Parquet output file. It also defines the work that a custom job must do.

1. The CLI resolves the class#

job submit takes <module>:<ClassName> and imports it, defaulting to domyn_swarm.jobs:ChatCompletionJob. --job-kwargs is parsed as JSON and passed to the constructor as configuration overrides. Put provider parameters such as temperature and top_p under request_params. The CLI passes these values to the client without interpreting them:

{"max_concurrency": 8, "request_params": {"temperature": 0.2}}

The swarm is located either by --name, rehydrated from the state record, or created fresh from --config.

2. A driver process starts beside the endpoint#

The compute backend submits a process with ENDPOINT and MODEL in its environment. Slurm uses srun, and Lepton uses a batch job. The process runs python -m domyn_swarm.jobs.cli.run.

The job does not execute inside the serving container. It is an HTTP client like any other.

3. The API version is resolved#

resolve_job_api decides which execution path applies, in this order:

  1. A class-level api_version >= 2 means the current API.

  2. Otherwise, an override of transform_items or transform_streaming means the current API.

  3. Otherwise, an override of transform or run means the legacy API.

  4. All other classes use the legacy API.

_ensure_new_api rejects legacy transform(df) jobs. To port one, implement transform_items(items). Method-based detection accepts a subclass that implements the current method without an explicit version flag.

4. run_job_unified drives the work#

run_job_unified provides these services around the job method:

  • Batching groups items instead of sending one item at a time.

  • Bounded concurrency limits in-flight requests with --max-concurrency.

  • Retries use Tenacity backoff for transient failures.

  • Checkpointing writes periodic results for resume after a failure.

  • Sharding splits input with --num-shards and --shard-mode.

  • Data backend selection chooses pandas, polars, or Ray with a matching runner.

See Checkpointing and resuming and Sharding and concurrency.

5. Your method runs#

A subclass must implement one method:

async def transform_items(self, items: list[Any]) -> list[Any]:
    ...

The method must return one result for each input item, in the same order. The framework controls item selection, scheduling, and result persistence.

SwarmJob also provides transform_streaming, which is the default path built on top of transform_items, so implementing the latter is enough. Override transform_streaming only when you need control over how items are consumed.

The method can use these attributes:

  • self.client: An AsyncOpenAI client for the swarm endpoint.

  • self.model: The model that the swarm serves.

  • self.kwargs: Provider parameters from request_params, whether supplied by --job-kwargs or the constructor.

  • self.output_cols: The columns that receive the results.

6. Results are joined and written#

The framework matches results to input rows by the --id-column value or a generated identifier. It writes the joined data to --output. A sharded directory output can contain one Parquet file per shard.

Because the join is by id rather than position, a resumed run can write rows it computed in an earlier attempt without recomputing them.

Why the contract is shaped this way#

A pure transform performs no I/O and has no retry-sensitive side effects. It also does not depend on item grouping. This contract lets the framework reorder, batch, retry, and persist work.

For this reason, transform_items takes and returns a list. Checkpointing stays outside the method.