trinity.buffer.schema.sql_schema module

SQLAlchemy models for different data.

class trinity.buffer.schema.sql_schema.TaskModel(**kwargs: Any)

Bases: Base

Model for storing tasks in SQLAlchemy.

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

raw_task = Column(None, JSON(), table=None, nullable=False)

classmethod from_dict(dict: Dict)

class trinity.buffer.schema.sql_schema.ExperienceModel(**kwargs: Any)

Bases: Base

SQLAlchemy model for Experience metadata.

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

timestamp = Column(None, DateTime(), table=None, server_default=DefaultClause(<sqlalchemy.sql.functions.now at 0x7f9f8a1e8110; now>, for_update=False))

task_id = Column(None, String(length=64), table=None)

run_id = Column(None, Integer(), table=None)

msg_id = Column(None, String(length=64), table=None)

model_version = Column(None, Integer(), table=None)

reward = Column(None, Float(), table=None)

consumed = Column(None, Integer(), table=None, default=ScalarElementColumnDefault(0))

to_experience(blob_bytes: bytes) → Experience

Load the experience from metadata + blob bytes.

classmethod from_experience(experience: Experience)

Create meta row from experience (blob stored separately).

class trinity.buffer.schema.sql_schema.BlobModel(**kwargs: Any)

Bases: Base

Unified blob storage model for all experience types.

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

experience_bytes = Column(None, LargeBinary(), table=None, nullable=False)

class trinity.buffer.schema.sql_schema.SFTDataModel(**kwargs: Any)

Bases: Base

SQLAlchemy model for SFT data metadata.

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

message_list = Column(None, JSON(), table=None)

to_experience(blob_bytes: bytes) → Experience

Load the experience from metadata + blob bytes.

classmethod from_experience(experience: Experience)

Create meta row from experience (blob stored separately).

class trinity.buffer.schema.sql_schema.DPODataModel(**kwargs: Any)

Bases: Base

SQLAlchemy model for DPO data metadata.

id = Column(None, Integer(), table=None, primary_key=True, nullable=False)

chosen_message_list = Column(None, JSON(), table=None)

rejected_message_list = Column(None, JSON(), table=None)

to_experience(blob_bytes: bytes) → Experience

Load the experience from metadata + blob bytes.

classmethod from_experience(experience: Experience)

Create meta row from experience (blob stored separately).

async trinity.buffer.schema.sql_schema.init_async_engine(db_url: str, table_name: str, schema_type: str | None) → Tuple

Create an async SQLAlchemy engine and table classes.

Returns:

(async_engine, table_cls) For experience/sft/dpo schema: (async_engine, meta_cls, blob_cls)

Return type:

For task schema