AI Engineer Portal

Your personal operating system for career transition.

Private mode

Exercise

Conversation memory manager with token budgeting

Conversation Memory Manager with Token Budgeting

Agents that run multi-turn conversations or long-running tasks quickly blow through context windows. A memory manager solves this by keeping the most important context within a token budget, using a combination of sliding window (keep recent messages) and summary compression (condense older messages into a summary).

What you are building

Create a MemoryManager class that:

Tracks conversation history — stores messages with role, content, and token counts.
Enforces a token budget — when adding a message would exceed the budget, compress older messages.
Uses sliding window — always keeps the N most recent messages verbatim.
Compresses with summaries — when messages are evicted from the window, they are compressed into a running summary using a provided summarizer function.
Preserves system messages — the system prompt is never evicted or compressed.
Provides a get_messages method — returns the current conversation formatted for an LLM API call: system message + summary (if any) + recent window.

Why this matters

Token management is one of the most important and least glamorous parts of agent engineering. Without it, agents either crash with context-too-long errors or silently lose important context. The sliding window + summary pattern is used by virtually every production agent that handles multi-turn conversations.

Understanding token budgeting also helps you reason about cost: if your agent uses 100K tokens per conversation, that is real money at scale.

Constraints

Use a provided count_tokens(text) -> int function (simulated as len(text.split()) for this exercise).
Use a provided summarize(messages) -> str function for compression.
The system message budget is separate from the conversation budget.
When the summary itself exceeds 25% of the budget, re-summarize it (recursive compression).

Agents / advanced / Step 5 of 8

Practice stage

Agent architecture patterns

Hint

Focus on explicit control surfaces — function schemas, state machines, and structured outputs. Agents are most useful when the task requires dynamic tool selection or multi-step reasoning. Start with the simplest pattern (single tool call) before reaching for ReAct loops.

Next drill

Multi-agent handoff protocol

Success criteria

- Tools return structured, typed responses
- Agent completes the task within a bounded number of steps
- All tool calls include error handling and retries
- Memory/state management prevents unbounded context growth

Review checklist

- Tool schemas validate inputs and handle errors gracefully
- Agent loop has explicit termination conditions
- State is serializable and inspectable between steps
- Cost and token usage are tracked per invocation
- Fallback behavior exists for tool call failures

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable

@dataclass
class Message:
    """A single conversation message with token count."""
    role: str  # "system", "user", "assistant"
    content: str
    token_count: int = 0

class MemoryManager:
    """Manages conversation history within a token budget."""

def __init__(
        self,
        token_budget: int,
        window_size: int = 10,
        count_tokens: Callable[[str], int] | None = None,
        summarize: Callable[[list[Message]], str] | None = None,
    ) -> None:
        self.token_budget = token_budget
        self.window_size = window_size
        self.count_tokens = count_tokens or (lambda text: len(text.split()))
        self.summarize = summarize or self._default_summarize
        self._system_message: Message | None = None
        self._messages: list[Message] = []
        self._summary: str | None = None
        self._summary_tokens: int = 0

@staticmethod
    def _default_summarize(messages: list[Message]) -> str:
        # TODO: simple concatenation-based summary
        raise NotImplementedError

def set_system_message(self, content: str) -> None:
        """Set the system message (never evicted)."""
        # TODO: store with token count
        raise NotImplementedError

def add_message(self, role: str, content: str) -> None:
        """Add a message, triggering compression if budget exceeded."""
        # TODO: create Message with token count
        # TODO: append to _messages
        # TODO: check if total tokens exceed budget
        # TODO: if over budget, compress oldest messages outside the window
        raise NotImplementedError

def _compress(self) -> None:
        """Move messages outside the window into the summary."""
        # TODO: identify messages to compress (outside window)
        # TODO: summarize them and update _summary
        # TODO: if summary itself is too large (>25% budget), re-summarize
        raise NotImplementedError

def _total_tokens(self) -> int:
        """Calculate total token usage across all components."""
        # TODO: sum system + summary + message tokens
        raise NotImplementedError

def get_messages(self) -> list[dict[str, str]]:
        """Return messages formatted for an LLM API call."""
        # TODO: return [system, summary-as-user-msg, ...recent messages]
        raise NotImplementedError

Practice

Generate a variation

Generate a new exercise variation to deepen understanding or practice a related concept.

Attempt history

Recent submissions

Before you submit, decide what a strong answer should make obvious to the reviewer.

No attempts yet.