ContextManager

Manages conversation context and compaction.

Quick Example

from mamba_agents.context import ContextManager, CompactionConfig

manager = ContextManager(config=CompactionConfig(strategy="hybrid"))

# Add messages
manager.add_messages([
    {"role": "user", "content": "Hello"},
    {"role": "assistant", "content": "Hi!"},
])

# Get messages
messages = manager.get_messages()

# Check state
state = manager.get_context_state()
print(f"Tokens: {state.token_count}")

# Compact if needed
if manager.should_compact():
    result = await manager.compact()

API Reference

ContextManager

ContextManager(
    config: CompactionConfig | None = None,
    token_counter: TokenCounter | None = None,
)

Manages conversation context and compaction.

Tracks message history, monitors token usage, and applies compaction strategies when thresholds are reached.

Initialize the context manager.

PARAMETER	DESCRIPTION
config	Compaction configuration. TYPE: CompactionConfig | None DEFAULT: None
token_counter	Token counter instance. TYPE: TokenCounter | None DEFAULT: None

Source code in src/mamba_agents/context/manager.py

def __init__(
    self,
    config: CompactionConfig | None = None,
    token_counter: TokenCounter | None = None,
) -> None:
    """Initialize the context manager.

    Args:
        config: Compaction configuration.
        token_counter: Token counter instance.
    """
    self._config = config or CompactionConfig()
    self._counter = token_counter or TokenCounter()
    self._history = MessageHistory()
    self._compaction_history: list[CompactionResult] = []
    self._strategy = self._create_strategy()

config property

config: CompactionConfig

Get the compaction configuration.

RETURNS	DESCRIPTION
CompactionConfig	CompactionConfig instance.

add_messages

add_messages(messages: list[dict[str, Any]]) -> None

Add messages to the history.

PARAMETER	DESCRIPTION
messages	Messages to add. TYPE: list[dict[str, Any]]

Source code in src/mamba_agents/context/manager.py

def add_messages(self, messages: list[dict[str, Any]]) -> None:
    """Add messages to the history.

    Args:
        messages: Messages to add.
    """
    for msg in messages:
        role = msg.get("role", "")
        content = msg.get("content", "")

        if role == "system" and self._config.preserve_system_prompt:
            self._history.system_prompt = content
        else:
            self._history.messages.append(msg)

get_messages

get_messages() -> list[dict[str, Any]]

Get all messages.

RETURNS	DESCRIPTION
list[dict[str, Any]]	List of messages.

Source code in src/mamba_agents/context/manager.py

def get_messages(self) -> list[dict[str, Any]]:
    """Get all messages.

    Returns:
        List of messages.
    """
    return self._history.get_messages()

get_token_count

get_token_count() -> int

Get current token count.

RETURNS	DESCRIPTION
int	Approximate token count.

Source code in src/mamba_agents/context/manager.py

def get_token_count(self) -> int:
    """Get current token count.

    Returns:
        Approximate token count.
    """
    messages = self.get_messages()
    count = self._counter.count_messages(messages)

    if self._history.system_prompt:
        count += self._counter.count(self._history.system_prompt)

    return count

should_compact

should_compact() -> bool

Check if compaction threshold is reached.

RETURNS	DESCRIPTION
bool	True if compaction should be triggered.

Source code in src/mamba_agents/context/manager.py

def should_compact(self) -> bool:
    """Check if compaction threshold is reached.

    Returns:
        True if compaction should be triggered.
    """
    return self.get_token_count() >= self._config.trigger_threshold_tokens

compact async

compact() -> CompactionResult

Apply compaction strategy to reduce context size.

RETURNS	DESCRIPTION
CompactionResult	CompactionResult with details of what was done.

Source code in src/mamba_agents/context/manager.py

async def compact(self) -> CompactionResult:
    """Apply compaction strategy to reduce context size.

    Returns:
        CompactionResult with details of what was done.
    """
    messages = self.get_messages()

    result = await self._strategy.compact(
        messages,
        self._config.target_tokens,
        self._config.preserve_recent_turns,
    )

    # Update history with compacted messages
    self._history.messages = result.messages
    self._compaction_history.append(result)

    return result

get_context_state

get_context_state() -> ContextState

Get the current context state.

RETURNS	DESCRIPTION
ContextState	ContextState with current information.

Source code in src/mamba_agents/context/manager.py

def get_context_state(self) -> ContextState:
    """Get the current context state.

    Returns:
        ContextState with current information.
    """
    return ContextState(
        token_count=self.get_token_count(),
        message_count=len(self._history),
        system_prompt=self._history.system_prompt,
        compaction_history=self._compaction_history.copy(),
    )

get_system_prompt

get_system_prompt() -> str | None

Get the system prompt.

RETURNS	DESCRIPTION
str | None	The system prompt or None.

Source code in src/mamba_agents/context/manager.py

def get_system_prompt(self) -> str | None:
    """Get the system prompt.

    Returns:
        The system prompt or None.
    """
    return self._history.system_prompt

set_system_prompt

set_system_prompt(prompt: str) -> None

Set the system prompt.

PARAMETER	DESCRIPTION
prompt	The system prompt. TYPE: str

Source code in src/mamba_agents/context/manager.py

def set_system_prompt(self, prompt: str) -> None:
    """Set the system prompt.

    Args:
        prompt: The system prompt.
    """
    self._history.system_prompt = prompt

clear

clear() -> None

Clear all context.

Source code in src/mamba_agents/context/manager.py

def clear(self) -> None:
    """Clear all context."""
    self._history.clear()
    self._compaction_history.clear()

get_compaction_history

get_compaction_history() -> list[CompactionResult]

Get history of compactions.

RETURNS	DESCRIPTION
list[CompactionResult]	List of CompactionResult objects.

Source code in src/mamba_agents/context/manager.py

def get_compaction_history(self) -> list[CompactionResult]:
    """Get history of compactions.

    Returns:
        List of CompactionResult objects.
    """
    return self._compaction_history.copy()

ContextState dataclass

ContextState(
    token_count: int,
    message_count: int,
    system_prompt: str | None,
    compaction_history: list[CompactionResult] = list(),
)

Current state of the context.

ATTRIBUTE	DESCRIPTION
token_count	Current token count. TYPE: int
message_count	Number of messages. TYPE: int
system_prompt	The system prompt. TYPE: str | None
compaction_history	History of compactions performed. TYPE: list[CompactionResult]