Adding conveniences for your model¶
TorchLens already knows how to feed text to a Hugging Face model, how to name ImageNet classes and where the attention heads of common transformer blocks live. Three registries let you teach it the same things about your own model:
| You want | Register with | Every later tl.trace of your model |
|---|---|---|
Pass raw input (a string, a record) straight to tl.trace |
tl.preprocessing.register |
converts the raw input before it reaches forward |
| Read class labels instead of logit indices | tl.postprocessing.register |
decodes the output into labelled top-k rows |
| Address a named piece of a module, such as attention heads | tl.facets.register |
serves it as trace.modules[...].facets[name] |
Each registers a plain function and affects future traces only: what a trace
recorded before you registered stays as it was (a later read-time
decode_output() call reads the registry as it is then). unregister(name) (or tl.facets.reset())
removes it again. The two processing namespaces load on import:
import torchlens.preprocessing and import torchlens.postprocessing.
Each example below runs on its own.
A custom input preprocessor¶
An input detector receives the model, the raw input and the rest of the
tl.trace call as keyword arguments. It returns None for input it does not
handle, and TorchLens carries on as if it were not registered. To claim the
input, it converts it and calls tl.trace again with the converted value,
passing **kwargs through so the caller's capture=, save= and other options
still apply. The detector returns None for a tensor, so the second tl.trace
call takes the ordinary path.
import torch
from torch import nn
import torchlens as tl
import torchlens.preprocessing
VOCAB = "abcdefghijklmnopqrstuvwxyz "
class CharModel(nn.Module):
"""Embeds lowercase characters and averages them into one vector."""
def __init__(self) -> None:
super().__init__()
self.embed = nn.Embedding(len(VOCAB), 8)
def forward(self, codes: torch.Tensor) -> torch.Tensor:
return self.embed(codes).mean(dim=1)
def encode_chars(text):
"""Turn a string into a batch of one row of character codes."""
return torch.tensor([[VOCAB.index(ch) for ch in text.lower() if ch in VOCAB]])
@tl.preprocessing.register(name="char_codes", priority=5)
def char_codes(model, raw, **kwargs):
"""Trace a CharModel called on a string."""
if not isinstance(model, CharModel) or not isinstance(raw, str):
return None # tensors and other models take the usual path
return tl.trace(model, encode_chars(raw), **kwargs)
trace = tl.trace(CharModel(), "hello world")
print(trace.input_layers) # one input layer: the character codes of "hello world"
tl.preprocessing.unregister("char_codes")
Detectors run in ascending priority= order (default 100); the built-in Hugging
Face text, multimodal and image routes sit at 10, 20 and 30, so a priority
below 10 goes first. A call that already chooses its own input handling skips
every detector: an explicit capture=tl.options.CaptureOptions(input_transform=...),
a backend=, or a chunked forward (CaptureOptions(chunk_size=...)).
A custom output decoder¶
An output detector receives the model's output and a meta dict and returns a
ResolvedPostprocessing record or None. For class labels the record names
the style "classification" and carries the labels in config["id2label"].
TorchLens runs the detectors once at the end of each capture (meta["style"]
is None then) and again whenever you call
trace.decode_output(style=...) with a style no built-in decoder handles.
meta["model_class_name"] is the class name of the traced model.
import torch
from torch import nn
import torchlens as tl
import torchlens.postprocessing
from torchlens.postprocessing import ResolvedPostprocessing
PET_LABELS = {0: "cat", 1: "dog", 2: "rabbit"}
class PetClassifier(nn.Module):
"""Scores a 4-number feature vector as one of three pets."""
def __init__(self) -> None:
super().__init__()
self.fc = nn.Linear(4, 3)
def forward(self, x: torch.Tensor) -> torch.Tensor:
return self.fc(x)
@tl.postprocessing.register(name="pet_labels", priority=5)
def pet_labels(outputs, meta):
"""Name the three logits of a PetClassifier."""
if meta.get("model_class_name") != "PetClassifier":
return None
if meta.get("style") not in (None, "pet_labels"):
return None
return ResolvedPostprocessing(
source="project",
identifier="pet_labels",
verified=True,
config={"num_labels": len(PET_LABELS), "id2label": PET_LABELS},
description="pet label bank",
style="classification",
selected_output_head=meta.get("head"),
label_source="pet_labels",
label_source_version="v1",
confidence=1.0,
)
trace = tl.trace(PetClassifier(), torch.randn(2, 4))
print(trace.output_table(k=1)) # one row per batch item: batch_item, label, prob
print(trace.provenance.output_postprocessor.identifier) # pet_labels
tl.postprocessing.unregister("pet_labels")
The built-in decoders (a Hugging Face config.id2label, verified ImageNet-1k
widths, Hugging Face text) sit at priorities 10 to 30, so a decoder at the
default priority 100 runs only when none of them claims the output; pass a
priority below 10 to go first. An on-demand trace.decode_output(style="pet_labels")
reads the registry as it is when you call it.
A custom facet¶
A facet recipe receives the TorchLens record of a module of the registered class
and returns named FacetSpecs: where a value lives (here, the output op of the
query projection) plus a chain of views over it (here, split into heads).
import torch
from torch import nn
import torchlens as tl
from torchlens.facets import FacetSpec
N_HEADS, D_HEAD = 2, 4
class TinyAttention(nn.Module):
"""Single-layer self-attention with two heads of width four."""
def __init__(self) -> None:
super().__init__()
self.q = nn.Linear(N_HEADS * D_HEAD, N_HEADS * D_HEAD)
self.k = nn.Linear(N_HEADS * D_HEAD, N_HEADS * D_HEAD)
def forward(self, x: torch.Tensor) -> torch.Tensor:
scores = self.q(x) @ self.k(x).transpose(-2, -1)
return scores.softmax(dim=-1) @ x
class Encoder(nn.Module):
def __init__(self) -> None:
super().__init__()
self.attn = TinyAttention()
def forward(self, x: torch.Tensor) -> torch.Tensor:
return self.attn(x)
@tl.facets.register(class_name="TinyAttention", target_scope="module", facets=("q_heads",))
def tiny_attention_heads(module):
"""Expose the query projection split into heads."""
prefix = "" if module.address == "self" else f"{module.address}."
q_proj = module.trace.modules[f"{prefix}q"]
q_out = module.trace.ops[q_proj.calls[0].output_ops[0]]
return {"q_heads": FacetSpec.from_home(q_out).heads(N_HEADS, D_HEAD)}
trace = tl.trace(Encoder(), torch.randn(1, 5, N_HEADS * D_HEAD))
print(trace.modules["attn"].facets["q_heads"].shape)
tl.facets.reset() # back to the built-in recipes
See Facets for the full recipe contract, including reading head counts from module config.
All three on one model¶
import torch
from torch import nn
import torchlens as tl
import torchlens.postprocessing
import torchlens.preprocessing
from torchlens.facets import FacetSpec
from torchlens.postprocessing import ResolvedPostprocessing
VOCAB = "abcdefghijklmnopqrstuvwxyz "
MOODS = {0: "calm", 1: "cheerful", 2: "grumpy"}
N_HEADS, D_HEAD = 2, 4
class TinyAttention(nn.Module):
def __init__(self) -> None:
super().__init__()
self.q = nn.Linear(N_HEADS * D_HEAD, N_HEADS * D_HEAD)
self.k = nn.Linear(N_HEADS * D_HEAD, N_HEADS * D_HEAD)
def forward(self, x: torch.Tensor) -> torch.Tensor:
scores = self.q(x) @ self.k(x).transpose(-2, -1)
return scores.softmax(dim=-1) @ x
class MoodClassifier(nn.Module):
"""Reads a lowercase string and scores it as one of three moods."""
def __init__(self) -> None:
super().__init__()
self.embed = nn.Embedding(len(VOCAB), N_HEADS * D_HEAD)
self.attn = TinyAttention()
self.head = nn.Linear(N_HEADS * D_HEAD, 3)
def forward(self, codes: torch.Tensor) -> torch.Tensor:
return self.head(self.attn(self.embed(codes)).mean(dim=1))
@tl.preprocessing.register(name="mood_text", priority=5)
def mood_text(model, raw, **kwargs):
if not isinstance(model, MoodClassifier) or not isinstance(raw, str):
return None
codes = torch.tensor([[VOCAB.index(ch) for ch in raw.lower() if ch in VOCAB]])
return tl.trace(model, codes, **kwargs)
@tl.postprocessing.register(name="mood_labels", priority=5)
def mood_labels(outputs, meta):
if meta.get("model_class_name") != "MoodClassifier" or meta.get("style") is not None:
return None
return ResolvedPostprocessing(
source="project",
identifier="mood_labels",
verified=True,
config={"num_labels": len(MOODS), "id2label": MOODS},
description="mood label bank",
style="classification",
selected_output_head=meta.get("head"),
)
@tl.facets.register(class_name="TinyAttention", target_scope="module", facets=("q_heads",))
def mood_heads(module):
q_proj = module.trace.modules[f"{module.address}.q"]
q_out = module.trace.ops[q_proj.calls[0].output_ops[0]]
return {"q_heads": FacetSpec.from_home(q_out).heads(N_HEADS, D_HEAD)}
trace = tl.trace(MoodClassifier(), "what a lovely day")
print(trace.output_table(k=1)["label"][0]) # one of calm, cheerful, grumpy
print(trace.modules["attn"].facets["q_heads"].shape)
tl.preprocessing.unregister("mood_text")
tl.postprocessing.unregister("mood_labels")
tl.facets.reset()
Where to go next¶
- Semantic input and output: the built-in routes, read-time
decode_output(style=, head=)and output provenance. - Facets: specs, gradients, intervention and patching on facets.