# How to Implement Custom CompactStrategy Subclasses in Forge

> Learn how to implement custom CompactStrategy subclasses in Forge by extending the abstract base class and implementing the compact() method. Preserve system prompts easily.

- Repository: [Antoine/forge](https://github.com/antoinezambelli/forge)
- Tags: how-to-guide
- Published: 2026-05-22

---

**To implement a custom CompactStrategy subclass, extend the abstract base class `CompactStrategy` in [`src/forge/context/strategies.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/strategies.py) and implement the `compact()` method that returns a tuple of (new message list, phase integer) while preserving the system prompt and original user input.**

Forge uses **compact strategies** to enforce token budgets during LLM conversations. When context windows fill up, these strategies decide which historical messages to prune while keeping critical information intact. By creating custom CompactStrategy subclasses, you can tailor the compaction logic to your specific workflow needs—whether that means keeping recent assistant messages, truncating tool results, or applying aggressive multi-phase filtering.

## Understanding the CompactStrategy Interface

The compaction abstraction lives in [`src/forge/context/strategies.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/strategies.py) and centers on the `CompactStrategy` abstract base class (ABC). This interface defines a single required method that all implementations must provide:

```python
class CompactStrategy(ABC):
    @abstractmethod
    def compact(
        self,
        messages: list[Message],
        budget_tokens: int,
        *,
        step_hint: str = "",
    ) -> tuple[list[Message], int]:
        ...

```

A `ContextManager` instance holds a concrete strategy and invokes `compact()` via `maybe_compact()` whenever token usage approaches the configured limit. The method receives the full conversation history and must return a **new list** (never mutating the original) plus an integer **phase** indicating compaction aggressiveness (`0` means no changes were made).

## Core Requirements for Custom Implementations

Your subclass must satisfy four specific contracts to remain compatible with Forge’s context management pipeline:

1. **Method Signature** – Implement exactly `compact(messages, budget_tokens, *, step_hint="") → tuple[list[Message], int]`.

2. **Preservation Guarantees** – Never drop `messages[0]` (the system prompt) or `messages[1]` (the original user input). These indexes are protected by convention.

3. **Token-Budget Logic** – Use the supplied `budget_tokens` alongside the helper `_estimate_tokens(messages)` to decide whether compaction is necessary.

4. **Return Value Contract** – Return a copy of the message list (e.g., `list(messages)`) and a phase number. Return phase `0` if unchanged, otherwise `1` (or higher values for custom phase indicators).

The repository includes three reference implementations—`NoCompact`, `SlidingWindowCompact`, and `TieredCompact`—that demonstrate these rules in practice.

## Building a Custom Compact Strategy

### Minimal Implementation Example

Below is a complete custom strategy called `RecentOnlyCompact`. It retains the system prompt, the initial user message, and only the most recent *N* assistant messages, discarding everything else when token usage exceeds a threshold.

```python
from forge.context.strategies import CompactStrategy, _estimate_tokens
from forge.core.messages import Message, MessageRole

class RecentOnlyCompact(CompactStrategy):
    """Keep only the last *keep_recent* assistant messages.

    This strategy is useful when the workflow is shallow and you only need
    the most recent assistant output to stay in context.
    """

    def __init__(self, keep_recent: int = 3, compact_threshold: float = 0.80):
        self.keep_recent = keep_recent
        self.compact_threshold = compact_threshold

    def compact(
        self,
        messages: list[Message],
        budget_tokens: int,
        *,
        step_hint: str = "",
    ) -> tuple[list[Message], int]:
        # Estimate current token usage

        current_tokens = _estimate_tokens(messages)

        # If below threshold, do nothing

        if current_tokens < int(budget_tokens * self.compact_threshold):
            return list(messages), 0

        # Identify indexes of the last keep_recent assistant messages

        recent_assistant_idxs = [
            i for i, m in enumerate(messages) if m.role == MessageRole.ASSISTANT
        ][-self.keep_recent :]

        # Build compacted list: keep first two messages, then recent assistants

        compacted = [
            m
            for i, m in enumerate(messages)
            if i < 2 or i in recent_assistant_idxs
        ]

        # Return new list and phase 1

        return compacted, 1

```

### Extending Built-in Strategies

For workflows requiring the multi-phase approach of `TieredCompact` but with bespoke rules, subclass the existing strategy and override specific phase helpers. This example drops tool results entirely in Phase 1 rather than truncating them:

```python
from forge.context.strategies import TieredCompact
from forge.core.messages import MessageType

class HybridTieredCompact(TieredCompact):
    """Same three-phase logic as TieredCompact but drops tool results
    completely in Phase 1 rather than truncating them.
    """

    def _phase1(self, messages, eligible_end):
        result = []
        for i, msg in enumerate(messages):
            if 2 <= i < eligible_end and msg.metadata.type == MessageType.TOOL_RESULT:
                # Skip the whole tool result instead of truncating

                continue
            result.append(msg)
        return result

```

## Integrating Your Strategy with ContextManager

Once defined, pass your custom strategy to the `ContextManager` constructor alongside your token budget. The manager handles the orchestration:

```python
from forge.context.manager import ContextManager

ctx = ContextManager(
    strategy=RecentOnlyCompact(keep_recent=4),
    budget_tokens=4000,
)

# During your workflow, the manager automatically evaluates compaction:

messages = ctx.maybe_compact(messages, step_index=current_step)

```

The `ContextManager` delegates to your strategy’s `compact()` method only when internal heuristics suggest the budget might be exceeded, respecting the `compact_threshold` you define within your implementation.

## Key Source Files Reference

When implementing custom CompactStrategy subclasses, reference these Forge source files:

- **[`src/forge/context/strategies.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/strategies.py)** – Defines the `CompactStrategy` ABC and built-in implementations (`NoCompact`, `SlidingWindowCompact`, `TieredCompact`).
- **[`src/forge/core/messages.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/core/messages.py)** – Provides `Message`, `MessageMeta`, `MessageRole`, and `MessageType` enums used for filtering logic.
- **[`src/forge/context/manager.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/manager.py)** – Houses `ContextManager`, which coordinates token budgeting and delegates to your compact strategy.
- **[`tests/unit/test_strategies.py`](https://github.com/antoinezambelli/forge/blob/main/tests/unit/test_strategies.py)** – Unit tests documenting expected behaviors; use this as a reference for assertions your custom strategy should satisfy.

## Summary

- Extend **`CompactStrategy`** from [`src/forge/context/strategies.py`](https://github.com/antoinezambelli/forge/blob/main/src/forge/context/strategies.py) to create custom compaction logic.
- Implement the **`compact()`** method with the exact signature `(list[Message], int, *, str) -> tuple[list[Message], int]`.
- Always preserve **`messages[0]`** (system prompt) and **`messages[1]`** (original user input) in your output.
- Return a **new list** (never mutate input) and a **phase integer** (`0` for no change, `1+` for custom phases).
- Leverage **`_estimate_tokens()`** to evaluate token counts against `budget_tokens`.
- Instantiate your strategy and pass it to **`ContextManager`** to activate it in your workflow.

## Frequently Asked Questions

### What does the phase integer returned by compact() represent?

The phase integer indicates how aggressive the compaction was. Return `0` when no messages are removed. Return `1` when your strategy performs its standard compaction. You may define additional phase values (e.g., `2`, `3`) to signal increasingly aggressive pruning, similar to how `TieredCompact` uses multiple phases.

### Can a custom strategy modify the messages list in place?

No. The `compact()` method must return a **new list** containing the filtered messages. Mutating the input list violates the contract expected by `ContextManager` and can cause unexpected side effects in your conversation history. Use `list(messages)` or list comprehensions to generate the output.

### How do I test a custom CompactStrategy implementation?

Write unit tests in a file similar to [`tests/unit/test_strategies.py`](https://github.com/antoinezambelli/forge/blob/main/tests/unit/test_strategies.py). Verify that your implementation preserves `messages[0]` and `messages[1]`, returns phase `0` when under budget, correctly estimates tokens using `_estimate_tokens()`, and returns a new list instance rather than the original reference.

### Should I subclass TieredCompact or CompactStrategy for multi-phase logic?

Subclass **`TieredCompact`** if you want to reuse its three-phase framework (removing nudges, then tool results, then reasoning) while overriding specific phase behaviors. Subclass the base **`CompactStrategy`** directly if you need a completely custom algorithm that does not follow the tiered pattern, such as the `RecentOnlyCompact` example above.