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(host: str, port: int, group_name: str, backend: str | Backend = 'nccl', timeout: float | None = None, world_size: int = -1, rank: int = -1, 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.remove_right_units(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.math_eval_utils module

Utility functions for strictly parsing and evaluating mathematical answers.

This module is a modified and simplified version of the official evaluation code for Qwen2.5-Math, designed for easier standalone use.

Original source: https://github.com/QwenLM/Qwen2.5-Math

Key modifications include: 1. Retained only the core parsing logic for the common qwen_boxed prompt format. 2. Consolidated essential parsing and evaluation functions from multiple files

into this single module.

3. Simplified benchmark handling and conditional logic for broader applicability.

4. Simplified or removed calls to external tools like TIR.

trinity.utils.math_eval_utils.verify_math_answer(response_text: str, ground_truth: str) → Tuple[float, Dict[str, Any]]

Strictly compare the equality of response and groundtruth.

trinity.utils.math_eval_utils.extract_answer(response_text: str) → str | None

Extract the equation from the string.

trinity.utils.math_eval_utils.strip_string(input_str: str | None) → str | None

Clean and normalize math answer strings.

trinity.utils.math_eval_utils.fix_fracs(string)

trinity.utils.math_eval_utils.fix_a_slash_b(string)

trinity.utils.math_eval_utils.fix_sqrt(string)

trinity.utils.math_eval_utils.convert_word_number(text: str) → str

trinity.utils.math_eval_utils.math_equal(prediction: str | None, reference: str | None) → bool

Checks the mathematical equality of two strings by trying different methods.

trinity.utils.math_eval_utils.numeric_equal(prediction: float, reference: float) → bool

trinity.utils.math_eval_utils.symbolic_equal(a: str, b: str) → bool

Compares two strings for symbolic equivalence using sympy.

trinity.utils.monitor module

Monitor

trinity.utils.monitor.gather_metrics(metric_list: List[Dict], prefix: str) → Dict

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]

classmethod default_args() → Dict

Return default arguments for the monitor.

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

Bases: Monitor

__init__(project: str, group: 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, group: str, name: str, role: str, config: Config | None = None)

Bases: Monitor

Monitor with Weights & Biases.

Parameters:

• base_url (Optional[str]) – The base URL of the W&B server. If not provided, use the environment variable WANDB_BASE_URL.

• api_key (Optional[str]) – The API key for W&B. If not provided, use the environment variable WANDB_API_KEY.

__init__(project: str, group: 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

classmethod default_args() → Dict

Return default arguments for the monitor.

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

Bases: Monitor

Monitor with MLflow.

Parameters:

• uri (Optional[str]) – The tracking server URI. If not provided, the default is http://localhost:5000.

• username (Optional[str]) – The username to login. If not provided, the default is None.

• password (Optional[str]) – The password to login. If not provided, the default is None.

__init__(project: str, group: 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

classmethod default_args() → Dict

Return default arguments for 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