trinity.buffer package

Subpackages

• trinity.buffer.operators package
  • Subpackages
    • trinity.buffer.operators.filters package
      • Submodules
      • Module contents
    • trinity.buffer.operators.mappers package
      • Submodules
      • Module contents
  • Submodules
    • trinity.buffer.operators.data_juicer_operator module
      • DataJuicerOperator
    • trinity.buffer.operators.experience_operator module
      • ExperienceOperator
  • Module contents
    • ExperienceOperator
      • ExperienceOperator.process()
      • ExperienceOperator.create_operators()
      • ExperienceOperator.close()
    • RewardFilter
      • RewardFilter.__init__()
      • RewardFilter.process()
    • RewardSTDFilter
      • RewardSTDFilter.__init__()
      • RewardSTDFilter.process()
    • RewardShapingMapper
      • RewardShapingMapper.__init__()
      • RewardShapingMapper.process()
    • DataJuicerOperator
      • DataJuicerOperator.__init__()
      • DataJuicerOperator.process()
      • DataJuicerOperator.close()
• trinity.buffer.pipelines package
  • Submodules
    • trinity.buffer.pipelines.experience_pipeline module
      • get_input_buffers()
      • ExperiencePipeline
    • trinity.buffer.pipelines.task_pipeline module
      • check_and_run_task_pipeline()
      • TaskPipeline
  • Module contents
    • ExperiencePipeline
      • ExperiencePipeline.__init__()
      • ExperiencePipeline.prepare()
      • ExperiencePipeline.process()
      • ExperiencePipeline.close()
    • TaskPipeline
      • TaskPipeline.__init__()
      • TaskPipeline.convert_pipeline_config()
      • TaskPipeline.process()
      • TaskPipeline.close()
    • check_and_run_task_pipeline()
• trinity.buffer.reader package
  • Submodules
    • trinity.buffer.reader.file_reader module
      • DummyProgressBar
      • BaseFileReader
      • ExperienceFileReader
      • TaskFileReader
    • trinity.buffer.reader.queue_reader module
      • QueueReader
    • trinity.buffer.reader.sql_reader module
      • SQLReader
  • Module contents
• trinity.buffer.schema package
  • Submodules
    • trinity.buffer.schema.formatter module
      • ExperienceFormatter
      • TaskFormatter
      • SFTFormatter
      • DPOFormatter
    • trinity.buffer.schema.sql_schema module
      • TaskModel
      • ExperienceModel
      • SFTDataModel
      • DPODataModel
      • init_engine()
  • Module contents
    • init_engine()
• trinity.buffer.storage package
  • Submodules
    • trinity.buffer.storage.file module
      • FileStorage
    • trinity.buffer.storage.queue module
      • is_database_url()
      • is_json_file()
      • linear_decay_priority()
      • QueueBuffer
      • AsyncQueue
      • AsyncPriorityQueue
      • QueueStorage
    • trinity.buffer.storage.sql module
      • SQLStorage
      • SQLExperienceStorage
      • SQLTaskStorage
  • Module contents
• trinity.buffer.writer package
  • Submodules
    • trinity.buffer.writer.file_writer module
      • JSONWriter
    • trinity.buffer.writer.queue_writer module
      • QueueWriter
    • trinity.buffer.writer.sql_writer module
      • SQLWriter
  • Module contents

Submodules

• trinity.buffer.buffer module
  • get_buffer_reader()
  • get_buffer_writer()
• trinity.buffer.buffer_reader module
  • BufferReader
    • BufferReader.read()
    • BufferReader.read_async()
• trinity.buffer.buffer_writer module
  • BufferWriter
    • BufferWriter.write()
    • BufferWriter.write_async()
    • BufferWriter.acquire()
    • BufferWriter.release()
• trinity.buffer.utils module
  • retry_session()
  • default_storage_path()

Module contents

class trinity.buffer.BufferReader

Bases: ABC

Interface of the buffer reader.

abstract read(batch_size: int | None = None) → List

Read from buffer.

abstract async read_async(batch_size: int | None = None) → List

Read from buffer asynchronously.

class trinity.buffer.BufferWriter

Bases: ABC

Interface of the buffer writer.

abstract write(data: List) → None

Write to buffer.

abstract async write_async(data: List) → None

Write to buffer asynchronously.

abstract async acquire() → int

Acquire the buffer writer.

Returns:

The reference count of the buffer after acquiring.

Return type:

int

abstract async release() → int

Release the buffer writer. After release, the buffer writer can not be used again.

Returns:

The reference count of the buffer after releasing.

Return type:

int

trinity.buffer.get_buffer_reader(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferReader

Get a buffer reader for the given dataset name.

trinity.buffer.get_buffer_writer(storage_config: StorageConfig, buffer_config: BufferConfig) → BufferWriter

Get a buffer writer for the given dataset name.