How to Customize the Headroom Transform Pipeline: 4 Proven Methods

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. 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 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), which typically returns CacheAligner → ContentRouter → optional ML compressor → RollingWindow. Each transform inherits from the abstract Transform base class defined in 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 (lines 84-88) skips the default _build_default_transforms method and uses your exact ordering.

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 (lines 54-59).

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 (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 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.

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 (lines 77-96), simulate calls apply with record_metrics=False on a deep-copied message list.

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.
• 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 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, base contracts in headroom/transforms/base.py, and configuration in 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 (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, 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}