trinity.explorer package

trinity.explorer package#

Subpackages#

Submodules#

Module contents#

Explorer package exports.

class trinity.explorer.Explorer(config: Config)[source]#

Bases: object

Responsible for exploring the taskset.

__init__(config: Config)[source]#
async benchmark() → bool[source]#

Benchmark the model checkpoints.

async eval()[source]#

Evaluation on all evaluation data samples.

async explore() → str[source]#
The timeline of the exploration process:
<β€”β€”β€”β€”β€”β€”β€”β€”β€”β€”β€” one period β€”β€”β€”β€”β€”β€”β€”β€”β€”β€”β€”β€”-> |
explorer | <β€”β€”β€”β€”β€”- step_1 ————–> | |
| <β€”β€”β€”β€”β€”- step_2 ————–> | |
… |
| <β€”β€”β€”β€”β€”- step_n β€”β€”β€”β€”β€”> | |
| <β€”β€”β€”β€”β€”β€”β€”- eval ——————–> | <– sync –> |

|--------------------------------------------------------------------------------------|

trainer | <– idle –> | <– step_1 –> | <– step_2 –> | … | <– step_n –> | <– sync –> |

async explore_step() → bool[source]#
async finish_current_steps() → None[source]#
classmethod get_actor(config: Config)[source]#

Get a Ray actor for the explorer.

async get_weight(name: str) → Tensor[source]#

Get the weight of the loaded model (For checkpoint weights update).

async is_alive() → bool[source]#

Check if the explorer is alive.

need_eval() → bool[source]#
async need_sync() → bool[source]#
async prepare() → None[source]#

Preparation before running.

async save_checkpoint() → None[source]#
async serve() → None[source]#

Run the explorer in serving mode.

In serving mode, the explorer starts an OpenAI compatible server to handle requests. Agent applications can be deployed separately and interact with the explorer via the API.

import openai


client = openai.OpenAI(
    base_url=f"{explorer_server_url}/v1",
    api_key="EMPTY",
)
response = client.chat.completions.create(
    model=config.model.model_path,
    messages=[{"role": "user", "content": "Hello!"}]
)
async set_state_dict_meta(state_dict_meta: List)[source]#

Set the state_dict meta on all model workers for NCCL weight sync.

Must be called after setup_weight_sync_group and before the first sync_model_weights call.

async setup_model_level_weight_sync_group()[source]#

Setup process group for each model, only used in serve mode.

async setup_weight_sync_group(master_address: str, master_port: int, world_size: int = None, group_name: str = None, timeout: int = None)[source]#
async shutdown() → None[source]#
async sync_weight() → None[source]#

Synchronize model weights.

async teardown_weight_sync_group()[source]#

Destroy the NCCL process group on all model workers.

class trinity.explorer.RolloutCoordinator(config: Config)[source]#

Bases: object

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

__init__(config: Config)[source]#

Create a coordinator with internally managed scheduler and pipeline.

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

Abort one batch and cleanup its running and staged state.

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

Finalize one eval batch and return aggregated eval metrics.

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

Finalize one train batch and return aggregated metrics.

classmethod get_actor(config: Config) → ActorHandle[RolloutCoordinator][source]#

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

async prepare() → None[source]#

Initialize the owned pipeline and scheduler.

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

Process one batch of experience payloads through the pipeline.

async shutdown() → None[source]#

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[source]#

Register a new batch and schedule its tasks.