trinity.buffer.schema.sql_schema module

trinity.buffer.schema.sql_schema module#

SQLAlchemy models for different data.

class trinity.buffer.schema.sql_schema.TaskModel(**kwargs: Any)[source]#

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

class trinity.buffer.schema.sql_schema.ExperienceModel(**kwargs: Any)[source]#

Bases: Base

SQLAlchemy model for Experience.

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

prompt = Column(None, Text(), table=None)#

response = Column(None, Text(), table=None)#

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

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

experience_bytes = Column(None, LargeBinary(), table=None)#

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

to_experience() → Experience[source]#: Load the experience from the database.

classmethod from_experience(experience: Experience)[source]#: Save the experience to database.

class trinity.buffer.schema.sql_schema.SFTDataModel(**kwargs: Any)[source]#

Bases: Base

SQLAlchemy model for SFT data.

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

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

experience_bytes = Column(None, LargeBinary(), table=None)#

to_experience() → Experience[source]#: Load the experience from the database.

classmethod from_experience(experience: Experience)[source]#: Save the experience to database.

class trinity.buffer.schema.sql_schema.DPODataModel(**kwargs: Any)[source]#

Bases: Base

SQLAlchemy model for DPO data.

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)#

experience_bytes = Column(None, LargeBinary(), table=None)#

to_experience() → Experience[source]#: Load the experience from the database.

classmethod from_experience(experience: Experience)[source]#: Save the experience to database.

trinity.buffer.schema.sql_schema.init_engine(db_url: str, table_name, schema_type: str | None) → Tuple[source]#: Get the sqlalchemy engine.