# How to Customize the Headroom Transform Pipeline: 4 Proven Methods

> Customize the Headroom transform pipeline with 4 proven methods. Control transforms via config flags, subclass base classes, or inject custom logic for enhanced circuit-breaker safety.

- Repository: [Tejas Chopra/headroom](https://github.com/chopratejas/headroom)
- Tags: how-to-guide
- Published: 2026-06-12

---

**To customize the Headroom transform pipeline, pass a custom list of transforms to the `TransformPipeline` constructor, toggle individual transforms via `HeadroomConfig` flags, or subclass the `Transform` base class to inject custom logic while inheriting built-in circuit-breaker safety.**

The Headroom library (chopratejas/headroom) provides a modular `TransformPipeline` that orchestrates LLM message compression through a series of configurable transforms. Whether you need to reorder the default compression steps, disable specific transforms like `CacheAligner`, or implement entirely custom compression strategies, you can customize the Headroom transform pipeline using the public API exposed in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py). This guide demonstrates four proven methods to modify pipeline behavior, from configuration tweaks to custom class implementations.

## Understanding the TransformPipeline Architecture

The `TransformPipeline` class in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) serves as the main orchestrator, iterating over a list of `Transform` objects while measuring token counts and recording metrics. According to the source code, customizing the pipeline follows this four-step flow:

1. **Instantiate `HeadroomConfig`** (or a subclass) and toggle individual transforms via its boolean fields like `cache_aligner.enabled`.
2. **Pass a custom list of `Transform` objects** to the `TransformPipeline` constructor to bypass the default builder.
3. **Optionally provide a `Provider`** (e.g., `OpenAIProvider`) to supply model-specific tokenizers.
4. **Run `apply` or `simulate`** on your messages to execute or preview transformations.

The default pipeline constructs transforms via `_build_default_transforms` (lines 99-133 of [`pipeline.py`](https://github.com/chopratejas/headroom/blob/main/pipeline.py)), which typically returns `CacheAligner` → `ContentRouter` → optional ML compressor → `RollingWindow`. Each transform inherits from the abstract `Transform` base class defined in [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py), implementing `should_apply` and `apply` methods.

## Method 1: Replace the Default Transform List

To completely customize the Headroom transform pipeline ordering or composition, pass a custom list to the `TransformPipeline` constructor. When you provide the `transforms` argument, the initialization logic in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) (lines 84-88) skips the default `_build_default_transforms` method and uses your exact ordering.

```python
from headroom.transforms import SmartCrusher, CacheAligner, RollingWindow
from headroom.transforms.pipeline import TransformPipeline

# Build a pipeline that runs SmartCrusher before CacheAligner

custom_transforms = [
    SmartCrusher(),          # Compress JSON tool results first

    CacheAligner(),          # Then detect volatile system content

    RollingWindow(),        # Finally drop old messages

]

pipeline = TransformPipeline(transforms=custom_transforms)
result = pipeline.apply(messages, model="gpt-4")

```

## Method 2: Toggle Individual Transforms via Configuration

For finer control without replacing the entire pipeline, use `HeadroomConfig` to enable or disable specific transforms. Each transform checks its configuration flag in `should_apply` before executing, as implemented in [`headroom/transforms/cache_aligner.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/cache_aligner.py) (lines 54-59).

```python
from headroom import TransformPipeline, HeadroomConfig

cfg = HeadroomConfig()
cfg.cache_aligner.enabled = False  # Disable CacheAligner detection

cfg.content_router.enabled = True  # Keep ContentRouter active

pipeline = TransformPipeline(config=cfg)

```

You can also use the convenience factory `create_pipeline` in [`headroom/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/pipeline.py) (lines 498-516) to build a pipeline with a custom `CacheAlignerConfig`.

## Method 3: Create Custom Transform Classes

To extend functionality, subclass `Transform` from [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) and implement the required interface. Custom transforms automatically participate in logging, timing, and the circuit-breaker safety net controlled by `HEADROOM_PIPELINE_BREAKER_THRESHOLD` and `HEADROOM_PIPELINE_BREAKER_COOLDOWN_S` environment variables.

```python
from headroom.transforms.base import Transform
from headroom.transforms.pipeline import TransformResult

class UppercaseTransform(Transform):
    name = "uppercase"

    def should_apply(self, messages, tokenizer, **kwargs):
        return kwargs.get("uppercase", False)

    def apply(self, messages, tokenizer, **kwargs):
        new_msgs = [
            {**msg, "content": msg["content"].upper() if isinstance(msg.get("content"), str) else msg["content"]}
            for msg in messages
        ]
        return TransformResult(
            messages=new_msgs,
            tokens_before=tokenizer.count_messages(messages),
            tokens_after=tokenizer.count_messages(new_msgs),
            transforms_applied=[self.name],
        )

# Inject the custom transform

pipeline = TransformPipeline(transforms=[UppercaseTransform(), SmartCrusher()])
result = pipeline.apply(messages, model="gpt-4", uppercase=True)

```

## Method 4: Simulate Changes Without Side Effects

The `simulate` method allows you to preview token savings without mutating the original message list. According to the source in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) (lines 77-96), `simulate` calls `apply` with `record_metrics=False` on a deep-copied message list.

```python
pipeline = TransformPipeline()
sim_result = pipeline.simulate(messages, model="gpt-4")
print(f"Would have saved {sim_result.tokens_before - sim_result.tokens_after} tokens")

```

## Summary

- **Pass custom transforms**: Supply a list to `TransformPipeline(transforms=[...])` to replace the default ordering built by `_build_default_transforms` in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py).
- **Toggle via config**: Use `HeadroomConfig` flags like `cache_aligner.enabled` to disable specific transforms without subclassing or replacing the pipeline.
- **Extend the base class**: Subclass `Transform` from [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) to implement custom compression logic that integrates with metrics and circuit breakers.
- **Simulate safely**: Use `pipeline.simulate()` to preview token savings via `TransformResult` without modifying original messages.
- **Source locations**: Core orchestration resides in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py), base contracts in [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py), and configuration in [`headroom/config.py`](https://github.com/chopratejas/headroom/blob/main/headroom/config.py).

## Frequently Asked Questions

### How do I change the order of transforms in the Headroom pipeline?

Pass a custom list to the `TransformPipeline` constructor using the `transforms` parameter. When you provide this argument, the initialization logic in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) (lines 84-88) automatically skips the default `_build_default_transforms` method and uses your exact ordering instead.

### Can I disable only the CacheAligner transform while keeping others active?

Yes. Instantiate `HeadroomConfig` and set `cfg.cache_aligner.enabled = False` before passing it to `TransformPipeline(config=cfg)`. The `CacheAligner.should_apply` method checks this boolean flag at lines 54-59 of [`headroom/transforms/cache_aligner.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/cache_aligner.py), causing it to skip execution while other transforms remain active.

### What methods must a custom transform implement?

Your class must inherit from `Transform` in [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) and implement two methods: `should_apply(self, messages, tokenizer, **kwargs)` returning a boolean, and `apply(self, messages, tokenizer, **kwargs)` returning a `TransformResult` object containing the modified messages, token counts, and a list of applied transform names.

### How do I test token savings without modifying my original messages?

Call `pipeline.simulate(messages, model="...")` instead of `apply()`. This method creates a deep copy of your messages and runs the pipeline without side effects, returning token counts and transformation metrics via `TransformResult` while leaving your original data unchanged.