trinity.buffer.schema
Submodules
trinity.buffer.schema.sql_schema module
Schema for SQLAlchemy models.
- 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)
- 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)[source]
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 [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)
- 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 [source]
Load the experience from the database.
- classmethod from_messages(messages: list[dict], tokenizer: Any, chat_template: str | None = None) SFTDataModel [source]
Convert a list of messages into a single instance of SFT data.
- 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)
- 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 [source]
Load the experience from the database.
- trinity.buffer.schema.sql_schema.create_dynamic_table(algorithm_type: AlgorithmType | None, table_name: str) Any [source]
Create a dynamic table based on the provided algorithm type and table name.
Module contents
- trinity.buffer.schema.create_dynamic_table(algorithm_type: AlgorithmType | None, table_name: str) Any [source]
Create a dynamic table based on the provided algorithm type and table name.
- class trinity.buffer.schema.Base(**kwargs: Any)
Bases:
object
The base class of the class hierarchy.
When called, it accepts no arguments and returns a new featureless instance that has no instance attributes and cannot be given any.
- __init__(**kwargs: Any) None
A simple constructor that allows initialization from kwargs.
Sets attributes on the constructed instance using the names and values in
kwargs
.Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.
- metadata: MetaData = MetaData()
- registry: registry = <sqlalchemy.orm.decl_api.registry object>