trinity.buffer.schema.sql_schema module

Schema for SQLAlchemy models.

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)

task_desc = Column(None, String(), table=None)

workflow_type = Column(None, String(), table=None)

reward_type = Column(None, String(), table=None)

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

Bases: Base

SQLAlchemy model for Experience.

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

serialized_exp = Column(None, LargeBinary(), table=None)

prompt = Column(None, String(), table=None)

response = Column(None, String(), table=None)

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

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

priority = Column(None, Float(), table=None, default=ScalarElementColumnDefault(0.0))

to_experience() → Experience

Load the experience from the database.

classmethod from_experience(experience: Experience)

Save the experience to database.

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

Bases: Base

SQLAlchemy model for SFT data.

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

serialized_exp = Column(None, LargeBinary(), table=None)

messages = Column(None, String(), table=None)

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

to_experience() → Experience

Load the experience from the database.

classmethod from_messages(messages: list[dict], tokenizer: Any, chat_template: str | None = None) → SFTDataModel

Convert a list of messages into a single instance of SFT data.

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

Bases: Base

SQLAlchemy model for DPO data.

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

serialized_exp = Column(None, LargeBinary(), table=None)

chosen = Column(None, LargeBinary(), table=None)

rejected = Column(None, LargeBinary(), table=None)

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

to_experience() → Experience

Load the experience from the database.

trinity.buffer.schema.sql_schema.create_dynamic_table(algorithm_type: str | None, table_name: str) → Any

Create a dynamic table based on the provided algorithm type and table name.