distributed_eval#

Optional, fault-tolerant distributed evaluation for Puzzletron.

This package is intentionally isolated from the existing local evaluator and rpc_eval scaffold. Import it explicitly when distributed evaluation is desired; importing modelopt.torch.puzzletron does not start services or change scoring behavior.

Classes

`AsyncEvaluationClient`
`AttemptRecord`
`AttemptStatus`
`Campaign`
`CampaignManifest`
`ErrorKind`
`EvaluationClient`	Blocking facade backed by one persistent asyncio loop thread.
`EvaluationError`
`EvaluationHandle`
`EvaluationRequest`
`EvaluationResult`
`ParallelismSpec`
`WorkerRecord`
`WorkerState`

Functions

prefix_cache_id

Stable identity reserved for a future topology-specific prefix cache.

class AsyncEvaluationClient#

Bases: object

__init__(coordinator)#

Parameters:: coordinator (AsyncEvaluationCoordinator)

async as_completed(handles)#

Parameters:: handles (Iterable[EvaluationHandle])
Return type:: AsyncIterator[EvaluationResult]

async cancel(handle)#

Parameters:: handle (EvaluationHandle | str)
Return type:: bool

async close()#

Return type:: None

classmethod from_campaign(campaign_dir, **kwargs)#

Parameters:: campaign_dir (str)
Return type:: AsyncEvaluationClient

lookup(handle)#

Parameters:: handle (EvaluationHandle | str)
Return type:: EvaluationResult | None

status()#

Return type:: dict[str, Any]

async submit(request)#

Parameters:: request (EvaluationRequest)
Return type:: EvaluationHandle

async submit_many(requests)#

Parameters:: requests (Iterable[EvaluationRequest])
Return type:: list[EvaluationHandle]

async wait(handle, *, timeout=None)#

Parameters:

handle (EvaluationHandle | str)
timeout (float | None)

Return type:

EvaluationResult

class AttemptRecord#

Bases: StrictModel

attempt_id: str#

error: EvaluationError | None#

finished_at: datetime | None#

leased_at: datetime#

metadata: dict[str, Any]#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

request_id: str#

schema_version: Literal[1]#

status: AttemptStatus#

worker_boot_id: str | None#

worker_id: str | None#

class AttemptStatus#

Bases: str, Enum

CANCELLED = 'cancelled'#

CONFLICT = 'conflict'#

DUPLICATE = 'duplicate'#

FAILED = 'failed'#

LEASED = 'leased'#

RETRY = 'retry'#

SUCCEEDED = 'succeeded'#

__new__(value)#

class Campaign#

Bases: object

__init__(root, manifest)#

Parameters:

root (str | Path)
manifest (CampaignManifest)

property campaign_id: str#

coordinator_lease()#

Return type:: Iterator[None]

classmethod create(root, manifest)#

Parameters:

root (str | Path)
manifest (CampaignManifest)

Return type:

Campaign

classmethod open(root)#

Parameters:: root (str | Path)
Return type:: Campaign

property root: Path#

validate_request(request)#

Parameters:: request (EvaluationRequest)
Return type:: None

class CampaignManifest#

Bases: StrictModel

automodel_recipe: dict[str, Any]#

property campaign_id: str#

data: dict[str, Any]#

descriptor: str#

evaluator_revision: str#

force_hf: bool#

metadata: dict[str, Any]#

metrics: dict[str, Any]#

model: dict[str, Any]#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

parallelism: ParallelismSpec#

precision: dict[str, Any]#

result_atol: float#

result_rtol: float#

schema_version: Literal[1]#

class ErrorKind#

Bases: str, Enum

CANCELLED = 'cancelled'#

CANDIDATE = 'candidate'#

INTERNAL = 'internal'#

INVALID_REQUEST = 'invalid_request'#

PROCESS_GROUP = 'process_group'#

RESOURCE_EXHAUSTED = 'resource_exhausted'#

TIMEOUT = 'timeout'#

TRANSPORT = 'transport'#

UNSUPPORTED = 'unsupported'#

WORKER_LOST = 'worker_lost'#

__new__(value)#

class EvaluationClient#

Bases: object

Blocking facade backed by one persistent asyncio loop thread.

__init__(campaign_dir, **kwargs)#

Parameters:: campaign_dir (str)

batch(requests, *, timeout=None)#

Parameters:

requests (Iterable[EvaluationRequest])
timeout (float | None)

Return type:

list[EvaluationResult]

cancel(handle)#

Parameters:: handle (EvaluationHandle | str)
Return type:: bool

close()#

Return type:: None

evaluate(request, *, timeout=None)#

Parameters:

request (EvaluationRequest)
timeout (float | None)

Return type:

EvaluationResult

classmethod from_campaign(campaign_dir, **kwargs)#

Parameters:: campaign_dir (str)
Return type:: EvaluationClient

lookup(handle)#

Parameters:: handle (EvaluationHandle | str)
Return type:: EvaluationResult | None

status()#

Return type:: dict[str, Any]

submit(request)#

Parameters:: request (EvaluationRequest)
Return type:: EvaluationHandle

class EvaluationError#

Bases: StrictModel

kind: ErrorKind#

message: str#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

retryable: bool | None#

classmethod set_default_retryable(value)#

traceback: str | None#

class EvaluationHandle#

Bases: StrictModel

campaign_id: str#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

request_id: str#

class EvaluationRequest#

Bases: StrictModel

campaign_id: str#

data: dict[str, Any]#

evaluator_revision: str#

classmethod from_wire(payload)#

Parameters:: payload (dict[str, Any])
Return type:: EvaluationRequest

handler: str#

metadata: dict[str, Any]#

metrics: dict[str, Any]#

model: dict[str, Any]#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

payload: dict[str, Any]#

precision: dict[str, Any]#

property request_id: str#

schema_version: Literal[1]#

to_wire()#

Return type:: dict[str, Any]

class EvaluationResult#

Bases: StrictModel

artifacts: dict[str, Any]#

campaign_id: str#

completed_at: datetime#

counts: dict[str, int | float]#

metrics: dict[str, Any]#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

provenance: dict[str, Any]#

reduction_state: dict[str, Any]#

request_id: str#

schema_version: Literal[1]#

timing: dict[str, float]#

class ParallelismSpec#

Bases: StrictModel

cp_size: int#

distributed_backend: str#

dp_size: int | None#

ep_size: int#

fsdp: bool#

gpus_per_task: int | None#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pp_size: int#

sequence_parallel: bool#

tp_size: int#

validate_world_and_gpu_ownership()#

world_size: int#

class WorkerRecord#

Bases: StrictModel

boot_id: str#

campaign_id: str#

capabilities: dict[str, Any]#

current_request_id: str | None#

property endpoint: str#

heartbeat_at: datetime#

host: str#

model_config = {'extra': 'forbid', 'frozen': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parallelism: ParallelismSpec#

port: int#

schema_version: Literal[1]#

started_at: datetime#

state: WorkerState#

worker_id: str#

class WorkerState#

Bases: str, Enum

BUSY = 'busy'#

DRAINING = 'draining'#

FAILED = 'failed'#

IDLE = 'idle'#

STARTING = 'starting'#

__new__(value)#

prefix_cache_id(*, campaign_id, model, data, prefix_signature, prefix_length, parallelism)#

Stable identity reserved for a future topology-specific prefix cache.

Parameters:

campaign_id (str)
model (Any)
data (Any)
prefix_signature (Any)
prefix_length (int)
parallelism (Any)

Return type:

str