Jobs

SwarmJob

Subclass this and implement transform_items to define a job. Batching, bounded concurrency, retries and checkpointing are provided by the framework.

Note

Import SwarmJob from the domyn_swarm package root. The domyn_swarm.jobs.base module is deprecated in favour of domyn_swarm.jobs.api.base, which is what is documented here.

class domyn_swarm.jobs.api.base.SwarmJob(*, name=None, endpoint=None, model='', provider='openai', input_column_name='messages', id_column_name=None, output_column_name=None, output_cols=None, checkpoint_interval=16, max_concurrency=2, retries=5, timeout=600, client=None, client_kwargs=None, output_mode=OutputJoinMode.APPEND, default_output_cols=None, data_backend=None, native_backend=False, backend_read_kwargs=None, backend_write_kwargs=None, native_batch_size=None, **extra_kwargs)

Bases: ABC

Abstract base class for distributed LLM processing jobs in the Domyn swarm framework.

This class provides a robust foundation for running large-scale language model tasks with built-in reliability features including automatic checkpointing, retry mechanisms, and concurrent processing capabilities.

Key Features:

• Automatic Checkpointing: Periodically saves progress to enable recovery from failures

• Concurrent Processing: Configurable parallelism with rate limiting

  and timeout handling

• Retry Logic: Built-in exponential backoff for handling transient failures

• Provider Agnostic: Supports multiple LLM providers (OpenAI, vLLM, etc.)

  via pluggable clients

• Callback System: Extensible event hooks for monitoring and custom behavior

• DataFrame Integration: Native pandas DataFrame support for batch processing

Architecture:

The class follows a template method pattern where subclasses implement the core transform_items() method while inheriting all reliability and concurrency infrastructure. Processing flows through: DataFrame → batching → transform_items → results → checkpointing.

Example

class MyLLMJob(SwarmJob):
    async def transform_items(self, items: list[Any]) -> list[Any]:
        # Process items using self.client
        results = []
        for item in items:
            response = await self.client.chat.completions.create(
                model=self.model, messages=item, **self.kwargs
            )
            results.append(response.choices[0].message.content)
        return results


job = MyLLMJob(model="gpt-4", max_concurrency=5)
results_df = await job.run(input_df, tag="experiment_1")

Variables:

• api_version (int) – API version for compatibility tracking (default: 2)

• endpoint – LLM service endpoint URL (from ENDPOINT env var or parameter)

• model – Model identifier (e.g., “gpt-4”, “claude-3-sonnet”)

• provider – LLM provider name (“openai”, “anthropic”, etc.)

• input_column_name – DataFrame column containing input data

• id_column_name – Optional column name used for stable row ids

• output_cols – DataFrame column(s) for storing results

• checkpoint_interval – Items processed between automatic checkpoints

• max_concurrency – Maximum concurrent requests allowed

• retries – Maximum retry attempts for failed requests

• timeout – Request timeout in seconds

• client – Initialized async LLM client instance

• results – Final processed DataFrame after job completion

Raises:

• RuntimeError – When ENDPOINT environment variable is missing

• ValueError – When required model name is not provided

• NotImplementedError – When required abstract methods are not implemented

Parameters:

• name (str | None)

• endpoint (str | None)

• model (str)

• provider (str)

• input_column_name (str)

• id_column_name (str | None)

• output_column_name (str | list | None)

• output_cols (str | list | None)

• checkpoint_interval (int)

• max_concurrency (int)

• retries (int)

• timeout (float)

• client_kwargs (dict | None)

• output_mode (OutputJoinMode)

• default_output_cols (list[str] | None)

• data_backend (str | None)

• native_backend (bool)

• backend_read_kwargs (dict | None)

• backend_write_kwargs (dict | None)

• native_batch_size (int | None)

Note

Subclasses must implement the transform_items() method which processes a list of items and returns results in the same order. The framework handles all infrastructure concerns including error handling, checkpointing, progress tracking, and result aggregation. The transform() method is deprecated and should not be used.

api_version: int = 2

register_callback(event, fn)

Register a named callback (e.g., ‘on_batch_done’).

Parameters:

• event (str)

• fn (Callable)

Return type:

None

get_callback(event)

Parameters:

event (str)

Return type:

Callable | None

async run(df, tag, checkpoint_dir='.checkpoints')

Run the job end-to-end with checkpointing support.

Parameters:

• df (DataFrame)

• tag (str)

• checkpoint_dir (str | Path)

Return type:

DataFrame

async batched(seq, fn)

Run a batched async pipeline over seq using fn.

Supports retrying and invokes the ‘on_batch_done’ callback if registered.

Parameters:

• seq (list)

• fn (Callable)

Return type:

list

async transform(df)

Parameters:

df (DataFrame)

to_kwargs()

Serialize the job’s constructor parameters (for remote reconstruction).

Return type:

dict

abstractmethod async transform_items(items)

Pure transform: items -> results (same order). No I/O or checkpointing.

Parameters:

items (list[Any])

Return type:

list[Any]

async transform_streaming(items, *, on_flush, checkpoint_every)

Run a streaming transform without retaining all outputs in memory.

Parameters:

• items (list[Any]) – Input items to process.

• on_flush (Callable[[list[int], list[Any]], Awaitable[None]]) – Callback invoked with (idxs, outputs) for flushed batches.

• checkpoint_every (int) – Number of items between flushes.

run_job_unified

async domyn_swarm.jobs.execution.dispatch.run_job_unified(job_factory, data, *, input_col, output_cols, nshards=1, store_uri=None, checkpoint_every=16, data_backend=None, native_backend=None, checkpointing=True, runner='pandas', ray_address=None, output_path=None, shard_output=False, shard_mode='id', global_resume=False)

Run a SwarmJob with backend-aware execution and checkpointing.

Parameters:

• job_factory (Callable[[], Any]) – Callable that returns a SwarmJob instance.

• data (Any) – Input dataset in backend-native form (pandas, polars, arrow, ray dataset, etc.).

• input_col (str) – Column name to read inputs from.

• output_cols (list[str] | None) – Optional list of output column names (None for dict outputs).

• nshards (int) – Number of shards to split input into for non-ray execution.

• store_uri (str | None) – Base checkpoint store URI (required when checkpointing is enabled).

• checkpoint_every (int) – Flush interval in items.

• data_backend (str | None) – Backend name override (defaults to the job’s data_backend or “pandas”).

• native_backend (bool | None) – Override for native execution (required for ray).

• checkpointing (bool) – Whether to read/write checkpoint state.

• runner (str) – Runner implementation to use for non-ray backends (“pandas” or “arrow”).

• ray_address (str | None) – Optional Ray cluster address (only used for ray backend).

• output_path (Path | None) – Optional output path used to enable direct shard writes when using the pandas runner and directory outputs.

• shard_output (bool) – If True and output_path is a directory, write one parquet file per shard (based on nshards) using checkpoint outputs as the source of truth when supported by the runner/backend (currently Polars).

• shard_mode (str) – Sharding strategy (“id” for stable id hashing, “index” for legacy order).

• global_resume (bool) – Resume by filtering inputs with global done ids across shards.

Returns:

Backend-native result for non-ray runs, or the Ray runner result. Returns None when the pandas or polars paths write outputs directly to a directory.

Raises:

• TypeError – If the job does not implement the streaming API.

• ValueError – If required id columns are missing or if checkpointing is misconfigured.

• RuntimeError – If the backend cannot be resolved.

Return type:

Any