# How to Configure the Transform Pipeline for Custom Compression in Headroom

> Learn to configure the transform pipeline for custom compression in Headroom. Easily override defaults or add new strategies by passing custom Transform objects to the TransformPipeline constructor.

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

---

**You configure the transform pipeline for custom compression in Headroom by passing a custom list of `Transform` objects to the `TransformPipeline` constructor, overriding the default order or injecting entirely new compression strategies.**

Headroom optimizes LLM token usage through a modular transform pipeline that processes requests before they reach the model. The `TransformPipeline` class in [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) orchestrates built-in transforms like `CacheAligner` and `SmartCrusher`, but also exposes APIs for complete customization. By configuring the transform pipeline for custom compression, you can tailor token reduction strategies to your specific data patterns and requirements.

## Understanding the Transform Pipeline Architecture

The **transform pipeline** is the core compression engine in Headroom. It executes a series of `Transform` objects sequentially, where each transform receives the message list, processes it, and returns a `TransformResult` containing token counts and modified messages.

In [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py), the `TransformPipeline` class manages this execution order. By default, it constructs a pipeline based on `HeadroomConfig` settings, but you can override this by providing a specific `transforms` parameter during instantiation.

## Building a Custom Transform Pipeline

### Step 1: Configure Headroom Settings

Start by creating a `HeadroomConfig` object from [`headroom/config.py`](https://github.com/chopratejas/headroom/blob/main/headroom/config.py). This configuration holds flags for enabling or disabling specific built-in transforms.

```python
from headroom import HeadroomConfig

cfg = HeadroomConfig()

# Modify specific settings before building the pipeline

cfg.smart_crusher.enabled = False

```

### Step 2: Define Your Transform Order

Create a list of `Transform` instances in the exact execution order you require. Each transform must inherit from `headroom.transforms.base.Transform` and implement the `apply` method.

```python
from headroom.transforms import CacheAligner, SmartCrusher, ContentRouter

custom_transforms = [
    CacheAligner(cfg.cache_aligner),          # 1️⃣ Prefix stabilization

    SmartCrusher(cfg.smart_crusher),          # 2️⃣ JSON-array compression

    ContentRouter(),                          # 3️⃣ Intelligent routing (code, logs, etc.)

    # Add your own custom transform here

]

```

### Step 3: Instantiate the Pipeline

Pass your configuration and custom transform list to `TransformPipeline`.

```python
from headroom import TransformPipeline

pipeline = TransformPipeline(config=cfg, transforms=custom_transforms)

```

If you pass `transforms=None`, the pipeline falls back to the default order defined in `HeadroomConfig`.

### Step 4: Execute and Monitor

Run the pipeline against your message list and inspect the results.

```python
result = pipeline.apply(messages, model="gpt-4o")
print("Tokens saved:", result.tokens_before - result.tokens_after)
print("Transforms applied:", result.transforms_applied)

```

## Creating Custom Transform Classes

To implement domain-specific compression, subclass `Transform` from [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) and override `should_apply` and `apply`.

```python
from headroom.transforms.base import Transform
from headroom.types import TransformResult
from headroom.tokenizer import Tokenizer

class MyUpperCaseTransform(Transform):
    """A demo transform that upper-cases user-visible text."""
    name = "uppercase"

    def should_apply(self, messages, tokenizer, **kwargs):
        # Apply only if there is at least one user/assistant message

        return any(m["role"] in {"user", "assistant"} for m in messages)

    def apply(self, messages, tokenizer: Tokenizer, **kwargs) -> TransformResult:
        new_messages = []
        for msg in messages:
            if msg["role"] in {"user", "assistant"}:
                msg = msg.copy()
                msg["content"] = msg["content"].upper()
            new_messages.append(msg)

        tokens_before = tokenizer.count_messages(messages)
        tokens_after = tokenizer.count_messages(new_messages)

        return TransformResult(
            messages=new_messages,
            tokens_before=tokens_before,
            tokens_after=tokens_after,
            transforms_applied=[self.name],
        )

```

Add your custom transform to the pipeline list:

```python
pipeline = TransformPipeline(
    config=cfg,
    transforms=[CacheAligner(cfg.cache_aligner), MyUpperCaseTransform(), ContentRouter()],
)

```

## Advanced Configuration Options

Headroom exposes several advanced knobs for pipeline tuning:

- **`HEADROOM_PIPELINE_BREAKER_THRESHOLD`** and **`HEADROOM_PIPELINE_BREAKER_COOLDOWN_S`**: Environment variables that configure a circuit-breaker in `TransformPipeline._breaker_*` methods, temporarily bypassing the entire pipeline after N consecutive failures.
- **`config.generate_diff_artifact`**: A boolean field on `HeadroomConfig` that enables detailed `DiffArtifact` recording of per-transform token changes.
- **`config.pipeline_extensions`**: A list field accepting third-party observability hooks or extensions that inject automatically.
- **`config.content_router.enable_*` flags**: Boolean switches inside `ContentRouterConfig` (referenced from `HeadroomConfig`) that toggle specific compressors (code-aware, smart-crusher, log-compressor).

**Important**: Configuration changes only affect new `TransformPipeline` instances. You must recreate the pipeline after modifying `HeadroomConfig`.

## Practical Code Examples

### Minimal Custom Pipeline (Default Router Only)

```python
from headroom import HeadroomConfig, TransformPipeline

cfg = HeadroomConfig()
pipeline = TransformPipeline(config=cfg)          # uses default order

result = pipeline.apply(messages, model="claude-3")
print(result.transforms_applied)               # e.g. ["cache_aligner", "smart_crusher", "kompress"]

```

### Fully Custom Order with User-Written Transform

```python
from headroom import HeadroomConfig, TransformPipeline
from headroom.transforms import CacheAligner, SmartCrusher, ContentRouter
from my_transforms import MyUpperCaseTransform

cfg = HeadroomConfig()
custom_order = [
    MyUpperCaseTransform(),           # 1️⃣ Upper-case first

    CacheAligner(cfg.cache_aligner),  # 2️⃣ Stabilize prefix

    SmartCrusher(cfg.smart_crusher),  # 3️⃣ Crush JSON arrays

    ContentRouter(),                  # 4️⃣ Route remaining text

]

pipeline = TransformPipeline(config=cfg, transforms=custom_order)
result = pipeline.apply(messages, model="gpt-4")
print("Saved:", result.tokens_before - result.tokens_after)
print("History:", result.transforms_applied)

```

### Switching Off a Built-In Transform

```python
from headroom import HeadroomConfig, TransformPipeline
from headroom.transforms import CacheAligner, ContentRouter

cfg = HeadroomConfig()
cfg.smart_crusher.enabled = False

pipeline = TransformPipeline(
    config=cfg,
    transforms=[CacheAligner(cfg.cache_aligner), ContentRouter()],  # SmartCrusher omitted

)

```

### Using the Convenience Helper `create_pipeline`

```python
from headroom import create_pipeline
from headroom.config import CacheAlignerConfig

custom_cache_cfg = CacheAlignerConfig(enabled=True, normalize_whitespace=False)
pipeline = create_pipeline(cache_aligner_config=custom_cache_cfg)
result = pipeline.apply(messages, model="gpt-4")

```

## Key Source Files

| File | Role |
|------|------|
| [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py) | Core `TransformPipeline` implementation, circuit-breaker logic, and default transform ordering |
| [`headroom/config.py`](https://github.com/chopratejas/headroom/blob/main/headroom/config.py) | Global configuration model (`HeadroomConfig`) and per-component configuration structs |
| [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) | Abstract `Transform` class that all custom transforms must subclass |
| [`headroom/transforms/content_router.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/content_router.py) | Automatic routing to specialized compressors (code, JSON, logs, plain text) |
| [`headroom/transforms/smart_crusher.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/smart_crusher.py) | JSON-array statistical compressor for tool outputs |
| [`headroom/transforms/cache_aligner.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/cache_aligner.py) | Prefix-stabilization transform that improves cache hit rates |

## Summary

- The **transform pipeline** in Headroom is controlled by the `TransformPipeline` class, which accepts a configurable list of `Transform` objects.
- You can **override the default compression order** by passing a custom list to the constructor, or use the `create_pipeline` helper for simplified configuration.
- **Custom transforms** must inherit from `Transform` in [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py) and implement `should_apply` and `apply` methods.
- **Configuration changes** require pipeline reinstantiation; modify `HeadroomConfig` before creating the `TransformPipeline` instance.
- Advanced features include **circuit-breaker protection** (via environment variables), **diff artifact generation**, and **content routing** with granular enable flags.

## Frequently Asked Questions

### How do I disable specific transforms in the Headroom pipeline?

Set the corresponding enabled flag to `False` in `HeadroomConfig` before instantiating the pipeline, or omit the transform from your custom list when calling `TransformPipeline`. For example, set `cfg.smart_crusher.enabled = False` to disable JSON compression, then pass your custom transform list excluding that component.

### What is the base class for creating custom transforms in Headroom?

All custom transforms must inherit from `Transform` defined in [`headroom/transforms/base.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/base.py). You must implement the `should_apply` method to determine when the transform runs, and the `apply` method to execute your compression logic and return a `TransformResult` with token counts and modified messages.

### How does the circuit breaker work in Headroom's transform pipeline?

The circuit breaker is implemented in `TransformPipeline._breaker_*` methods within [`headroom/transforms/pipeline.py`](https://github.com/chopratejas/headroom/blob/main/headroom/transforms/pipeline.py). It monitors for consecutive failures and temporarily bypasses the entire pipeline when failures exceed the threshold set by the `HEADROOM_PIPELINE_BREAKER_THRESHOLD` environment variable, resetting after the cooldown period specified by `HEADROOM_PIPELINE_BREAKER_COOLDOWN_S`.

### Can I change the pipeline configuration after instantiation?

No. The `TransformPipeline` captures configuration settings at instantiation time. If you modify `HeadroomConfig` after creating the pipeline instance, those changes will not take effect. You must create a new `TransformPipeline` instance with the updated configuration to apply changes.