trading_agent.agent.context_compressor

Compresses a conversation into a summary when it fills too much of the model's context window.

A model can only read so many tokens at once: 32,768 for Gemma and Qwen as configured. When a conversation passes 70% of that, the agent loop first gives the desk a silent memory-flush turn to save what matters with add_note or Memory.md, and then this compressor asks the same model to summarise the older messages. The conversation starts again from that summary, followed by the current turn's messages, which are kept word for word.

If a single long turn passes 85% between tool rounds, the compressor runs at once without the flush, because Ollama would otherwise drop the oldest text, the instructions included, without saying so.

Typical usage example:

compressor = ContextCompressor('main', 32768, 0.70, 0.85, event_bus) if compressor.is_full(conversation): await compressor.compress(conversation, backend, len(conversation.messages), 'threshold', trigger_id)

FLUSH_REQUEST module-attribute

FLUSH_REQUEST = "[Memory flush: your conversation is about to be compressed into a summary to free up context. Before it is, save anything you will need later that is not saved yet: record lessons with add_note, and put lasting facts in Memory.md. Do not place, change or cancel orders in this turn. Reply with one short line saying what you saved, or 'Nothing to save'.]"

SUMMARY_FOCUS module-attribute

SUMMARY_FOCUS = "open positions and orders with their ids, prices and quantities; the decisions made and why; the user's instructions and preferences; plans and pending tasks; errors and what was learned from them"

ContextCompressor

ContextCompressor(desk_name, context_window, threshold, emergency_threshold, event_bus)

Decides when one backend's conversation is too full, and compresses it.

Creates the compressor.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk, used to tag events.	required
context_window	int	How many tokens the model is given.	required
threshold	float	The share of the window at which to compress before a turn, such as 0.70.	required
emergency_threshold	float	The share at which to compress in the middle of a turn, such as 0.85.	required
event_bus	EventBus	Where each compression is published.	required

Source code in src/trading_agent/agent/context_compressor.py

def __init__(
    self,
    desk_name: str,
    context_window: int,
    threshold: float,
    emergency_threshold: float,
    event_bus: event_bus_module.EventBus,
):
    """Creates the compressor.

    Args:
        desk_name (str): The desk, used to tag events.
        context_window (int): How many tokens the model is given.
        threshold (float): The share of the window at which to compress before a turn, such as 0.70.
        emergency_threshold (float): The share at which to compress in the middle of a turn, such as 0.85.
        event_bus (event_bus_module.EventBus): Where each compression is published.
    """
    self._desk_name = desk_name
    self._context_window = context_window
    self._threshold = threshold
    self._emergency_threshold = emergency_threshold
    self._event_bus = event_bus

is_full

is_full(conversation)

Says whether the conversation has reached the normal threshold.

Parameters:

Name	Type	Description	Default
conversation	Conversation	The conversation.	required

Returns:

Type	Description
bool	True when it should be compressed before the next turn.

Source code in src/trading_agent/agent/context_compressor.py

def is_full(self, conversation: conversation_module.Conversation) -> bool:
    """Says whether the conversation has reached the normal threshold.

    Args:
        conversation (conversation_module.Conversation): The conversation.

    Returns:
        bool: True when it should be compressed before the next turn.
    """
    limit = self._threshold * self._context_window
    return conversation.estimated_tokens() >= limit

is_overflowing

is_overflowing(conversation)

Says whether the conversation has reached the emergency threshold.

Parameters:

Name	Type	Description	Default
conversation	Conversation	The conversation.	required

Returns:

Type	Description
bool	True when it must be compressed before the next model call.

Source code in src/trading_agent/agent/context_compressor.py

def is_overflowing(
    self,
    conversation: conversation_module.Conversation,
) -> bool:
    """Says whether the conversation has reached the emergency threshold.

    Args:
        conversation (conversation_module.Conversation): The conversation.

    Returns:
        bool: True when it must be compressed before the next model call.
    """
    limit = self._emergency_threshold * self._context_window
    return conversation.estimated_tokens() >= limit

compress async

compress(conversation, backend, keep_from, reason, trigger_id)

Summarises the messages before keep_from and restarts the conversation from the summary.

Parameters:

Name	Type	Description	Default
conversation	Conversation	The conversation to compress.	required
backend	ModelBackend	The model that writes the summary.	required
keep_from	int	The index of the first message to keep word for word, normally the current turn's user message.	required
reason	str	"threshold", "emergency" or "command", for the event.	required
trigger_id	str	The trigger of the turn that caused it, for the event.	required

Returns:

Type	Description
bool	True when the conversation was compressed, False when there was nothing older to summarise.

Raises:

Type	Description
BackendUnavailableError	The model could not write the summary.

Source code in src/trading_agent/agent/context_compressor.py

async def compress(
    self,
    conversation: conversation_module.Conversation,
    backend: backend_module.ModelBackend,
    keep_from: int,
    reason: str,
    trigger_id: str,
) -> bool:
    """Summarises the messages before keep_from and restarts the conversation from the summary.

    Args:
        conversation (conversation_module.Conversation): The conversation to compress.
        backend (backend_module.ModelBackend): The model that writes the summary.
        keep_from (int): The index of the first message to keep word for word, normally the current turn's user message.
        reason (str): "threshold", "emergency" or "command", for the event.
        trigger_id (str): The trigger of the turn that caused it, for the event.

    Returns:
        bool: True when the conversation was compressed, False when there was nothing older to summarise.

    Raises:
        backend_module.BackendUnavailableError: The model could not write the summary.
    """
    earlier = conversation.messages[:keep_from]
    kept = conversation.messages[keep_from:]
    if not earlier:
        return False
    tokens_before = conversation.estimated_tokens()
    summary = await self._summary(earlier, backend)
    share = round(100 * tokens_before / self._context_window)
    block = '\n'.join(
        [
            f'[Your earlier conversation was compressed because it filled about {share}% of your context window. Your notes and Memory.md hold what you chose to keep, and search_sessions finds the full record. Summary of the earlier conversation:]',
            summary,
            '[End of the summary.]',
        ]
    )
    conversation.restart(
        [
            messages.ConversationMessage.user(block),
            *kept,
        ]
    )
    self._event_bus.publish(
        event_module.Event.now(
            event_module.EventKind.CONTEXT_COMPRESSED,
            self._desk_name,
            {
                'trigger_id': trigger_id,
                'backend': backend.name,
                'reason': reason,
                'tokens_before': tokens_before,
                'context_window': self._context_window,
                'messages_summarised': len(earlier),
                'messages_kept': len(kept),
                'summary_characters': len(summary),
            },
        )
    )
    return True