trinity.explorer.rollout_coordinator module

Rollout coordinator for async batch submission and finalize.

class trinity.explorer.rollout_coordinator.BatchLifecycleState(*values)

基类：str, Enum

Lifecycle states for one submitted batch.

PENDING = 'pending'

RUNNING = 'running'

FINALIZING = 'finalizing'

FINALIZED = 'finalized'

ABORTED = 'aborted'

class trinity.explorer.rollout_coordinator.BatchState(batch_id: int | str, batch_type: ~typing.Literal['train', 'eval'], expected_task_count: int, statuses: ~typing.Dict[int | str, ~typing.Any] = <factory>, min_wait_num: int | None = None, state: ~trinity.explorer.rollout_coordinator.BatchLifecycleState = BatchLifecycleState.PENDING, final_result: dict | None = None, finalize_lock: ~asyncio.locks.Lock = <factory>)

基类：object

In-memory state tracked for one train or eval batch.

batch_id: int | str

batch_type: Literal['train', 'eval']

expected_task_count: int

statuses: Dict[int | str, Any]

min_wait_num: int | None = None

state: BatchLifecycleState = 'pending'

final_result: dict | None = None

finalize_lock: Lock

property completed_task_count: int

Return the number of completed tasks tracked by status.

__init__(batch_id: int | str, batch_type: ~typing.Literal['train', 'eval'], expected_task_count: int, statuses: ~typing.Dict[int | str, ~typing.Any] = <factory>, min_wait_num: int | None = None, state: ~trinity.explorer.rollout_coordinator.BatchLifecycleState = BatchLifecycleState.PENDING, final_result: dict | None = None, finalize_lock: ~asyncio.locks.Lock = <factory>) → None

class trinity.explorer.rollout_coordinator.RolloutCoordinator(config: Config, rollout_model: List[InferenceModel], auxiliary_models: List[List[InferenceModel]] | None = None)

基类：object

Own scheduler-side batch state and expose batch-level finalize APIs.

__init__(config: Config, rollout_model: List[InferenceModel], auxiliary_models: List[List[InferenceModel]] | None = None)

Create a coordinator with internally managed scheduler and pipeline.

async prepare() → None

Initialize the owned pipeline and scheduler.

async shutdown() → None

Stop background work and close owned dependencies.

async submit_batch(*, batch_id: int | str, tasks: list[Task], batch_type: Literal['train', 'eval'], min_wait_num: int | None = None) → None

Register a new batch and schedule its tasks.

async finalize_train_batch(batch_id: int, *, timeout: float | None = None) → dict

Finalize one train batch and return aggregated metrics.

async finalize_eval_batch(batch_id: str, *, timeout: float | None = None) → dict

Finalize one eval batch and return aggregated eval metrics.

async abort_batch(batch_id: int | str, *, reason: str, keep_partial_results: bool = False) → None

Abort one batch and cleanup its running and staged state.

async process_experiences(payloads: list[bytes]) → dict

Process one batch of experience payloads through the pipeline.

classmethod get_actor(config: Config, models: List, auxiliary_models: List) → ActorHandle

Init rollout coordinator for the task-event-completion path.