zensols.lmtask package

Submodules

zensols.lmtask.app module

Task-specialized language model training and inference.

class zensols.lmtask.app.Application(config_factory, task_factory)

Bases: object

Task-specialized language model training and inference.

__init__(config_factory, task_factory)

benchmark()

Test the model and output benchmark files.

property benchmark_runner: BenchmarkRunner

Used to lTrain, test, score and render one configured LMTask benchmark.

config_factory: ConfigFactory

Used to create configured application and training resources.

dataset_sample(max_sample=1)

Print sample(s) of the configured (--config) dataset.

Parameters:

max_sample (int) – the number of sample to print

instruct(task_name, instruction, role=None, output_format=None)

Generate text by inferencing with the model.

Parameters:

• task_name (str) – the task that generates the result

• instruction (str) – added to the prompt to instruction the model

• role (str) – the role the model takes

• output_format (_Format) – data format for the output

report(configs, base_dir=None)

Create a report.

Parameters:

base_dir (Path) – root directory to create files, or render if not given

show_task(task_name=None)

Print the configuration of a task if --name is given, otherise a list of available tasks.

Parameters:

task_name (str) – the task that creates the prompt and parses the result

show_trainer(long_output=False)

Print configuration and dataset stats of the configured (--config) trainer.

Parameters:

long_output (bool) – verbosity

stream(task_name, prompt)

Stream generated text from the model.

Parameters:

• task_name (str) – the task that generates the result

• prompt (str) – the prompt text as input to the model

task_factory: TaskFactory

Create tasks used to fulfill CLI requests.

test(output_file=PosixPath('-'), output_format=None)

Test a trained model on a configured (--config) dataset.

Parameters:

• output_file (Path) – output file name, - for standard out

• output_format (_Format) – data format for the output

property tester: Tester

The currently configured test.Tester.

train()

Train a new model on a configured (--config) dataset.

property trainer: Trainer

The currently configured train.Trainer.

zensols.lmtask.benchmark module

Reproducible LMTask benchmark execution and reporting.

The benchmark JSON file is the source of truth. Markdown is rendered from that structured result so reports can be regenerated without retraining a model.

class zensols.lmtask.benchmark.BenchmarkResult(name, task_name, model_name, config_file, created, git, environment, datasets, training, testing, metrics, notes=())

Bases: Dictable

Complete machine-readable benchmark record.

__init__(name, task_name, model_name, config_file, created, git, environment, datasets, training, testing, metrics, notes=())

config_file: str

The LMTask configuration file used for the benchmark.

created: str

The timezone-aware ISO-8601 timestamp recording when the benchmark result was created.

datasets: DatasetSplitsResult

The train, validation and held-out test split metadata.

property describer: DataDescriber

Create a data describer for the results in this benchmark.

environment: EnvironmentResult

The software, CUDA and hardware environment information.

git: GitResult

The LMTask repository revision and working-tree state.

metrics: MetricsResult

The performance metrics for this benchmark.

model_name: str

The human-readable model name used in reports.

name: str

Name of the run copied from BenchmarkRunner.name.

notes: tuple[str, ...] = ()

Optional free-form benchmark annotations.

task_name: str

The human-readable task or dataset name used in reports.

testing: TestingResult

The held-out task-inference metadata.

training: TrainingResult

The model-training diagnostics and persisted adapter metadata.

class zensols.lmtask.benchmark.BenchmarkRunner(name, task_name, model_name, config_file, trainer, tester, metrics_calculator, executor, result_dir, template_dir, temporary_dir, detail_template='overview.md.jinja2', prediction_format='jsonl', class_count_column=None)

Bases: Dictable

Train, test, score and render one configured LMTask benchmark.

__init__(name, task_name, model_name, config_file, trainer, tester, metrics_calculator, executor, result_dir, template_dir, temporary_dir, detail_template='overview.md.jinja2', prediction_format='jsonl', class_count_column=None)

class_count_column: str | None = None

The dataset column used for per-split class counts; None disables class counting.

clear()

Remove all cached data.

config_file: Path

The configuration file used for training and testing.

property datasets: DatasetSplitsResult

Collect metadata for configured train, validation and test splits.

detail_template: str = 'overview.md.jinja2'

The Jinja2 template filename used to render the per-benchmark Markdown report.

executor: Executor

The command executor used to collect external provenance such as Git and NVIDIA information.

property has_cached_result: bool

Whether there has been a benchmark cached for this instance yet.

metrics_calculator: MetricsCalculator

The task-specific metric implementation selected by the trainconf configuration.

model_name: str

Name of the model used for training/testing.

name: str

Name of the run, usually taken from the lmtask_benchmark:name, which is composed of the dataset and model name.

prediction_format: str = 'jsonl'

The persisted prediction format; reserved for future formats beyond JSONL.

property result: BenchmarkResult

Train, test, score, persist and render this benchmark.

result_dir: Path

Output director for the generated benchmark files.

save_benchmark()

Write the benchmark files.

Return type:

BenchmarkResult

task_name: str

Name of the task.

template_dir: Path

The directory containing benchmark Jinja2 templates.

temporary_dir: Path

Directory to store temporary files.

tester: Tester

The configured LMTask tester used for held-out inference.

trainer: Trainer

The configured LMTask trainer.

class zensols.lmtask.benchmark.DatasetSplit(name, examples, class_counts=None)

Bases: Dictable

Dataset split size and optional class distribution.

__init__(name, examples, class_counts=None)

class_counts: dict[str, int] | None = None

The class label counts or None when it does not apply.

property class_counts_to_series: Series | None

The class counts as a series.

examples: int

The number of examples in the split.

name: str

Logical split name, such as train, validation or test.

class zensols.lmtask.benchmark.DatasetSplitsResult(splits)

Bases: Dictable

Contains the dataset splits.

__init__(splits)

splits: tuple[DatasetSplit, ...]

The dataset splits.

class zensols.lmtask.benchmark.EnvironmentResult(python, os, kernel, cuda_visible_devices, cuda_runtime, nvidia_driver, packages=<factory>, gpus=())

Bases: Dictable

Software and hardware environment used for the benchmark.

__init__(python, os, kernel, cuda_visible_devices, cuda_runtime, nvidia_driver, packages=<factory>, gpus=())

cuda_runtime: str | None

The CUDA runtime version used by the installed PyTorch build; None when CUDA is unavailable.

cuda_visible_devices: str | None

The value of CUDA_VISIBLE_DEVICES; None when not set.

gpus: tuple[GpuResult, ...] = ()

The CUDA devices visible to the benchmark process.

kernel: str

The operating-system kernel version.

nvidia_driver: str | None

The NVIDIA driver version reported by nvidia-smi; None when unavailable.

os: str

The operating-system description.

packages: dict[str, str]

The relevant Python package names mapped to exact installed versions.

python: str

The Python interpreter version.

class zensols.lmtask.benchmark.GitResult(commit, describe, dirty)

Bases: Dictable

Repository revision used for the benchmark.

__init__(commit, describe, dirty)

commit: str

The full Git commit SHA.

describe: str

The human-readable revision returned by git describe.

dirty: bool

Whether the repository contained uncommitted or untracked changes when the benchmark was generated.

class zensols.lmtask.benchmark.GpuResult(index, name, total_memory, peak_allocated=None, peak_reserved=None)

Bases: Dictable

One CUDA device visible to the benchmark process.

__init__(index, name, total_memory, peak_allocated=None, peak_reserved=None)

index: int

The CUDA device index visible to the current process.

name: str

The GPU card name.

peak_allocated: int | None = None

The peak memory allocated by PyTorch during benchmark execution in bytes; None when unavailable.

peak_reserved: int | None = None

The peak memory reserved by the PyTorch CUDA allocator during benchmark execution in bytes; None when unavailable.

total_memory: int

The total device memory in bytes.

class zensols.lmtask.benchmark.TestingResult(test_result, predictions_file)

Bases: Dictable

Held-out task testing facts.

__init__(test_result, predictions_file)

property count: int

Nummber of examples in the test set.

property elapsed_seconds: float

Time taken to evaluate the test set.

predictions_file: Path

The path to the predictions JSONL file.

test_result: TestResult

Result from test.Tester.

class zensols.lmtask.benchmark.TrainingResult(elapsed_seconds, global_step, training_loss, metrics, result_dir, adapter_size=None)

Bases: Dictable

Training facts captured from LMTask and the generated adapter.

__init__(elapsed_seconds, global_step, training_loss, metrics, result_dir, adapter_size=None)

adapter_size: int | None = None

The size in bytes of adapter_model.safetensors; None when unavailable.

elapsed_seconds: float

Time in seconds it took to train.

global_step: int

The final trainer global step.

metrics: dict[str, Any]

The task-specific benchmark metrics computed from tester predictions.

result_dir: str

The directory where JSON, JSONL and Markdown benchmark artifacts are written.

training_loss: float

The final training loss reported by the trainer.

zensols.lmtask.cli module

Command line entry point to the application.

class zensols.lmtask.cli.ApplicationFactory(*args, **kwargs)

Bases: ApplicationFactory

__init__(*args, **kwargs)

classmethod get_application(args=None)

Get a text generator instance.

classmethod get_benchmark_runner()

classmethod get_task_factory()

Get the factory that creates tasks.

Return type:

TaskFactory

zensols.lmtask.dataset module

An implementation of a dataset generator task.TaskDatasetFactory.

class zensols.lmtask.dataset.LoadedTaskDatasetFactory(task, text_field='text', messages_field='messages', source=None, load_args=<factory>, pre_process=None, post_process=None)

Bases: TaskDatasetFactory

A utility class meant to be created from an application configuration. This class creates a dataframe used by Trainer and optionally does post processing (i.e. filtering and mapping).

__init__(task, text_field='text', messages_field='messages', source=None, load_args=<factory>, pre_process=None, post_process=None)

static clear_generator_cache()

load_args: dict[str, Any]

Additional arguments given to datasets.load_dataset().

post_process: str | Callable = None

Code to call after the dataset is created and the task has applied any template.

See:

pre_process

pre_process: str | Callable = None

Code to call after the dataset is created but before the task applies any template. If this is a string exec() is used to evaluate it. Otherwise it is treated as a callable where the old dataset is the input and the returned value is the replaced dataset.

source: str | Path | Stash | DataFrame | Dataset = None

Used as the source data in the created dataset.

zensols.lmtask.gemma4 module

Temporary fix to get Gemma 4 to fine-tune with LoRA.

class zensols.lmtask.gemma4.Gemma4GeneratorResource(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>)

Bases: GeneratorResource

A class to “monkey patch” the an open issue with using Gemma 4 for text only SFT with LoRA.

Link:

https://github.com/huggingface/peft/issues/3129

Link:

https://huggingface.co/google/gemma-4-31B/discussions/3

__init__(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>)

class zensols.lmtask.gemma4.Gemma4HFTrainerResource(model_args=None, cache=True, generator_resource=None, peft_config=None)

Bases: HFTrainerResource

See:

Gemma4GeneratorResource

__init__(model_args=None, cache=True, generator_resource=None, peft_config=None)

zensols.lmtask.generate module

Facade to HuggingFace text generation.

class zensols.lmtask.generate.CachingGenerator(_delegate, _stash, _hasher=<factory>)

Bases: TextGenerator

A generator that caches response using a hash of the model input as a key.

__init__(_delegate, _stash, _hasher=<factory>)

clear()

Clear any model state.

class zensols.lmtask.generate.ConfigGeneratorResource(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>, code_tokenizer=None, code_model=None)

Bases: GeneratorResource

Allows inline Python codee to configure tokenizers and models.

__init__(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>, code_tokenizer=None, code_model=None)

code_model: str = None

The Python code to configure the model. When called, model will be an instance of PreTrainedModel.

code_tokenizer: str = None

The Python code to configure the tokenizer. When called, tokenizer will be an instance of PreTrainedTokenizer.

class zensols.lmtask.generate.ConstantTextGenerator(config_factory, response, post_init_source=None)

Bases: TextGenerator

A generator that responses with response with every generation call for the purpose of debugging.

__init__(config_factory, response, post_init_source=None)

config_factory: ConfigFactory

Used to set optional mock attributes in post_init_source.

post_init_source: str = None

Python source code to run in the initializer.

response: str

The fixed response for each generate() call or the prompt if None.

class zensols.lmtask.generate.GenerateTask(name, description, request_class, response_class, generator, resource, train_add_eos=False)

Bases: Task

Uses a TextGenerator (generator) to generate a response.

__init__(name, description, request_class, response_class, generator, resource, train_add_eos=False)

clear()

Clear any generator state or cache.

generator: TextGenerator

A client facade of a chat or instruct-based large language model.

resource: GeneratorResource

The class that creates resources such as the tokenizer and model. This should be the base model resource so training tasks do not depend on the model they will eventually create.

This is also used by InstructTask for its chat template.

train_add_eos: bool = False

Whether to add the end of sentence token to the output when mapping the dataset for training. Newer versions of the trl.SFTTrainer class add (and force) this already.

class zensols.lmtask.generate.GeneratorOutput(model_output, parsed)

Bases: Dictable

Container instances of model output from TextGenerator.

__init__(model_output, parsed)

model_output: str

The unmodified raw model output.

parsed: tuple[str, ...]

The processed model output with special tokens stripped.

class zensols.lmtask.generate.GeneratorResource(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>)

Bases: Dictable

A client facade of a chat-based large language model.

__init__(name, model_id, model_class=<class 'transformers.models.auto.modeling_auto.AutoModelForCausalLM'>, tokenizer_class=<class 'transformers.models.auto.tokenization_auto.AutoTokenizer'>, peft_model_id=None, peft_model_class=<class 'peft.auto.AutoPeftModelForCausalLM'>, model_desc=None, system_role_name='system', tokenizer_args=<factory>, model_args=<factory>)

clear(include_cuda=True)

Clear the cached tokenizer, model and optionally CUDA.

classmethod get_model_path(model_id, parent=None)

Create a normalized file name from a HF model ID string useful for creating checkpoint directory names.

Parameters:

• model_id (str) – the model ID (i.e. meta-llama/Llama-3.1-8B)

• parent (Path) – the base directory used in the return value if given

Return type:

Path

property model: PreTrainedModel

The LLM.

model_args: dict[str, Any]

The arguments given to the HF model from_pretrained method.

model_class

The class used to create the model with from_pretrained().

alias of AutoModelForCausalLM

model_desc: str = None

A human readable description of the model this resource contains.

property model_file_name: str

A normalized file name friendly string based on model_desc.

model_id: str | Path

The HF model ID or path to the model.

name: str

The section of this configured instance in the application config.

peft_model_class

The class used to create the model with from_pretrained().

alias of AutoPeftModelForCausalLM

peft_model_id: str | Path = None

The HF model ID or path to the Peft model or None if there is none.

system_role_name: str = 'system'

The default name of the system’s role.

property tokenizer: PreTrainedTokenizer

The model’s tokenzier.

tokenizer_args: dict[str, Any]

The arguments given to the HF tokenizer from_pretrained method.

tokenizer_class

The class used to create the tokenizer with from_pretrained().

alias of AutoTokenizer

class zensols.lmtask.generate.ModelTextGenerator(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>)

Bases: TextGenerator

An implementation that uses HuggingFace framework classes from GeneratorResource to answer queries.

__init__(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>)

chat_template_args: dict[str, Any]

Arguments given to apply_chat_template. Some models require chat templates that all instruct.InstructTask should add. For example, Qwen 3 always needs add_generation_prompt=True.

This only is used (and should only be set) in generators used by instruct tasks.

clear()

Clear any model state.

generate_params: dict[str, Any]

Parameters given to the model’s inference method for each prompt.

generation_config: dict[str, Any]

The generation parameter for the model defaults generation_config.

remove_generation_config: tuple[str, ...] = ()

Attributes to set to None on the generation config.

resource: GeneratorResource

The class that creates resources such as the tokenizer and model.

stream(prompt, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, width=80)

Stream the model’s output from a prompt input.

Parameters:

• prompt (str) – the input to give to the model

• writer (TextIOBase) – the data sink

• width (int) – the maximum width of each line’s streamed text; if None, no modification will be done on the text output

stream_args: dict[str, Any]

The arguments given to the streamer in stream().

tokenize_decode_params: dict[str, Any]

Parameters to add or override in the model tokenize call.

tokenize_params: dict[str, Any]

Parameters to add or override in the model tokenize call.

class zensols.lmtask.generate.ReplaceTextGenerator(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>, replacements=())

Bases: ModelTextGenerator

A text generator that generates response by replacing regular expressions. This is helpful for removing special tokens.

__init__(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>, replacements=())

replacements: tuple[tuple[str | Pattern, str, str | None], ...] = ()

The a tuple (<regular expression>, <replacement>[, flags]) to replace in the parsed output from the model. String patters are compiled with re.compile().

The third element is a comma-separate list of regular expression re flags, such as DOTALL gets passed as re.subs(..., flags=re.DOTALL).

class zensols.lmtask.generate.TextGenerator

Bases: Dictable

A client facade of a chat-based large language model.

__init__()

clear()

Clear any model state.

generate(prompt)

Generate a textual response (usually from a large langauge model).

Return type:

GeneratorOutput

zensols.lmtask.hf module

HuggingFace trainer wrapper.

class zensols.lmtask.hf.HFTrainerResource(model_args=None, cache=True, generator_resource=None, peft_config=None)

Bases: TrainerResource

Uses HuggingFaceTrainer for training the model.

__init__(model_args=None, cache=True, generator_resource=None, peft_config=None)

generator_resource: GeneratorResource = None

The resource used to the source checkpoint.

peft_config: LoraConfig = None

The PEFT low-rank adapter configuration.

class zensols.lmtask.hf.HuggingFaceTrainer(config, resource, train_params, eval_params, train_source, eval_source, peft_output_dir, result_file)

Bases: Trainer

The HuggingFace trainer.

__init__(config, resource, train_params, eval_params, train_source, eval_source, peft_output_dir, result_file)

zensols.lmtask.instruct module

Task implementations.

class zensols.lmtask.instruct.InstructModelTextGenerator(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>, replacements=())

Bases: ReplaceTextGenerator

A generator that uses instruct based models for inference.

__init__(resource, tokenize_params=<factory>, tokenize_decode_params=<factory>, generate_params=<factory>, generation_config=<factory>, remove_generation_config=(), stream_args=<factory>, chat_template_args=<factory>, replacements=())

class zensols.lmtask.instruct.InstructTask(name, description, request_class, response_class, generator, resource, train_add_eos=False, role='You are a helpful assistant.', train_template='### Question: {{ instruction }}\\n### Answer: {{ output }}', inference_template='{{request.instruction}}', chat_template_args=<factory>, apply_chat_template=True, train_apply_chat_template=False)

Bases: GenerateTask

A task that is resolved using instructions given to the language model.

Important: If InstructTaskRequest.model_input is non-None that value is used verbatim and InstructTaskRequest.instruction is ignored.

__init__(name, description, request_class, response_class, generator, resource, train_add_eos=False, role='You are a helpful assistant.', train_template='### Question: {{ instruction }}\\n### Answer: {{ output }}', inference_template='{{request.instruction}}', chat_template_args=<factory>, apply_chat_template=True, train_apply_chat_template=False)

apply_chat_template: bool = True

Whether format the prompt into one that conforms to the model’s instruct syntax.

chat_template_args: dict[str, Any]

Arguments given to apply_chat_template.

inference_template: str | Path = '{{request.instruction}}'

The instructions given to generator.

role: str = 'You are a helpful assistant.'

The role of the chat dialogue.

train_apply_chat_template: bool = False

Whether to add apply_chat_template parameters to the apply_chat_template call during training. If this is False, a conversational messages with dictionary list is used instead.

train_template: str | Path = '### Question: {{ instruction }}\n### Answer: {{ output }}'

Used to create format the datasets training text generator.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.instruct.InstructTaskRequest(model_input=None, instruction=None)

Bases: TaskRequest

A request that has a query portion to be added to the compiled prompt.

__init__(model_input=None, instruction=None)

instruction: Any = None

The instruction given to the model to complete the task.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, include_instruction=True)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.instruct.NShotTaskRequest(model_input=None, instruction=None, examples=None)

Bases: InstructTaskRequest

A request that adds training examples to the prompt.

__init__(model_input=None, instruction=None, examples=None)

examples: tuple[Any, ...] = None

The examples given for N-shot learning.

zensols.lmtask.metric module

Task-level benchmark metrics.

Metric semantics are intentionally separated from benchmark execution so each task configuration can select the appropriate scorer (classification, regression, multilabel, generation, etc.).

class zensols.lmtask.metric.ClassMetric(label, precision, recall, f1, support)

Bases: Dictable

Metrics computed for a single class.

__init__(label, precision, recall, f1, support)

f1: float

F1 score for the class.

label: str

The class label represented by this result.

precision: float

Precision for the class.

recall: float

Recall for the class.

support: int

The number of gold examples belonging to the class.

class zensols.lmtask.metric.ClassificationMetricsCalculator(label_column='label', prediction_column='prediction', averages=('micro', 'macro', 'weighted'), primary_average='macro', primary_metric='f1', labels=None)

Bases: MetricsCalculator

Single-label classification metrics.

__init__(label_column='label', prediction_column='prediction', averages=('micro', 'macro', 'weighted'), primary_average='macro', primary_metric='f1', labels=None)

averages: tuple[str, ...] = ('micro', 'macro', 'weighted')

The averaging strategies used for aggregate precision, recall and F1 metrics.

calculate(df)

Return task-specific benchmark metrics.

Parameters:

df (DataFrame) – tester output containing gold labels and model predictions

Return type:

MetricsResult

Returns:

the computed task metrics

labels: tuple[str, ...] | None = None

The ordered valid class labels; None infers labels from the gold/reference column.

primary_average: str = 'macro'

The averaging strategy used for the primary metric.

primary_metric: str = 'f1'

The metric name used as the benchmark primary result.

class zensols.lmtask.metric.Metric(name, value, average=None)

Bases: Dictable

A scalar benchmark metric.

__init__(name, value, average=None)

property abbrev: str

The average (if there is one) appended to the abbrevation.

average: str | None = None

The averaging strategy, such as micro, macro or weighted; None when averaging does not apply.

property descriptor: str

The average (if there is one) appended to the name.

name: str

The metric name, such as accuracy, precision, recall or f1.

value: float

The computed scalar metric value.

class zensols.lmtask.metric.MetricsCalculator(label_column='label', prediction_column='prediction')

Bases: Dictable

Calculate task-level metrics from tester prediction rows.

__init__(label_column='label', prediction_column='prediction')

abstract calculate(df)

Return task-specific benchmark metrics.

Parameters:

df (DataFrame) – tester output containing gold labels and model predictions

Return type:

MetricsResult

Returns:

the computed task metrics

label_column: str = 'label'

The dataframe column containing gold/reference values.

prediction_column: str = 'prediction'

The dataframe column containing model predictions.

class zensols.lmtask.metric.MetricsResult(primary, support, metrics, per_class=(), invalid_count=0)

Bases: Dictable

Task-specific metrics returned by a MetricsCalculator.

__init__(primary, support, metrics, per_class=(), invalid_count=0)

property aggregate_dataframe: DataFrame

The aggregate metric results as a dataframe

property aggregate_row: DataFrameDescriber

The aggregate metric and counts in row-form as a data describer.

property count: int

The sum of the per-class suppport labels.

invalid_count: int = 0

The number of predictions that could not be interpreted as valid task outputs.

metrics: tuple[Metric, ...]

The aggregate scalar metrics computed for the task.

per_class: tuple[ClassMetric, ...] = ()

Optional per-class metrics; empty when per-class reporting does not apply.

property per_class_dataframe: DataFrameDescriber

The per class metrics as a dataframe.

primary: Metric

The metric considered the primary benchmark result for the task.

support: int

The total number of examples used to compute metrics.

zensols.lmtask.proto module

Prototyping.

class zensols.lmtask.proto.PrototypeApplication(config_factory, app, prompt='Once upon a time, in a galaxy, far far away,')

Bases: object

Used by the Python REPL for prototyping.

CLI_META = {'is_usage_visible': False}

__init__(config_factory, app, prompt='Once upon a time, in a galaxy, far far away,')

app: Application

config_factory: ConfigFactory

prompt: str = 'Once upon a time, in a galaxy, far far away,'

proto(run=0)

zensols.lmtask.task module

Task implementations.

class zensols.lmtask.task.JSONTaskResponse(request, model_output_raw, model_output, robust_json=True)

Bases: TaskResponse

A task that parses the responses as JSON. The JSON is parsed as much as possible and does not raise errors when the json is incomplete.

__init__(request, model_output_raw, model_output, robust_json=True)

property any_failures: bool

Whether any failures were created during JSON parsing.

property model_output_json: Failure | str

The response attribute parsed as JSON.

Raises:

json.decoder.JSONDecodeError – if the JSON failed to parse

See:

obj:robust_json

robust_json: bool = True

Whether to return Failure from model_output_json instead of raising from parse failures.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, include_request=False, include_model_output=False, include_json=True)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.task.Task(name, description, request_class, response_class)

Bases: Dictable

Subclasses turn a prompt and query into a response from an LLM.

__init__(name, description, request_class, response_class)

clear()

Clear any generator state or cache.

description: str

A description of the task.

name: str

The name of the task.

prepare_dataset(ds, factory)

Massage the any data for training necessary to train this task. This might involve apply templates and/or adding terminating tokens.

Return type:

Dataset

prepare_request(request)

Return a request with the contents populated with a formatted prompt.

Return type:

TaskRequest

process(request)

Invoke the generator to query the LLM, then return a JSON formatted data.

Parameters:

query – a query that is phrased with the assumption that JSON is given as a response

Return type:

TaskResponse

request_class: type[TaskRequest]

The response data.

response_class: type[TaskResponse]

The response data.

class zensols.lmtask.task.TaskDatasetFactory(task, text_field='text', messages_field='messages')

Bases: Dictable

Subclasses create a dataframes used by Trainer and optionally does post processing (i.e. filtering and mapping).

__init__(task, text_field='text', messages_field='messages')

create()

Create a new dataset based on source.

Return type:

Dataset

Returns:

the new dataset after modification by post_process

messages_field: str = 'messages'

The target conversational field used by the trainer or None to not add it.

task: Task

The task that helps format text in datasets.

text_field: str = 'text'

The target text field used by the trainer.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

exception zensols.lmtask.task.TaskDatasetFactoryError(message, prompt=None)

Bases: TaskError

Raised when TaskDatasetFactory instances can not create datasets.

__firstlineno__ = 161

__module__ = 'zensols.lmtask.task'

__static_attributes__ = ()

exception zensols.lmtask.task.TaskError(message, prompt=None)

Bases: APIError

Raised for any LLM specific error in this API.

__annotations__ = {}

__firstlineno__ = 24

__init__(message, prompt=None)

__module__ = 'zensols.lmtask.task'

__static_attributes__ = ('prompt',)

class zensols.lmtask.task.TaskFactory(config_factory, _task_pattern)

Bases: Dictable

Creates instances of Task using create().

__init__(config_factory, _task_pattern)

config_factory: ConfigFactory

The factory that creates tasks.

create(name)

Create a new instance of a task with name per the app config.

See:

task_names()

Return type:

Task

property task_names: set[str]

The names of the tasks available to create with create().

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, short=False)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.task.TaskObject

Bases: PersistableContainer, Dictable

Base class for task requests and responses.

__init__()

class zensols.lmtask.task.TaskRequest(model_input=None)

Bases: TaskObject

The input request to the LLM via Task.process(). In most cases, obj:model_input can be used to skip the prompt compilation step.

__init__(model_input=None)

model_input: str = None

The text given verbatim to the model. This is some combination of both querty and prompt.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.task.TaskResponse(request, model_output_raw, model_output)

Bases: TaskObject

The happy-path response given by Task.

__init__(request, model_output_raw, model_output)

model_output: str

This task instance’s parsed response text given by the model.

model_output_raw: str

The model output verbatim.

request: TaskRequest

The request used to generated this response.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, include_request=False, include_model_output=True, include_model_output_raw=False)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

zensols.lmtask.test module

Classes to test a model on datasets.

exception zensols.lmtask.test.TestError

Bases: APIError

__annotations__ = {}

__firstlineno__ = 26

__module__ = 'zensols.lmtask.test'

__static_attributes__ = ()

class zensols.lmtask.test.TestResult(prediction_col, raw_col, time_elapsed, predictions)

Bases: PersistableContainer, Dictable

Results from the a test run.

__init__(prediction_col, raw_col, time_elapsed, predictions)

property dataframe: DataFrame

The dataframe representation of predictions.

prediction_col: str

Add a prediction column to add.

predictions: tuple[dict[str, Any]]

The predictions as dict results, each as a row.

raw_col: bool

The column of the raw model output to add or None to not add it.

time_elapsed: int

Time in seconds it took to test.

write_jsonl(writer)

Write predictions to a JSONL file or data sink.

class zensols.lmtask.test.Tester(source, task, result_file, prediction_col='prediction', raw_col=None, result_mapper=<function Tester.<lambda>>, limit=None)

Bases: Dictable

Tests the fit of the model on a dataset.

__init__(source, task, result_file, prediction_col='prediction', raw_col=None, result_mapper=<function Tester.<lambda>>, limit=None)

limit: int | None = None

The limit on the number of test cases to process.

load_result()

Load the model results.

Return type:

TestResult

prediction_col: str = 'prediction'

Add a prediction column to add.

raw_col: bool = None

The column of the raw model output to add or None to not add it.

property result_exists: bool

Whether the trained model already exists.

result_file: Path

The file to save the training statistics for benchmarking.

result_mapper()

Map the result by calling with the single task.TaskResponse.

save_result(result)

Save the result to the file system.

source: TaskDatasetFactory

A factory that creates new datasets used to evaluation.

task: Task

The task used for to test the model.

test()

Run the tests and return the results with the predictions. The prediction is added as with column (key) prediction_col.

Return type:

tuple[dict[str, Any]]

zensols.lmtask.torchconfig module

CUDA access and utility module.

Copied from zensols.deeplearn, which is a heavy dependency package. Remove this module if that dependency is ever added.

class zensols.lmtask.torchconfig.CudaInfo

Bases: Writable

A utility class that provides information about the CUDA configuration for the current (hardware) environment.

get_devices(format=False)

Return type:

dict[int, dict[str, Any]]

property gpu_available: bool

property num_devices: int

Return number of devices connected.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Class representation as number of devices connected and about them.

See:

cuda

class zensols.lmtask.torchconfig.TorchConfig(use_gpu=True, data_type=torch.float32, cuda_device_index=None, device_name=None)

Bases: PersistableContainer, Writable

A utility class that provides access to CUDA APIs. It provides information on the current CUDA configuration and convenience methods to create, copy and modify tensors. These are handy for any given CUDA configuration and can back off to the CPU when CUDA isn’t available.

__init__(use_gpu=True, data_type=torch.float32, cuda_device_index=None, device_name=None)

Initialize this configuration.

Parameters:

• use_gpu (bool) – whether or not to use CUDA/GPU

• data_type (type) – the default data type to use when creating new tensors in this configuration

• cuda_device_index (int) – the CUDA device to use, which defaults to 0 if CUDA if use_gpu is True

• device_name (str) – the string name of the device to use (i.e. cpu or mps); if provided, overrides cuda_device_index

cat(*args, **kwargs)

Concatenate tensors in to one tensor using torch.cat.

Return type:

Tensor

clone(tensor, requires_grad=True)

Clone a tensor.

Return type:

Tensor

static close(a, b)

Return whether or not two tensors are equal. This does an exact cell comparison.

Return type:

bool

property cpu_device: torch.device

Return the CPU CUDA device, which is the device type configured to utilize the CPU (rather than the GPU).

classmethod cpu_device_name()

The string name of the torch CPU device.

Return type:

str

cross_entropy_pad(size)

Create a padded tensor of size size using the repeated pad ignore_index.

Return type:

Tensor

property cuda_configs: tuple[TorchConfig, ...]

Return a new set of configurations, one for each CUDA device.

property cuda_device_index: int | None

Return the CUDA device index if CUDA is being used for this configuration. Otherwise return None.

property cuda_devices: tuple[device, ...]

Return all cuda devices.

property device: device

Return the torch device configured.

empty(*args, **kwargs)

Return a new tesor using torch.empty.

Return type:

Tensor

static empty_cache()

Empty the CUDA torch cache. This releases memory in the GPU and should not be necessary to call for normal use cases.

static equal(a, b)

Return whether or not two tensors are equal. This does an exact cell comparison.

Return type:

bool

float(*args, **kwargs)

Return a new tensor using torch.tensor as a float type.

Return type:

Tensor

property float_type: type

Return the float type that represents this configuration, converting to the corresponding precision from integer if necessary.

Returns:

the float that represents this data, or None if neither float nor int

from_iterable(array)

Return a one dimenstional tensor created from array using the type and device in the current instance configuration.

Return type:

Tensor

from_numpy(arr)

Return a new tensor generated from a numpy aray using torch.from_numpy. The array type is converted if necessary.

Return type:

Tensor

get_peak_memory(format=False)

Return peak allocated and reserved CUDA memory for visible devices.

Return type:

dict[int, dict[str, int | str]]

classmethod get_random_seed()

Get the cross system random seed, meaning the seed applied to CUDA and the Python random library.

Return type:

int

classmethod get_random_seed_context()

Return the random seed context given to set_random_seed() to restore across models for consistent results.

Return type:

dict[str, Any]

property gpu_available: bool

Return whether or not CUDA GPU access is available.

static in_memory_tensors()

Returns all in-memory tensors and parameters.

See:

show_leaks()

Return type:

list[Tensor]

property info: CudaInfo

Return the CUDA information, which include specs of the device.

classmethod init(spawn_multiproc='spawn', seed_kwargs={})

Initialize the PyTorch framework. This includes:

• Configuration of PyTorch multiprocessing so subprocesses can access the GPU, and

• Setting the random seed state.

The needs to be initialized at the very beginning of your program if you are training a new model. Note: this should be called when testing a model, but not when inferencing a production model.

Example:

def main():
    from zensols.deeplearn import TorchConfig
    TorchConfig.init()

Note: this method is separate from set_random_seed() because that method is called by the framework to reset the seed after a model is unpickled.

See:

torch.multiprocessing

See:

set_random_seed()

int(*args, **kwargs)

Return a new tensor using torch.tensor as a int type.

Return type:

Tensor

property int_type: type

Return the int type that represents this configuration, converting to the corresponding precision from integer if necessary.

Returns:

the int that represents this data, or None if neither int nor float

classmethod is_on_cpu(arr)

Return True if the passed tensor is on the CPU.

Return type:

bool

is_sparse(arr)

Return whether or not a tensor a sparse.

Return type:

bool

property numpy_data_type: type[dtype]

Return the numpy type that corresponds to this instance’s configured data_type.

ones(*args, **kwargs)

Return a new tensor of zeros using torch.ones.

Return type:

Tensor

reset_peak_memory_stats()

Reset peak CUDA memory statistics on all visible devices.

same_device(tensor_or_model)

Return whether or not a tensor or model is in the same memory space as this configuration instance.

Return type:

bool

classmethod set_random_seed(seed=0, disable_cudnn=True, rng_state=True)

Set the random number generator for PyTorch.

Parameters:

• seed (int) – the random seed to be set

• disable_cudnn (bool) – if True disable NVidia’s backend cuDNN hardware acceleration, which might have non-deterministic features

• rng_state (bool) – set the CUDA random state array to zeros

See:

Torch Random Seed

See:

Reproducibility

singleton(*args, **kwargs)

Return a new tensor using torch.tensor.

Return type:

Tensor

sparse(indicies, values, shape)

Create a sparce tensor from indexes and values.

property tensor_class: type[dtype]

Return the class type based on the current configuration of this instance. For example, if using torch.float32 on the GPU, torch.cuda.FloatTensor is returned.

to(tensor_or_model)

Copy the tensor or model to the device this to that of this configuration.

Return type:

Module | Tensor

classmethod to_cpu_deallocate(*arrs)

Safely copy detached memory to the CPU and delete local instance (possibly GPU) memory to speed up resource deallocation. If the tensor is already on the CPU, it’s simply passed back. Otherwise the tensor is deleted.

This method is robust with None, which are skipped and substituted as None in the output.

Parameters:

arrs (tuple[Tensor, ...]) – the tensors the copy to the CPU (if not already)

Return type:

tuple[Tensor, ...] | Tensor

Returns:

the singleton tensor if only one arrs is passed; otherwise, the CPU copied tensors from the input

to_type(arr)

Convert the type of the given array to the type of this instance.

Return type:

Tensor

property using_cpu: bool

Return True if this configuration is using the CPU device.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write the contents of this instance to writer using indention depth.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

write_device_tensors(writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Like write_in_memory_tensors(), but filter on this instance’s device.

Parameters:

filter_device – if given, write only tensors matching this device

See:

TorchConfig

classmethod write_in_memory_tensors(writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, filter_device=None)

Prints in-memory tensors and parameters.

Parameters:

filter_device (device) – if given, write only tensors matching this device

See:

TorchConfig

zeros(*args, **kwargs)

Return a new tensor of zeros using torch.zeros.

Return type:

Tensor

class zensols.lmtask.torchconfig.printopts(**kwargs)

Bases: object

Object used with a with scope that sets options, then sets them back.

Example:

with printopts(profile='full', linewidth=120):
    print(tensor)

See:

PyTorch Documentation

DEFAULTS = {'edgeitems': 3, 'linewidth': 80, 'precision': 4, 'profile': 'default', 'sci_mode': None, 'threshold': 1000}

__init__(**kwargs)

zensols.lmtask.torchtype module

CUDA access and utility module.

class zensols.lmtask.torchtype.TorchTypes

Bases: object

A utility class to convert betwen numpy and torch classes. It also provides metadata for types that make other conversions, such as same precision cross types (i.e. int64 -> float64).

FLOAT_TO_INT = {torch.float16: torch.int16, torch.float32: torch.int32, torch.float64: torch.int64}

FLOAT_TYPES = frozenset({torch.float16, torch.float32, torch.float64})

INT_TO_FLOAT = {torch.int16: torch.float16, torch.int32: torch.float32, torch.int64: torch.float64}

INT_TYPES = frozenset({torch.int16, torch.int32, torch.int64})

NAME_TO_TYPE = {'bool': {'cpu': <class 'torch.BoolTensor'>, 'desc': 'Boolean', 'gpu': <class 'torch.cuda.BoolTensor'>, 'name': 'bool', 'numpy': <class 'bool'>, 'types': {torch.bool}}, 'float16': {'cpu': <class 'torch.HalfTensor'>, 'desc': '16-bit floating point', 'gpu': <class 'torch.cuda.HalfTensor'>, 'name': 'float16', 'numpy': <class 'numpy.float16'>, 'sparse': <class 'torch.sparse.HalfTensor'>, 'types': {torch.float16}}, 'float32': {'cpu': <class 'torch.FloatTensor'>, 'desc': '32-bit floating point', 'gpu': <class 'torch.cuda.FloatTensor'>, 'name': 'float32', 'numpy': <class 'numpy.float32'>, 'sparse': <class 'torch.sparse.FloatTensor'>, 'types': {torch.float32}}, 'float64': {'cpu': <class 'torch.DoubleTensor'>, 'desc': '64-bit floating point', 'gpu': <class 'torch.cuda.DoubleTensor'>, 'name': 'float64', 'numpy': <class 'numpy.float64'>, 'sparse': <class 'torch.sparse.DoubleTensor'>, 'types': {torch.float64}}, 'int16': {'cpu': <class 'torch.ShortTensor'>, 'desc': '16-bit integer (signed)', 'gpu': <class 'torch.cuda.ShortTensor'>, 'name': 'int16', 'numpy': <class 'numpy.int16'>, 'sparse': <class 'torch.sparse.ShortTensor'>, 'types': {torch.int16}}, 'int32': {'cpu': <class 'torch.IntTensor'>, 'desc': '32-bit integer (signed)', 'gpu': <class 'torch.cuda.IntTensor'>, 'name': 'int32', 'numpy': <class 'numpy.int32'>, 'sparse': <class 'torch.sparse.IntTensor'>, 'types': {torch.int32}}, 'int64': {'cpu': <class 'torch.LongTensor'>, 'desc': '64-bit integer (signed)', 'gpu': <class 'torch.cuda.LongTensor'>, 'name': 'int64', 'numpy': <class 'numpy.int64'>, 'sparse': <class 'torch.sparse.LongTensor'>, 'types': {torch.int64}}, 'int8': {'cpu': <class 'torch.CharTensor'>, 'desc': '8-bit integer (signed)', 'gpu': <class 'torch.cuda.CharTensor'>, 'name': 'int8', 'numpy': <class 'numpy.int8'>, 'sparse': <class 'torch.sparse.CharTensor'>, 'types': {torch.int8}}, 'uint8': {'cpu': <class 'torch.ByteTensor'>, 'desc': '8-bit integer (unsigned)', 'gpu': <class 'torch.cuda.ByteTensor'>, 'name': 'uint8', 'numpy': <class 'numpy.uint8'>, 'sparse': <class 'torch.sparse.ByteTensor'>, 'types': {torch.uint8}}}

A map of type to metadata.

TYPES = [{'cpu': <class 'torch.FloatTensor'>, 'desc': '32-bit floating point', 'gpu': <class 'torch.cuda.FloatTensor'>, 'name': 'float32', 'numpy': <class 'numpy.float32'>, 'sparse': <class 'torch.sparse.FloatTensor'>, 'types': {torch.float32}}, {'cpu': <class 'torch.DoubleTensor'>, 'desc': '64-bit floating point', 'gpu': <class 'torch.cuda.DoubleTensor'>, 'name': 'float64', 'numpy': <class 'numpy.float64'>, 'sparse': <class 'torch.sparse.DoubleTensor'>, 'types': {torch.float64}}, {'cpu': <class 'torch.HalfTensor'>, 'desc': '16-bit floating point', 'gpu': <class 'torch.cuda.HalfTensor'>, 'name': 'float16', 'numpy': <class 'numpy.float16'>, 'sparse': <class 'torch.sparse.HalfTensor'>, 'types': {torch.float16}}, {'cpu': <class 'torch.ByteTensor'>, 'desc': '8-bit integer (unsigned)', 'gpu': <class 'torch.cuda.ByteTensor'>, 'name': 'uint8', 'numpy': <class 'numpy.uint8'>, 'sparse': <class 'torch.sparse.ByteTensor'>, 'types': {torch.uint8}}, {'cpu': <class 'torch.CharTensor'>, 'desc': '8-bit integer (signed)', 'gpu': <class 'torch.cuda.CharTensor'>, 'name': 'int8', 'numpy': <class 'numpy.int8'>, 'sparse': <class 'torch.sparse.CharTensor'>, 'types': {torch.int8}}, {'cpu': <class 'torch.ShortTensor'>, 'desc': '16-bit integer (signed)', 'gpu': <class 'torch.cuda.ShortTensor'>, 'name': 'int16', 'numpy': <class 'numpy.int16'>, 'sparse': <class 'torch.sparse.ShortTensor'>, 'types': {torch.int16}}, {'cpu': <class 'torch.IntTensor'>, 'desc': '32-bit integer (signed)', 'gpu': <class 'torch.cuda.IntTensor'>, 'name': 'int32', 'numpy': <class 'numpy.int32'>, 'sparse': <class 'torch.sparse.IntTensor'>, 'types': {torch.int32}}, {'cpu': <class 'torch.LongTensor'>, 'desc': '64-bit integer (signed)', 'gpu': <class 'torch.cuda.LongTensor'>, 'name': 'int64', 'numpy': <class 'numpy.int64'>, 'sparse': <class 'torch.sparse.LongTensor'>, 'types': {torch.int64}}, {'cpu': <class 'torch.BoolTensor'>, 'desc': 'Boolean', 'gpu': <class 'torch.cuda.BoolTensor'>, 'name': 'bool', 'numpy': <class 'bool'>, 'types': {torch.bool}}]

A list of dicts containig conversions between types.

classmethod all_types()

Return type:

List[dict]

classmethod float_to_int(torch_type)

Return type:

Type

classmethod get_numpy_type(torch_type)

Return type:

Type

classmethod get_sparse_class(torch_type)

Return type:

Type

classmethod get_tensor_class(torch_type, cpu_type)

Return type:

Type

classmethod int_to_float(torch_type)

Return type:

Type

classmethod is_float(torch_type)

Return type:

bool

classmethod is_int(torch_type)

Return type:

bool

classmethod type_from_string(type_name)

Return type:

dtype

classmethod types()

Return type:

Dict[str, List[dict]]

zensols.lmtask.train module

Continued Pretraining and supervised fine-tuning training.

exception zensols.lmtask.train.TrainError

Bases: APIError

__annotations__ = {}

__firstlineno__ = 29

__module__ = 'zensols.lmtask.train'

__static_attributes__ = ()

class zensols.lmtask.train.TrainResult(train_output, peft_output_dir, train_params, config, time_elapsed)

Bases: Dictable

The trained model config, location and configuration used to train it.

__init__(train_output, peft_output_dir, train_params, config, time_elapsed)

config: Configurable

The application configuration used to configure the trainer.

property global_step: int

The global step from train_output.

property metrics: dict[str, float]

Training metrics from train_output.

peft_output_dir: Path

The directory of the models checkpoints.

time_elapsed: int

Time in seconds it took to train.

train_output: TrainOutput

The output returned from the trainer.

train_params: dict[str, Any]

The training parameters used to configure the trainer.

property training_loss: float

The training loss from train_output.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, include_training_arguments=False, include_config=False)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.train.Trainer(config, resource, train_params, eval_params, train_source, eval_source, peft_output_dir, result_file)

Bases: Dictable

A configurable supervised fine-tuning trainer wrapper.

__init__(config, resource, train_params, eval_params, train_source, eval_source, peft_output_dir, result_file)

config: Configurable

Used to save to the model result.

eval_params: dict[str, Any]

The evaluation parameters used to configure the trainer.

eval_source: TaskDatasetFactory

A factory that creates new datasets used to evaluation.

load_result()

Load the model results.

Return type:

TrainResult

property model_exists: bool

Whether the trained model already exists.

peft_output_dir: Path

The directory in which to save the PEFT adapter.

resource: TrainerResource

Used to create the model and tokenizer.

result_file: Path

The file to save the training statistics for benchmarking.

save_result(result)

Save the result to the file system.

train()

Train the model.

Return type:

TrainResult

train_params: dict[str, Any]

The training parameters used to configure the trainer.

train_source: TaskDatasetFactory

A factory that creates new datasets used to train using this instance.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, include_training_arguments=False)

Write this instance as either a Writable or as a Dictable. If class attribute _DICTABLE_WRITABLE_DESCENDANTS is set as True, then use the write() method on children instead of writing the generated dictionary. Otherwise, write this instance by first creating a dict recursively using asdict(), then formatting the output.

If the attribute _DICTABLE_WRITE_EXCLUDES is set, those attributes are removed from what is written in the write() method.

Note that this attribute will need to be set in all descendants in the instance hierarchy since writing the object instance graph is done recursively.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

class zensols.lmtask.train.TrainerResource(model_args=None, cache=True)

Bases: Dictable, Primeable

Configures and instantiates the base mode, PEFT mode, and the tokenizer.

__init__(model_args=None, cache=True)

cache: bool = True

Whether to cache the tokenizer and model.

property model: PreTrainedModel

The base model.

model_args: dict[str, Any] = None

The parameters that create the base model and tokenzier.

property peft_model: PeftModelForCausalLM

The PEFT (Parameter-Efficient Fine-Tuning) such as LoRA.

prime()

property tokenizer: PythonBackend

The base tokenizer.

Module contents