trinity.utils

Submodules

trinity.utils.distributed module

For distributed training with multiple process groups.

trinity.utils.distributed.is_ipv6_address(ip_str: str) → bool

trinity.utils.distributed.init_process_group(backend: str | Backend | None = None, init_method: str | None = None, timeout: float | None = None, world_size: int = -1, rank: int = -1, store: Store | None = None, group_name: str | None = None, pg_options: Any | None = None)

trinity.utils.dlc_utils module

class trinity.utils.dlc_utils.ClusterStatus

Bases: object

__init__()

finish() → None

running() → bool

trinity.utils.dlc_utils.get_dlc_env_vars() → dict

trinity.utils.dlc_utils.is_running() → bool

Check if ray cluster is running.

trinity.utils.dlc_utils.wait_for_ray_setup() → None

trinity.utils.dlc_utils.wait_for_ray_worker_nodes(world_size: int) → None

trinity.utils.dlc_utils.setup_ray_cluster(namespace: str)

trinity.utils.dlc_utils.stop_ray_cluster(namespace: str)

Stop the ray cluster by sending a signal to the cluster status actor.

trinity.utils.eval_utils module

trinity.utils.eval_utils.simple_answer_parser(response: str) → str

trinity.utils.eval_utils.find_boxed_answer(raw_answer, timeout=10)

Find answers from solutions where the answers are enclosed in LaTeX’s boxed tag

Parameters:

• raw_answer (str) – raw answer from model

• timeout (int) – timeout in seconds for regex

Returns:

answer if found, otherwise None

Return type:

str

trinity.utils.eval_utils.extract_solution(solution_str)

Extract the equation from the solution string.

trinity.utils.eval_utils.validate_equation(equation_str, available_numbers)

Validate that equation only uses available numbers and each number once.

trinity.utils.eval_utils.evaluate_equation(equation_str)

Safely evaluate the arithmetic equation using eval() with precautions.

trinity.utils.eval_utils.validate_think_pattern(text)

Validate whether the <think> </think> tag is properly formatted.

trinity.utils.eval_utils.compute_score(solution_str, ground_truth) → float

trinity.utils.eval_utils.is_equiv(str1, str2, verbose=False)

trinity.utils.eval_utils.remove_boxed(s)

trinity.utils.eval_utils.last_boxed_only_string(string)

trinity.utils.eval_utils.fix_fracs(string)

trinity.utils.eval_utils.fix_a_slash_b(string)

trinity.utils.eval_utils.remove_right_units(string)

trinity.utils.eval_utils.fix_sqrt(string)

trinity.utils.eval_utils.strip_string(string)

trinity.utils.log module

Logging configuration for vLLM.

class trinity.utils.log.NewLineFormatter(fmt, datefmt=None)

Bases: Formatter

Adds logging prefix to newlines to align multi-line messages.

__init__(fmt, datefmt=None)

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

trinity.utils.log.get_logger(name: str, level: int = 10) → Logger

Get a logger with the given name and level.

Parameters:

• name (str) – The name of the logger.

• level (int, optional) – The level of the logger. Defaults to logging.DEBUG.

Returns:

The logger with the given name and level.

Return type:

logging.Logger

trinity.utils.monitor module

Monitor

class trinity.utils.monitor.Monitor(project: str, name: str, role: str, config: Config | None = None)

Bases: ABC

__init__(project: str, name: str, role: str, config: Config | None = None) → None

abstract log_table(table_name: str, experiences_table: DataFrame, step: int)

Log a table

abstract log(data: dict, step: int, commit: bool = False) → None

Log metrics.

abstract close() → None

Close the monitor

calculate_metrics(data: dict[str, List[float] | float], prefix: str | None = None) → dict[str, float]

class trinity.utils.monitor.TensorboardMonitor(project: str, name: str, role: str, config: Config | None = None)

Bases: Monitor

__init__(project: str, name: str, role: str, config: Config | None = None) → None

log_table(table_name: str, experiences_table: DataFrame, step: int)

Log a table

log(data: dict, step: int, commit: bool = False) → None

Log metrics.

close() → None

Close the monitor

class trinity.utils.monitor.WandbMonitor(project: str, name: str, role: str, config: Config | None = None)

Bases: Monitor

__init__(project: str, name: str, role: str, config: Config | None = None) → None

log_table(table_name: str, experiences_table: DataFrame, step: int)

Log a table

log(data: dict, step: int, commit: bool = False) → None

Log metrics.

close() → None

Close the monitor

trinity.utils.plugin_loader module

Load modules from custom directory

trinity.utils.plugin_loader.load_plugins(plugin_dir: str) → None

Load plugin modules from a directory.

trinity.utils.plugin_loader.load_from_file(file_path: str)

Load modules from a Python file

Parameters:

file_path (str) – The python file path.

Returns:

The loaded module.

Return type:

Any

trinity.utils.registry module

class trinity.utils.registry.Registry(name: str)

Bases: object

A class for registry.

__init__(name: str)

Parameters:

name (str) – The name of the registry.

property name: str

Get name of current registry.

Returns:

The name of current registry.

Return type:

str

property modules: dict

Get all modules in current registry.

Returns:

A dict storing modules in current registry.

Return type:

dict

list() → None

Logging the list of module in current registry.

get(module_key) → Any

Get module named module_key from in current registry. If not found, return None.

Parameters:

module_key (str) – specified module name

Returns:

the module object

Return type:

Any

register_module(module_name: str, module_cls: Type | None = None, force=False, lazy=False)

Register module class object to registry with the specified module name.

Parameters:

• module_name (str) – The module name.

• module_cls (Type) – module class object

• force (bool) – Whether to override an existing class with the same name. Default: False.

• lazy (bool) – Whether to register the module class object lazily. Default: False.

Example

trinity.utils.timer module

Timer context manager

class trinity.utils.timer.Timer(metrics_dict, key_name)

Bases: object

__init__(metrics_dict, key_name)

Module contents