How to Configure the Transform Pipeline for Custom Compression in Headroom

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 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, 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. This configuration holds flags for enabling or disabling specific built-in transforms.

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.

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.

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.

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 and override should_apply and apply.

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:

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)

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

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

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

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	Core TransformPipeline implementation, circuit-breaker logic, and default transform ordering
headroom/config.py	Global configuration model (HeadroomConfig) and per-component configuration structs
headroom/transforms/base.py	Abstract Transform class that all custom transforms must subclass
headroom/transforms/content_router.py	Automatic routing to specialized compressors (code, JSON, logs, plain text)
headroom/transforms/smart_crusher.py	JSON-array statistical compressor for tool outputs
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 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. 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. 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.

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