Skip to content

Reference

Built from docstrings on each docs build.

astro_weights.edits

Edit types and recipes.

Edit dataclass

A set of low-rank deltas against one base model.

from_deltas classmethod

from_deltas(deltas: Iterable[LowRankDelta], base_model: str, base_sha256: dict[str, str] | None = None, meta: dict[str, object] | None = None) -> Edit

Build an edit, zero-padding every delta to the largest rank.

LowRankDelta dataclass

dW = B @ A for one linear module; B is [out, r], A is [r, in].

padded

padded(rank: int) -> LowRankDelta

Return this delta at rank by appending zero columns/rows (dense unchanged).

from_dense

from_dense(module: str, dW: ndarray, rank: int, max_rel_residual: float | None = None) -> tuple[LowRankDelta, float]

Truncated SVD of dW to rank; returns (delta, rel_residual).

rel_residual = |dW - B @ A|_F / |dW|_F, measured on the stored float32 factors. Raises ValueError if it exceeds max_rel_residual.

build_edit

build_edit(recipe: dict[str, Any], weights: Callable[[str], ndarray], directory: Path | str) -> Edit

Build the Edit a recipe describes; weights(module) returns that module's W0.

load_recipe

load_recipe(path: Path | str) -> dict[str, Any]

Read and validate a recipe file; arrays are not loaded.

save_recipe

save_recipe(path: Path | str, recipe: dict[str, Any]) -> None

Write a recipe as JSON with sorted keys and a trailing newline.

column_shift

column_shift(module: str, shift: ndarray, idx: ndarray, coef: ndarray, n_in: int) -> LowRankDelta

Bake a constant shift into input columns: W = W0 + outer(shift, c), c[idx] = coef.

masked_rank1

masked_rank1(module: str, W0: ndarray, u: ndarray, mask: ndarray, alpha: float) -> LowRankDelta

Low-rank form of W = W0 - alpha * outer(u, mask * (u @ W0)).

orthogonal_random

orthogonal_random(u: ndarray, seed: int) -> np.ndarray

Seeded unit vector orthogonal to u (random-edit control).

project_out

project_out(module: str, W0: ndarray, v: ndarray, strength: float = 1.0) -> LowRankDelta

Remove direction v from the layer's output: W = W0 - strength * outer(vhat, vhat @ W0).

rank1

rank1(module: str, b: ndarray, a: ndarray) -> LowRankDelta

dW = outer(b, a).

scale_column

scale_column(module: str, W0: ndarray, j: int, s: float) -> LowRankDelta

Multiply input column j by s: dW = (s - 1) * outer(W0[:, j], e_j).

scale_head

scale_head(module: str, W0: ndarray, head: int, n_heads: int, s: float) -> LowRankDelta

Multiply the input columns of one attention head's block by s.

top_k_mask

top_k_mask(scores: ndarray, k: int) -> np.ndarray

float32 0/1 vector with ones at the k largest scores.

zero_head

zero_head(module: str, W0: ndarray, head: int, n_heads: int) -> LowRankDelta

Zero one attention head's input columns (scale_head with s = 0).

astro_weights.export

Export and import of edits.

GGUF functions live in astro_weights.export.gguf (needs the gguf extra).

read_base_weight

read_base_weight(model_dir: Path | str, key: str) -> np.ndarray

Read one base-model weight as float32, following the shard index if present.

read_tensor

read_tensor(file: Path | str, key: str) -> np.ndarray

Read one F32, F16 or BF16 tensor as float32; ValueError on any problem.

tensor_sha256

tensor_sha256(file: Path | str, key: str) -> str

sha256 of the tensor's raw stored bytes.

model_card

model_card(edit: Edit, kind: str, license: str, title: str | None = None, summary: str | None = None, results: dict[str, str] | None = None) -> str

Render a model card for edit; kind is "adapter" or "gguf".

write_card

write_card(directory: Path | str, text: str, *, overwrite: bool = True) -> Path

Write text as README.md in directory and return its path.

Replaces an existing README.md unless overwrite=False, which raises FileExistsError instead.

load_merged_tensors

load_merged_tensors(path: Path | str) -> dict[str, np.ndarray]

Load merged weights keyed by module name, checking sha256, shape and dtype.

The manifest sits next to the tensors, so a matching hash only proves the two files agree with each other. It is not a signature and does not show the weights are authentic.

Raises ValueError on a missing file, unreadable weights, a malformed manifest, or any mismatch.

save_merged_tensors

save_merged_tensors(tensors: dict[str, ndarray], path: Path | str, base_model: str, edit_manifest_sha256: str | None = None) -> None

Write merged weights keyed by module name; stored as <module>.weight.

Accepts float32 or float16 arrays only. bf16 is not a numpy dtype: callers pass tensor.float().numpy(). edit_manifest_sha256 links the adapter manifest.

load_peft

load_peft(directory: Path | str) -> Edit

Load a PEFT LoRA adapter as an Edit with scaling folded into B.

read_peft

read_peft(directory: Path | str) -> tuple[Edit, list[str]]

Parse an adapter; ValueError on structural problems, hash problems returned.

save_peft

save_peft(edit: Edit, out_dir: Path | str, *, overwrite: bool = False) -> None

Write edit as a PEFT LoRA adapter (scaling 1) with an astro manifest.

The files are written to a temp directory, then moved in: beside out_dir when it is new, inside it when it exists (so its parent need not be writable and a mount point works). A non-empty out_dir raises FileExistsError unless overwrite is true; with overwrite only the three adapter files are replaced and any other file in out_dir stays. On any failure the temp directory is removed.

verify_peft

verify_peft(directory: Path | str) -> list[str]

Human-readable problems with an adapter directory; empty when fine.

astro_weights.export.gguf

GGUF export: llama.cpp LoRA adapter files and tensor splice into a base GGUF.

Needs the optional gguf extra (pip install astro-weights[gguf]).

Orientation: GGUF stores dims reversed relative to numpy/torch. GGUFReader tensor.shape is GGUF order (ne0 first); tensor.data is numpy order, i.e. reversed(tensor.shape). Example, Qwen3.5-9B-Q4_K_M blk.15.ffn_down.weight: reader shape [12288, 4096] ([intermediate, hidden]) -> numpy (4096, 12288) = [hidden, intermediate], the torch nn.Linear layout of HF mlp.down_proj.weight ([out, in]). So an HF [out, in] array is written and spliced as is, no transpose.

gguf_tensor_name

gguf_tensor_name(module: str) -> str

Map an HF module name to its GGUF base tensor name.

save_lora_gguf

save_lora_gguf(edit: Edit, path: Path | str, architecture: str, *, overwrite: bool = False) -> None

Write edit as a llama.cpp LoRA adapter GGUF (alpha = rank, so scale 1).

read_lora_gguf

read_lora_gguf(path: Path | str) -> tuple[dict[str, tuple[np.ndarray, np.ndarray]], dict[str, object]]

Read an adapter GGUF: ({base tensor name: (B, A)}, metadata).

Only F32 and F16 tensors are accepted. B and A are the raw factors; the LoRA scale is not applied: delta = (meta["alpha"] / rank) * B @ A, with rank = A.shape[0].

splice_gguf

splice_gguf(base: Path | str, out: Path | str, replace: Mapping[str, ndarray], out_type: str = 'q8_0') -> None

Write a copy of base with the named tensors replaced, stored as out_type.

Other tensors are copied as raw bytes with their original type. Replacement arrays are HF [out, in] weights; a transposed array is accepted and transposed back.

verify_splice

verify_splice(base: Path | str, out: Path | str, names: list[str], seed: int = 0) -> list[str]

Problems found comparing a spliced GGUF to its base; empty means ok.

astro_weights.runners.hf

Apply and remove an Edit on a transformers model through PEFT.

resolve_modules

resolve_modules(model: Module, edit: Edit) -> dict[str, str]

Map each edit module name to the model's name for the same linear layer.

The same checkpoint is named model.layers.N... or model.language_model.layers.N... depending on the class that loads it. An exact name wins. Otherwise the name is matched from its last layers.<int> onward, among linear layers of the edit's shape. No match, or more than one, raises: a wrong guess would edit the wrong layer. A model that already carries an adapter raises too.

rounding_error

rounding_error(model: Module, edit: Edit) -> dict[str, float]

Relative error of each delta after its factors are cast to the weight dtype.

Per edit module name, |cast(B) @ cast(A) - B @ A|_F / |B @ A|_F, where cast rounds to that module's weight dtype (what apply loads); the math is float64. A float32 or float64 model gives 0.0, as does a zero delta. A factor that overflows the dtype gives inf. The model is not modified.

apply

apply(model: Module, edit: Edit, adapter_name: str = 'astro', *, max_rounding_error: float | None = None) -> PeftModel

Wrap only the edit's modules in LoRA layers carrying the edit's factors.

Edit module names map to model names through resolve_modules. The factors are cast to each weight's dtype; with max_rounding_error set, a module whose rounding_error exceeds it raises ValueError. Every ValueError is raised before the model is touched. Raises RuntimeError (after unwrapping again) if the adapter did not wrap every edit module.

remove

remove(peft_model: PeftModel) -> nn.Module

Drop the LoRA layers (no merge); base weights were never modified.

swap

swap(peft_model: PeftModel, edit: Edit, adapter_name: str = 'astro') -> PeftModel

Replace the applied edit with another: remove, then apply on the base.

The new edit's modules are resolved first, so a bad edit raises before anything is removed and peft_model stays usable.

merged_weights

merged_weights(model: Module, edit: Edit) -> dict[str, torch.Tensor]

Edited weights W0 + B @ A per module, keyed by the model's module name.

model must be the plain base model with floating-point weights. The sum is done in at least float32 (float64 weights stay float64) and cast back to the weight's dtype; results live on CPU. The model is not modified.

astro_weights.runners.generate

Run one eval arm: greedy, records written as made, settings recorded.

generate_records

generate_records(model: Any, tokenizer: Any, items: Sequence[Mapping[str, Any]], arm: str, scorer: Callable[[str, str], bool], max_new_tokens: int, seed: int = 0, chat: bool = True, *, out: Path | str | None = None, resume: bool = False, batch_size: int = 1) -> list[Record]

Generate and score each item; items carry item_id, prompt, gold.

scorer(gold, output) sees the full decoded text. If the model emits a think block, wrap the scorer yourself (e.g. with answer_part), or a \boxed{} inside the thinking can be graded. The model runs in eval mode (restored after).

With out, each record is appended to that JSONL file (the write_jsonl line format) and flushed as soon as it exists, so a crash keeps what was done. A non-empty out raises FileExistsError unless resume=True. Resuming keeps the records already in the file and skips their items; a last line cut short by a crash is dropped and redone. A record whose arm or seed differs, whose item_id is not in items, or whose prompt hash differs from the prompt sent now raises ValueError: the file belongs to another run. The returned list holds every record in the order of items, resumed or new.

batch_size > 1 generates that many items at once with left padding (the tokenizer needs a pad token; its padding_side is restored after). n_output_tokens counts tokens up to and including the first EOS. Greedy outputs can differ from batch_size=1: padding can change float rounding, and a near tie may then resolve the other way. Use one batch size for both arms of a paired comparison. Batches sit at fixed positions of items, and a batch is scored before any of its records is written, so a resume with the same batch_size rebuilds the batches of an uninterrupted run. (Records an earlier run wrote with another batch_size are kept as they are.) The default batch_size=1 runs each prompt alone, unpadded.

run_header

run_header(model: Any, tokenizer: Any, max_new_tokens: int, seed: int, chat: bool, *, batch_size: int = 1) -> dict[str, Any]

Settings that decide the outputs, for the run log.

astro_weights.runners.endpoint

Run one eval arm against an OpenAI-compatible HTTP API (llama-server, vLLM).

Standard library plus astro_weights.eval only: no torch, no extra dependency. Serve the base and the edit under two model names and call this once per name.

generate_records_http

generate_records_http(base_url: str, model: str, items: Sequence[Mapping[str, Any]], arm: str, scorer: Callable[[str, str], bool], max_new_tokens: int, *, seed: int = 0, api_key: str | None = None, timeout: float = 120.0, extra_body: Mapping[str, Any] | None = None, out: str | Path | None = None) -> list[Record]

Generate and score each item over {base_url}/v1/chat/completions.

One request per item, temperature 0. extra_body keys are merged last into the request. The output is message.content; reasoning_content is ignored, so wrap scorer yourself if the model thinks inline. out appends each record to a JSONL file as it is made; the file must not already hold records.

run_header_http

run_header_http(base_url: str, model: str, max_new_tokens: int, seed: int, extra_body: Mapping[str, Any] | None = None, *, api_key: str | None = None, timeout: float = 120.0) -> dict[str, Any]

Settings that decide the outputs, plus what the server reports, for the run log.

The server's /v1/models entry and system_fingerprint are best effort: left out when the server does not give them. The api key is never recorded.

astro_weights.directions

Difference-of-means directions and a gate for whether a direction is real.

GateResult dataclass

Statistics of :func:direction_gate.

split_half_cos: cosine between the mean directions of two random halves. null_975: upper quantile of |cos| for sign-shuffled pairs vs the full direction. It is element int(0.975 * draws) - 1 of the sorted draws, so with the default 20 draws it is the 19th of 20 values (about the 95th percentile, not a true 97.5th). heldout_auc: fraction of half-B pairs with positive projection on the half-A direction. per_pair_cos: mean cosine of half-B pairs with the half-A direction.

A statistic is NaN when a mean it needs has zero norm (for example the half-A mean).

direction_from_means

direction_from_means(mean_a: ndarray, mean_b: ndarray) -> np.ndarray

Unit vector normalize(mean_a - mean_b), float32; points b -> a.

direction_gate

direction_gate(pair_diffs: ndarray, seed: int = 0, draws: int = 20) -> GateResult

Is the direction of one layer real? pair_diffs is [P, H], pairs oriented a - b.

Pure noise gives split-half cosine near 0 and AUC near 0.5; the random-direction cosine floor is 1/sqrt(H). Needs draws >= 1 and finite input. Results with fewer than about 20 pairs are not meaningful: the halves are tiny and null_975 is near 1.

directions_from_pairs

directions_from_pairs(res_a: ndarray, res_b: ndarray) -> np.ndarray

Per-layer directions [L, H] from per-pair residual means [P, L, H].

neuron_scores

neuron_scores(u: ndarray, W_down: ndarray, act_contrast: ndarray) -> np.ndarray

(u @ W_down) * act_contrast, shape [intermediate].

act_contrast is the mean difference of the down_proj input, [intermediate].

stability

stability(u: ndarray, pair_diffs: ndarray) -> float

Mean cosine of each row of pair_diffs [n_pairs, hidden] with u.

astro_weights.eval

Paired eval statistics from per-item records. Never runs a model.

PairedResult dataclass

Paired statistics; scores and token counts are per-item means over seeds.

n_clusters is the number of distinct cluster_ids, or None when records carry none. When set, se, ci95, tokens_se, both MDEs and tokens_ci95 account for clustering, and small_n also flags fewer than 100 clusters. McNemar always treats items as independent.

Record dataclass

One generation for one (item, seed) in one arm.

answer_part

answer_part(text: str) -> str

Text after the last "", or all of it.

grade_choice

grade_choice(gold: str, answer: str) -> bool

True if the last \boxed{} holds the gold letter (A-J).

grade_exact

grade_exact(gold: str, answer: str) -> bool

True if the stripped strings are equal.

grade_math

grade_math(gold: str, answer: str, strip_commas: bool = False) -> bool

True if the answer's last boxed value equals gold, checked by math_verify.

last_boxed

last_boxed(s: str) -> str | None

Content of the last \boxed{...} in s (brace matched), or None.

check_headers

check_headers(base: dict[str, Any], edit: dict[str, Any], ignore: Collection[str] = ('device',)) -> None

Raise ValueError listing every key that differs or is missing on one side.

header_path

header_path(records_path: str | Path) -> Path

Header file for a records file: X.jsonl -> X.header.json.

read_header

read_header(path: str | Path) -> dict[str, Any]

Read a header file; ValueError if it is not a JSON object.

write_header

write_header(path: str | Path, header: dict[str, Any]) -> None

Write header as JSON with sorted keys.

compare

compare(base: Sequence[Record], edit: Sequence[Record], n_boot: int = 10000, seed: int = 0) -> PairedResult

Paired comparison of two arms; scores and tokens are averaged over seeds per item first.

If records carry cluster_id (every record of both arms, the same id for an item across arms and seeds), SEs, the score CI and the MDEs use the cluster-robust SE (see _clustered_se) and the token bootstrap resamples clusters. With fewer than 2 clusters these are NaN. Without cluster_id items are treated as independent.

exceeds_floor

exceeds_floor(result: PairedResult, floor: PairedResult) -> dict[str, bool]

Whether result is significant and larger than the base-vs-base noise floor.

floor is compare(base_run_1, base_run_2): the same arm run twice under the same settings. True for a metric when its CI excludes 0 and |diff| > |floor diff| + 1.96 floor SE. NaN anywhere gives False.

format_report

format_report(r: PairedResult, floor: PairedResult | None = None) -> str

A few plain lines summarising a PairedResult; floor adds noise-floor verdicts.

pair

pair(base: Sequence[Record], edit: Sequence[Record]) -> list[tuple[Record, Record]]

Join two arms on (item_id, seed), sorted; raise if they do not line up.

prompt_hash

prompt_hash(text: str) -> str

SHA-256 hex digest of the UTF-8 prompt.

read_jsonl

read_jsonl(path: str | Path) -> list[Record]

Read records written by write_jsonl; blank lines are skipped.

write_jsonl

write_jsonl(path: str | Path, records: Iterable[Record]) -> None

Write one JSON object per line.