vh/heretic
1
0
Fork 0
mirror of https://github.com/p-e-w/heretic.git synced 2026-10-02 08:51:27 -07:00
Code Issues Packages Projects Releases Wiki Activity

feat: headless operation + end-to-end tests (#392)

Browse Source 
* fix: remove notebook input shims

Closes #280

* feat: support headless operation (no interactive input)

* fix: prevent infinite loops

* feat: add end-to-end tests

* ci: run tests in CI

* ci: fix test output ordering

* fix: replace home-cooked `set_seed` function with Transformers builtin

* feat: print PyTorch config when running tests

* feat: print additional information

* experiment: try to standardize test environment

* fix: revert environment changes

* feat: support multiple valid hashes for each output file

* feat: add test output hashes for CI

* feat: add test output hashes for CI (alternative environment)

* feat: add hashes for Windows (#394)

* fix: Hash on windows

* trigger ci

* fix: prefer .yaml (used widely than .toml for model configs)

* use removeprefix

* docs: restore commet

* use removeprefix again

* tests: Add windows hash files for all test models

* trigger ci

* fix: minor cleanup

* clean merge mismatch

* remove unnecessary CRLF replace, now that we support more SUMS files

* fix: use binary mode for hashes everywhere

---------

Co-authored-by: Vinay Umrethe <umrethevinay@gmail.com>
This commit is contained in:
Philipp Emanuel Weidmann
2026-06-27 13:41:48 +05:30
committed by GitHub
co-authored by Vinay Umrethe
parent 3f68a0d4e5
commit 0146b2760f
27 changed files with 715 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/heretic/config.py
+68 -10
View File
@@ -4,7 +4,12 @@
from enum import Enum
from typing import Dict
from pydantic import BaseModel, Field
from pydantic import (
BaseModel,
Field,
NonNegativeInt,
PositiveInt,
)
from pydantic_settings import (
BaseSettings,
CliSettingsSource,
@@ -181,12 +186,12 @@ class Settings(BaseSettings):
),
)
batch_size: int = Field(
batch_size: NonNegativeInt = Field(
default=0, # auto
description="Number of input sequences to process in parallel (0 = auto).",
)
max_batch_size: int = Field(
max_batch_size: PositiveInt = Field(
default=128,
description="Maximum batch size to try when automatically determining the optimal batch size.",
# When storing a settings object, the batch size is already fixed,
@@ -194,7 +199,7 @@ class Settings(BaseSettings):
exclude=True,
)
max_response_length: int = Field(
max_response_length: PositiveInt = Field(
default=100,
description="Maximum number of tokens to generate for each response.",
)
@@ -247,6 +252,12 @@ class Settings(BaseSettings):
exclude=True,
)
print_debug_information: bool = Field(
default=False,
description="Whether to print additional information that can help with debugging.",
exclude=True,
)
print_residual_geometry: bool = Field(
default=False,
description="Whether to print detailed information about residuals and refusal directions.",
@@ -311,7 +322,7 @@ class Settings(BaseSettings):
),
)
full_normalization_lora_rank: int = Field(
full_normalization_lora_rank: PositiveInt = Field(
default=3,
description=(
'The rank of the LoRA adapter to use when "full" row normalization is used. '
@@ -332,12 +343,12 @@ class Settings(BaseSettings):
),
)
n_trials: int = Field(
n_trials: PositiveInt = Field(
default=200,
description="Number of abliteration trials to run during optimization.",
)
n_startup_trials: int = Field(
n_startup_trials: NonNegativeInt = Field(
default=60,
description="Number of trials that use random sampling for the purpose of exploration.",
)
@@ -418,14 +429,61 @@ class Settings(BaseSettings):
exclude=True,
)
max_shard_size: PositiveInt | str = Field(
default="5GB",
description="Maximum size for individual safetensors files generated when exporting a model.",
)
export_strategy: ExportStrategy | None = Field(
default=None,
description='How to export the model: "merge", "adapter", or unset to prompt the user.',
)
max_shard_size: int | str = Field(
default="5GB",
description="Maximum size for individual safetensors files generated when exporting a model.",
checkpoint_action: str | None = Field(
default=None,
description='Action to take in case a checkpoint exists: "continue", "restart", or unset to prompt the user.',
)
trial_index: NonNegativeInt | None = Field(
default=None,
description="Index (in the sorted Pareto front) of the trial to use, or unset to prompt the user.",
)
n_additional_trials: PositiveInt | None = Field(
default=None,
description="Number of additional trials to run, or unset to prompt the user.",
)
model_action: str | None = Field(
default=None,
description='Action to take with the decensored model: "save", "upload", or unset to prompt the user.',
)
save_directory: str | None = Field(
default=None,
description="Directory to save the model to, or unset to prompt the user.",
exclude=True,
)
upload_repo_id: str | None = Field(
default=None,
description="Name of the Hugging Face repository to upload the model to, or unset to prompt the user.",
exclude=True,
)
upload_repo_private: bool | None = Field(
default=None,
description="Whether the Hugging Face repository to upload the model to should be private, or unset to prompt the user.",
)
upload_reproducibility_information: str | None = Field(
default=None,
description='Which reproducibility information to add to the Hugging Face repository: "full", "basic", "none", or unset to prompt the user.',
)
ignore_mismatches: bool | None = Field(
default=None,
description="Whether to attempt to reproduce the model even if there are environment mismatches, or unset to prompt the user.",
)
refusal_markers: list[str] = Field(
src/heretic/main.py
+233 -129
View File
@@ -80,6 +80,7 @@ from .reproduce import (
)
from .system import empty_cache, get_accelerator_info
from .utils import (
ask_if_unset,
format_duration,
format_exception,
get_file_sha256,
@@ -89,11 +90,6 @@ from .utils import (
load_prompts,
print,
print_memory_usage,
prompt_password,
prompt_path,
prompt_select,
prompt_text,
set_seed,
upload_reproduce_folder,
)
@@ -108,10 +104,10 @@ def obtain_export_strategy(
Returns an export strategy, or None if cancelled.
"""
if settings.export_strategy is not None:
return settings.export_strategy
if settings.quantization == QuantizationMethod.BNB_4BIT:
if (
settings.quantization == QuantizationMethod.BNB_4BIT
and settings.export_strategy is None
):
print()
print(
"The model was loaded with quantization. Merging requires reloading the base model."
@@ -155,27 +151,29 @@ def obtain_export_strategy(
print()
strategy = prompt_select(
"How do you want to export the model?",
choices=[
Choice(
title="Merge the abliteration LoRA and export the full model"
+ (
""
if settings.quantization == QuantizationMethod.NONE
else " (requires sufficient RAM)"
return ask_if_unset(
settings.export_strategy,
questionary.select(
"How do you want to export the model?",
choices=[
Choice(
title="Merge the abliteration LoRA and export the full model"
+ (
""
if settings.quantization == QuantizationMethod.NONE
else " (requires sufficient RAM)"
),
value=ExportStrategy.MERGE,
),
value=ExportStrategy.MERGE,
),
Choice(
title="Export the abliteration LoRA only (can be merged later)",
value=ExportStrategy.ADAPTER,
),
],
Choice(
title="Export the abliteration LoRA only (can be merged later)",
value=ExportStrategy.ADAPTER,
),
],
style=Style([("highlighted", "reverse")]),
),
)
return strategy
def run():
# Enable expandable segments to reduce memory fragmentation on multi-GPU setups.
@@ -254,7 +252,7 @@ def run():
)
return
if not check_environment(reproduction_information):
if not check_environment(settings, reproduction_information):
return
print()
@@ -266,10 +264,22 @@ def run():
if settings.seed is None:
settings.seed = random.randint(0, 2**32 - 1)
set_seed(settings.seed)
transformers.set_seed(settings.seed)
print(get_accelerator_info())
if settings.print_debug_information:
print()
print(torch.__config__.show().strip())
print()
print(
f"torch.backends.mkldnn.enabled = [bold]{torch.backends.mkldnn.enabled}[/]"
)
print(f"torch.get_num_threads() = [bold]{torch.get_num_threads()}[/]")
print(
f"torch.get_num_interop_threads() = [bold]{torch.get_num_interop_threads()}[/]"
)
# We don't need gradients as we only do inference.
torch.set_grad_enabled(False)
@@ -320,15 +330,17 @@ def run():
choices = []
if existing_study.user_attrs["finished"]:
print()
print(
(
"[green]You have already processed this model.[/] "
"You can show the results from the previous run, allowing you to export models or to run additional trials. "
"Alternatively, you can ignore the previous run and start from scratch. "
"This will delete the checkpoint file and all results from the previous run."
if settings.checkpoint_action is None:
print()
print(
(
"[green]You have already processed this model.[/] "
"You can show the results from the previous run, allowing you to export models or to run additional trials. "
"Alternatively, you can ignore the previous run and start from scratch. "
"This will delete the checkpoint file and all results from the previous run."
)
)
)
choices.append(
Choice(
title="Show the results from the previous run",
@@ -336,15 +348,17 @@ def run():
)
)
else:
print()
print(
(
"[yellow]You have already processed this model, but the run was interrupted.[/] "
"You can continue the previous run from where it stopped. This will override any specified settings. "
"Alternatively, you can ignore the previous run and start from scratch. "
"This will delete the checkpoint file and all results from the previous run."
if settings.checkpoint_action is None:
print()
print(
(
"[yellow]You have already processed this model, but the run was interrupted.[/] "
"You can continue the previous run from where it stopped. This will override any specified settings. "
"Alternatively, you can ignore the previous run and start from scratch. "
"This will delete the checkpoint file and all results from the previous run."
)
)
)
choices.append(
Choice(
title="Continue the previous run",
@@ -366,19 +380,29 @@ def run():
)
)
print()
choice = prompt_select("How would you like to proceed?", choices)
if settings.checkpoint_action is None:
print()
if choice == "continue":
action = ask_if_unset(
settings.checkpoint_action,
questionary.select(
"How would you like to proceed?",
choices=choices,
style=Style([("highlighted", "reverse")]),
),
)
if action is None or action == "":
return
if action == "continue":
settings = Settings.model_validate_json(
existing_study.user_attrs["settings"]
)
elif choice == "restart":
elif action == "restart":
os.unlink(study_checkpoint_file)
backend = JournalFileBackend(study_checkpoint_file, lock_obj=lock_obj)
storage = JournalStorage(backend)
elif choice is None or choice == "":
return
model = Model(settings)
print()
@@ -619,7 +643,7 @@ def run():
min_weight_distance = trial.suggest_float(
f"{component}.min_weight_distance",
1.0,
0.6 * last_layer_index,
max(0.6 * last_layer_index, 1.0),
)
parameters[component] = AbliterationParameters(
@@ -709,7 +733,9 @@ def run():
if len(study.trials) == settings.n_trials:
study.set_user_attr("finished", True)
while True:
trial_loop_active = True
while trial_loop_active:
if not reproduction_mode:
# If no trials at all have been evaluated, the study must have been stopped
# by pressing Ctrl+C while the first trial was running. In this case, we just
@@ -766,18 +792,24 @@ def run():
print()
print("[bold green]Optimization finished![/]")
print()
print(
(
"The following trials resulted in Pareto optimal combinations of refusals and KL divergence. "
"After selecting a trial, you will be able to save the model, upload it to Hugging Face, "
"chat with it to test how well it works, or run standard benchmarks on it. "
"You can return to this menu later to select a different trial. "
"[yellow]Note that KL divergence values above 0.5 usually indicate significant damage to the original model's capabilities.[/]"
)
)
while True:
if settings.trial_index is None:
print()
print(
(
"The following trials resulted in Pareto optimal combinations of refusals and KL divergence. "
"After selecting a trial, you will be able to save the model, upload it to Hugging Face, "
"chat with it to test how well it works, or run standard benchmarks on it. "
"You can return to this menu later to select a different trial. "
"[yellow]Note that KL divergence values above 0.5 usually indicate significant damage to the original model's capabilities.[/]"
)
)
while trial_loop_active:
# Ensure a predefined trial is only processed once.
if settings.trial_index is not None:
trial_loop_active = False
if reproduction_mode:
parameters = reproduction_information["parameters"]
metrics = reproduction_information["metrics"]
@@ -797,8 +829,19 @@ def run():
print()
print("Restoring model from reproduction information...")
else:
print()
trial = prompt_select("Which trial do you want to use?", choices)
if settings.trial_index is None:
print()
trial = ask_if_unset(
None
if settings.trial_index is None
else best_trials[settings.trial_index],
questionary.select(
"Which trial do you want to use?",
choices=choices,
style=Style([("highlighted", "reverse")]),
),
)
if trial is None or trial == "":
return
@@ -806,8 +849,11 @@ def run():
if trial == "continue":
while True:
try:
n_additional_trials = prompt_text(
"How many additional trials do you want to run?"
n_additional_trials = ask_if_unset(
settings.n_additional_trials,
questionary.text(
"How many additional trials do you want to run?"
),
)
if n_additional_trials is None or n_additional_trials == "":
n_additional_trials = 0
@@ -866,22 +912,46 @@ def run():
reset_trial_model()
while True:
print()
action = prompt_select(
"What do you want to do with the decensored model?",
[
"Save the model to a local folder",
"Upload the model to Hugging Face",
"Chat with the model",
"Benchmark the model",
Choice(
title="Exit program"
if reproduction_mode
else "Return to the trial selection menu",
value="",
),
],
action_loop_active = True
while action_loop_active:
# Ensure a predefined action is only executed once.
if settings.model_action is not None:
action_loop_active = False
if settings.model_action is None:
print()
action = ask_if_unset(
settings.model_action,
questionary.select(
"What do you want to do with the decensored model?",
choices=[
Choice(
title="Save the model to a local folder",
value="save",
),
Choice(
title="Upload the model to Hugging Face",
value="upload",
),
Choice(
title="Chat with the model",
value="chat",
),
Choice(
title="Benchmark the model",
value="benchmark",
),
Choice(
title="Exit program"
if reproduction_mode
else "Return to the trial selection menu",
value="",
),
],
style=Style([("highlighted", "reverse")]),
),
)
if action is None or action == "":
@@ -895,8 +965,14 @@ def run():
# the optimized model.
try:
match action:
case "Save the model to a local folder":
save_directory = prompt_path("Path to the folder:")
case "save":
save_directory = ask_if_unset(
settings.save_directory,
questionary.path(
"Path to the folder:",
only_directories=True,
),
)
if not save_directory:
continue
@@ -951,13 +1027,20 @@ def run():
f"[bold]{filename}:[/] [red]File not found[/]"
)
case "Upload the model to Hugging Face":
case "upload":
# We don't use huggingface_hub.login() because that stores the token on disk,
# and since this program will often be run on rented or shared GPU servers,
# it's better to not persist credentials.
token = huggingface_hub.get_token()
if not token:
token = prompt_password("Hugging Face access token:")
# NOTE: Unlike for most other values obtained from interactive inputs, it is
# not possible to set the token via the settings. This is a security
# precaution to prevent exporting the token under all circumstances.
# For scripting, the correct way to set the token is through the HF_TOKEN
# environment variable, or through the HF token file.
token = questionary.password(
"Hugging Face access token:"
).ask()
if not token:
continue
@@ -969,17 +1052,32 @@ def run():
email = user.get("email", "no email found")
print(f"Logged in as [bold]{fullname} ({email})[/]")
repo_id = prompt_text(
"Name of repository:",
default=f"{user['name']}/{Path(settings.model).name}-heretic",
repo_id = ask_if_unset(
settings.upload_repo_id,
questionary.text(
"Name of repository:",
default=f"{user['name']}/{Path(settings.model).name}-heretic",
),
)
if not repo_id:
continue
visibility = prompt_select(
"Should the repository be public or private?",
[
"Public",
"Private",
],
visibility = ask_if_unset(
None
if settings.upload_repo_private is None
else (
"Private"
if settings.upload_repo_private
else "Public"
),
questionary.select(
"Should the repository be public or private?",
choices=[
"Public",
"Private",
],
style=Style([("highlighted", "reverse")]),
),
)
if visibility is None:
continue
@@ -1004,31 +1102,37 @@ def run():
)
if is_reproducible:
print(
(
"Heretic can add information to the repository that allows others to reproduce the model. "
"This is optional, but valuable to the community as both a learning tool and to preserve computational work already done. "
"Guaranteeing reproducibility requires basic system information (Python and OS version, CPU and GPU/accelerator info) "
"as tensor operations can give different results in different system environments. "
"[bold]The information does not include any file system paths or other private data.[/]"
if settings.upload_reproducibility_information is None:
print(
(
"Heretic can add information to the repository that allows others to reproduce the model. "
"This is optional, but valuable to the community as both a learning tool and to preserve computational work already done. "
"Guaranteeing reproducibility requires basic system information (Python and OS version, CPU and GPU/accelerator info) "
"as tensor operations can give different results in different system environments. "
"[bold]The information does not include any file system paths or other private data.[/]"
)
)
)
reproducibility_information = prompt_select(
"Which reproducibility information do you want to add?",
[
Choice(
title="Full: Settings, package versions, and system information",
value="full",
),
Choice(
title="Basic: Settings and package versions",
value="basic",
),
Choice(
title="Don't add any reproducibility information",
value="none",
),
],
reproducibility_information = ask_if_unset(
settings.upload_reproducibility_information,
questionary.select(
"Which reproducibility information do you want to add?",
choices=[
Choice(
title="Full: Settings, package versions, and system information",
value="full",
),
Choice(
title="Basic: Settings and package versions",
value="basic",
),
Choice(
title="Don't add any reproducibility information",
value="none",
),
],
style=Style([("highlighted", "reverse")]),
),
)
if reproducibility_information is None:
continue
@@ -1174,7 +1278,7 @@ def run():
f"[bold]{filename}:[/] [red]File not found[/]"
)
case "Chat with the model":
case "chat":
print()
print(
"[cyan]Press Ctrl+C at any time to return to the menu.[/]"
@@ -1186,11 +1290,10 @@ def run():
while True:
try:
message = prompt_text(
message = questionary.text(
"User:",
qmark=">",
unsafe=True,
)
).unsafe_ask()
if not message:
break
chat.append({"role": "user", "content": message})
@@ -1204,7 +1307,7 @@ def run():
# Ctrl+C/Ctrl+D
break
case "Benchmark the model":
case "benchmark":
benchmarks = questionary.checkbox(
"Which benchmarks do you want to run?",
[
@@ -1219,16 +1322,17 @@ def run():
if not benchmarks:
continue
scope = prompt_select(
scope = questionary.select(
(
"Do you want to benchmark the original model along with the decensored model? "
"Benchmarking both models allows you to compare the scores, but it takes twice as much time."
),
[
choices=[
"Benchmark only the decensored model",
"Benchmark both models",
],
)
style=Style([("highlighted", "reverse")]),
).ask()
if scope is None:
continue
benchmark_original_model = scope == "Benchmark both models"
src/heretic/model.py
+5
View File
@@ -586,11 +586,16 @@ class Model:
W = W - W_org
# Use a low-rank SVD to get an approximation of the matrix.
r = self.peft_config.r
# svd_lowrank is randomized:
# https://github.com/pytorch/pytorch/blob/20919052303c0b5ba87f8bf7e19237dc33ab09d3/torch/_lowrank.py#L108-L109
# Reseed immediately before the call so restoring a trial is independent of RNG history.
torch.manual_seed(self.settings.seed)
# "It's safe to call this function if CUDA is not available;
# in that case, it is silently ignored."
torch.cuda.manual_seed_all(self.settings.seed) # ty:ignore[invalid-argument-type]
U, S, Vh = torch.svd_lowrank(W, q=2 * r + 4, niter=6)
# Truncate it to the part we want to store in the LoRA adapter.
# Note: svd_lowrank actually returns V, so transpose it to get Vh.
U = U[:, :r]
src/heretic/reproduce.py
+27 -18
View File
@@ -12,6 +12,7 @@ from typing import Any, cast
from urllib.request import urlopen
import cpuinfo
import questionary
import torch
from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.utils import (
@@ -19,15 +20,16 @@ from huggingface_hub.utils import (
disable_progress_bars,
enable_progress_bars,
)
from questionary import Choice
from questionary import Choice, Style
from rich.table import Table
from .config import Settings
from .system import (
get_accelerator_info_dict,
get_heretic_version_info,
get_requirements_dict,
)
from .utils import print, prompt_select
from .utils import ask_if_unset, print
def collect_reproducibles(path: str):
@@ -192,7 +194,10 @@ def format_version_information(version_information: dict[str, Any]) -> str:
return f"{version}-unknown-{random.randint(2**16, 2**17)}"
def check_environment(reproduction_information: dict[str, Any]) -> bool:
def check_environment(
settings: Settings,
reproduction_information: dict[str, Any],
) -> bool | None:
mismatch_severity: MismatchSeverity | None = None
system_mismatches = []
@@ -361,22 +366,26 @@ def check_environment(reproduction_information: dict[str, Any]) -> bool:
)
)
print()
choice = prompt_select(
"How would you like to proceed?",
[
Choice(
title="Attempt to reproduce the model anyway",
value=True,
),
Choice(
title="Exit program",
value=False,
),
],
)
if settings.ignore_mismatches is None:
print()
return choice
return ask_if_unset(
settings.ignore_mismatches,
questionary.select(
"How would you like to proceed?",
choices=[
Choice(
title="Attempt to reproduce the model anyway",
value=True,
),
Choice(
title="Exit program",
value=False,
),
],
style=Style([("highlighted", "reverse")]),
),
)
else:
# There are no mismatches at all, so there is nothing to confirm.
return True
src/heretic/utils.py
+15 -110
View File
@@ -1,23 +1,19 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
# Copyright (C) 2025-2026 Philipp Emanuel Weidmann <pew@worldwidemann.com> + contributors
import getpass
import hashlib
import json
import os
import platform
import random
import tempfile
import traceback
from dataclasses import dataclass
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from typing import Any, TypeVar
from typing import TypeVar
import huggingface_hub
import numpy as np
import questionary
import tomli_w
import torch
from datasets import DatasetDict, ReadInstruction, load_dataset, load_from_disk
@@ -28,7 +24,7 @@ from huggingface_hub.utils import validate_repo_id
from optuna import Trial
from optuna.trial import FrozenTrial
from psutil import Process
from questionary import Choice, Style
from questionary import Question
from rich.console import Console
from .config import DatasetSpecification, Settings
@@ -41,6 +37,9 @@ from .system import (
is_xpu_available,
)
T = TypeVar("T")
print = Console(highlight=False).print
@@ -67,99 +66,6 @@ def print_memory_usage():
p("Driver (reserved) MPS memory", torch.mps.driver_allocated_memory())
def is_notebook() -> bool:
# Check for specific environment variables (Colab, Kaggle).
# This is necessary because when running as a subprocess (e.g. !heretic),
# get_ipython() might not be available or might not reflect the notebook environment.
if os.getenv("COLAB_GPU") or os.getenv("KAGGLE_KERNEL_RUN_TYPE"):
return True
# Check IPython shell type (for library usage).
try:
from IPython import get_ipython # ty:ignore[unresolved-import]
shell = get_ipython()
if shell is None:
return False
shell_name = shell.__class__.__name__
if shell_name in ["ZMQInteractiveShell", "Shell"]:
return True
if "google.colab" in str(shell.__class__):
return True
return False
except (ImportError, NameError, AttributeError):
return False
def prompt_select(message: str, choices: list[Any]) -> Any:
if is_notebook():
print()
print(message)
real_choices = []
for i, choice in enumerate(choices, 1):
if isinstance(choice, Choice):
print(f"[{i}] {choice.title}")
real_choices.append(choice.value)
else:
print(f"[{i}] {choice}")
real_choices.append(choice)
while True:
try:
selection = input("Enter number: ")
index = int(selection) - 1
if 0 <= index < len(real_choices):
return real_choices[index]
print(
f"[red]Please enter a number between 1 and {len(real_choices)}[/]"
)
except ValueError:
print("[red]Invalid input. Please enter a number.[/]")
else:
return questionary.select(
message,
choices=choices,
style=Style([("highlighted", "reverse")]),
).ask()
def prompt_text(
message: str,
default: str = "",
qmark: str = "?",
unsafe: bool = False,
) -> str:
if is_notebook():
print()
result = input(f"{message} [{default}]: " if default else f"{message}: ")
return result if result else default
else:
question = questionary.text(message, default=default, qmark=qmark)
if unsafe:
return question.unsafe_ask()
else:
return question.ask()
def prompt_path(message: str) -> str:
if is_notebook():
return prompt_text(message)
else:
return questionary.path(message, only_directories=True).ask()
def prompt_password(message: str) -> str:
if is_notebook():
print()
return getpass.getpass(message)
else:
return questionary.password(message).ask()
def format_duration(seconds: float) -> str:
seconds = round(seconds)
hours, seconds = divmod(seconds, 3600)
@@ -186,6 +92,16 @@ def format_exception(error: Exception) -> str:
return traceback.format_exc().strip()
def ask_if_unset(value: T, question: Question, unsafe: bool = False) -> T:
if value is None:
if unsafe:
return question.unsafe_ask()
else:
return question.ask()
else:
return value
def is_hf_path(path: str) -> bool:
"""Checks whether a path likely refers to a Hugging Face repository."""
@@ -297,9 +213,6 @@ def load_prompts(
]
T = TypeVar("T")
def batchify(items: list[T], batch_size: int) -> list[list[T]]:
return [items[i : i + batch_size] for i in range(0, len(items), batch_size)]
@@ -386,14 +299,6 @@ def generate_requirements_txt() -> str:
return "\n".join(requirements) + "\n"
def set_seed(seed: int):
"""Sets the seed for all RNGs."""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
def format_hf_link(
path: str,
commit: str | None = None,