trinity.buffer.pipelines.experience_pipeline module

trinity.buffer.pipelines.experience_pipeline module#

trinity.buffer.pipelines.experience_pipeline.get_input_buffers(pipeline_config: ExperiencePipelineConfig) → Dict[source]#

Get input buffers for the experience pipeline.

trinity.buffer.pipelines.experience_pipeline.default_input_save_path(config: Config) → str[source]#

Compute the default input_save_path for the experience pipeline.

This is the single source of truth for where the pipeline writes its input database when save_input is enabled but no path is provided: a SQLite database at <checkpoint_job_dir>/buffer/explorer_output.db.

It is derived purely from the config components (no Ray/GPU side effects), so callers that load a config without a full validation pass — e.g. trinity view — resolve the exact same path the pipeline writes to at run time. The abspath mirrors GlobalConfigValidator making a relative checkpoint_root_dir absolute, keeping the two contexts consistent.

class trinity.buffer.pipelines.experience_pipeline.ExperiencePipeline(config: Config)[source]#

Bases: object

A class to process experiences.

__init__(config: Config)[source]#
async prepare() → None[source]#
async process(exp_bytes: bytes) → Dict[source]#

Process a batch of experiences.

Parameters:

exp_bytes (bytes) – Serialized experiences to process. These experiences are typically generated by an explorer in one step.

Returns:

A dictionary containing metrics collected during the processing of experiences.

Return type:

Dict

async process_serialized_chunks(exp_chunks: list[bytes]) → Dict[source]#

Process a batch assembled from multiple serialized task payloads.

async close() → None[source]#