Dynamic Context Eviction Algorithms for LLMs
Source: mortalapps.com- Dynamic Context Eviction Algorithms manage the limited context window of Large Language Models (LLMs) by intelligently removing less critical information.
- They solve the problem of context window exhaustion, preventing agents from losing crucial conversational history or relevant data due to token limits.
- In production, effective eviction policies reduce operational costs by minimizing token usage while maintaining agent performance and coherence.
- These algorithms prioritize context items based on factors like recency, relevance, and token cost, ensuring high-value information persists.
- Implementing dynamic policies allows for adaptive agent behavior, crucial for long-running conversations or complex, data-intensive tasks.
Why This Matters
Large Language Models operate within a finite context window, a critical constraint for AI agents engaging in multi-turn conversations or complex tasks. As interactions progress, this window can quickly fill, leading to the loss of older, potentially relevant information. This problem, known as context window exhaustion, directly impacts an agent's ability to maintain coherence, recall past details, and execute long-running plans. Dynamic context eviction algorithms were invented to address this by providing a principled mechanism to decide which pieces of information to discard when the context window is full, ensuring that the most valuable data remains. Ignoring this can lead to agents exhibiting 'forgetfulness,' generating irrelevant responses, or failing to complete tasks due to a lack of necessary context. In production, this translates to degraded user experience, increased token costs from redundant information, and unreliable agent performance. Engineers must implement these algorithms to build robust, cost-effective, and intelligent agents. When to use this approach versus simply truncating context depends on the application's sensitivity to information loss; for critical applications where every piece of context might matter, dynamic eviction offers a superior, more intelligent solution than naive truncation, which risks discarding vital information indiscriminately.
Core Concepts
The effective management of an LLM's context window is fundamental to agent performance. Dynamic context eviction algorithms are at the heart of this management.
- Context Window: The maximum sequence length (measured in tokens) that an LLM can process in a single inference call. This limit is imposed by the model architecture and computational resources.
- Token Budget: The specific allocation of tokens within the context window reserved for agent input, output, and internal reasoning. Eviction occurs when new information exceeds this budget.
- Context Item: A discrete unit of information stored in the agent's working memory, such as a user message, an agent's thought, a tool output, or a retrieved document. Each item has an associated token count.
- Eviction Policy: The specific algorithm or set of rules used to determine which context items to remove when the token budget is exceeded. Policies aim to retain the most valuable information.
- Recency: A metric indicating how recently a context item was added or accessed. More recent items are often considered more relevant for ongoing tasks.
- Relevance: A qualitative or quantitative measure of an item's importance to the current task or conversation. This can be determined by semantic similarity, explicit tagging, or LLM-based scoring.
- Token Cost: The number of tokens an item consumes. Eviction policies often consider this to maximize the utility of the remaining token budget by removing larger, less valuable items.
- Dynamic Eviction: An approach where the eviction policy adapts or uses multiple criteria (recency, relevance, cost) to make more intelligent decisions than static policies like FIFO or LRU alone.
| Policy Type | Description | Primary Metric | Trade-offs |
|---|---|---|---|
| Static | Fixed rule, e.g., always remove the oldest item. | Age | Simple, but can remove critical context. |
| Dynamic | Uses multiple factors (recency, relevance, cost) to score items for removal. | Score | Complex, but retains more valuable context. |
How It Works
Dynamic context eviction operates as a continuous process within an AI agent's memory management subsystem, triggered whenever new information threatens to exceed the LLM's context window token limit. The lifecycle involves several distinct phases:
1. Context Item Ingestion
When a new piece of information (e.g., user input, tool output, internal thought) is generated, it is first encapsulated as a ContextItem. This item includes its content, a timestamp of creation/last access, an estimated token count, and potentially an initial relevance score. The ContextManager attempts to add this item to the current context window.
2. Token Budget Check
Before adding the new item, the ContextManager calculates the total token count of all existing items plus the new item. If this sum is less than or equal to the configured MAX_TOKENS (or TOKEN_BUDGET), the item is simply added, and the process concludes.
3. Eviction Trigger and Policy Application
If the token budget is exceeded, the eviction process is triggered. The ContextManager invokes the configured dynamic eviction policy. This policy evaluates all existing ContextItems based on their attributes (recency, relevance, token cost) to assign an 'eviction score' to each. The goal is to identify items that, if removed, would free up enough tokens while minimizing the loss of critical information.
4. Item Selection and Removal
Items are sorted based on their eviction scores (e.g., lowest score first, indicating least importance). The policy iteratively selects and removes items from the context window until enough tokens are freed to accommodate the new item, plus a small buffer if configured. If, after removing all evictable items, there still isn't enough space, the new item might be rejected, or the oldest/largest remaining item might be forcibly removed, potentially logging a critical warning.
5. Context Window Update
Once sufficient space is created, the new ContextItem is added to the context window. The ContextManager then updates its internal state, including the total token count and potentially re-indexing items for faster retrieval or scoring in subsequent cycles. Metadata for evicted items (e.g., for auditing or long-term memory consolidation) may be sent to a separate component.
Failure Cases
- Insufficient Eviction: If all existing items are considered critical or non-evictable (e.g., system prompts, immutable instructions), and the new item still exceeds the budget, the system must decide whether to reject the new item, truncate it, or raise an error. A robust system will log this event and potentially trigger a human-in-the-loop intervention.
- Policy Misconfiguration: An poorly configured policy might inadvertently evict highly relevant items, leading to degraded agent performance. Monitoring agent coherence and task completion rates helps detect this.
- Performance Degradation: Complex scoring functions or large context windows can introduce latency during eviction. This requires optimization or approximation strategies, especially in high-throughput scenarios.
Architecture
The conceptual architecture for dynamic context eviction centers around a dedicated ContextManager component that orchestrates the storage, retrieval, and eviction of context items. This manager acts as the single source of truth for the agent's active context window.
Execution typically begins with an Agent Orchestrator or Agent Core generating a new piece of information (e.g., a thought, tool call, or response). This new ContextItem (a structured object containing content, token count, timestamp, and initial relevance) is passed to the ContextManager.
The ContextManager maintains an internal Context Window Store, which is an ordered collection of ContextItems. When a new item arrives, the ContextManager first queries the Token Counter to determine the new item's token length and then calculates the total token count if the new item were added. If this exceeds the LLM Token Budget, the ContextManager invokes the Eviction Policy Engine.
The Eviction Policy Engine, based on its configured algorithm (e.g., LRU, Recency-Weighted Relevance, Token-Cost Prioritization), scores the existing items in the Context Window Store. It then instructs the ContextManager to remove the lowest-scoring items until the LLM Token Budget can accommodate the new item. Evicted items might be sent to a Long-Term Memory component for archival or further processing. Finally, the new ContextItem is added to the Context Window Store.
Outputs from the ContextManager include the current, optimized context window (a list of ContextItems) which is then passed to the LLM Adapter for prompt construction and inference. The ContextManager also emits Eviction Events for monitoring and auditing purposes.
Understanding Context Item Scoring for Eviction
Dynamic context eviction algorithms move beyond simple FIFO (First-In, First-Out) or LRU (Least Recently Used) by introducing a scoring mechanism that quantifies the 'value' of each context item. This score determines its priority for retention. Items with lower scores are candidates for eviction. The primary challenge is defining a scoring function that accurately reflects an item's importance to the current agent task.
1. Recency-Weighted Relevance (RWR)
This policy combines the recency of an item with its semantic relevance to the current query or ongoing task. The intuition is that recent items are often more pertinent, but older items might still be highly relevant if they contain core facts or instructions.
Scoring Function: Score(item) = (α * Recency(item)) + (β * Relevance(item))
Recency(item): Can be calculated as1 / (time_since_last_access + 1)ore^(-decay_rate * time_since_last_access). A higher value indicates more recent. Thetime_since_last_accessis typically measured in seconds or agent turns.Relevance(item): This is often the most complex component. It can be derived from:- Semantic Similarity: Using embedding models to calculate cosine similarity between the item's embedding and the current query's embedding.
Relevance(item) = cosine_similarity(item_embedding, current_query_embedding). - Keyword Overlap: A simpler approach, counting shared keywords with the current query.
- LLM-based Scoring: The LLM itself can be prompted to rate the relevance of an item given the current context and task. This is computationally expensive but potentially most accurate.
αandβ: Weighting coefficients (e.g.,α=0.6, β=0.4) that balance the importance of recency versus semantic relevance. These are hyper-parameters tuned for specific agent behaviors.
Items with lower RWR scores are evicted first. This policy is effective for conversational agents where both recent turns and semantically related older information are important.
2. Token-Cost Prioritization
This policy focuses on maximizing the remaining useful token budget by prioritizing the eviction of items that provide the least value per token. It's particularly useful when token costs are a primary concern.
Scoring Function: Score(item) = Value(item) / TokenCount(item)
Value(item): This can be a simplified relevance score (e.g., a binary flag for 'critical' vs. 'non-critical'), or theRelevance(item)component from the RWR policy. If no explicit value is assigned, it can default to1for all items, effectively becoming an 'evict largest first' policy for items of equal perceived value.TokenCount(item): The number of tokens consumed by the item. This is typically pre-calculated upon ingestion.
Items with lower scores (i.e., low value per token) are evicted first. This policy ensures that for a given amount of 'value' removed, the maximum number of tokens are freed. A variant might prioritize evicting items with the highest TokenCount if their Value is below a certain threshold.
3. Hybrid and Adaptive Policies
Production systems often employ hybrid policies that combine aspects of the above. For instance, an agent might always preserve system prompts and a fixed number of recent user/agent turns (a 'protected' zone), then apply RWR or Token-Cost Prioritization to the remaining 'evictable' context. Adaptive policies can dynamically adjust α and β weights based on the agent's current task state or observed performance (e.g., increasing β when the agent is performing a retrieval-heavy task).
Eviction Process Flow
- Identify Eviction Target: Calculate
tokens_to_free = (current_total_tokens + new_item_tokens) - MAX_TOKENS. Iftokens_to_free <= 0, no eviction needed. - Score All Evictable Items: Iterate through all
ContextItems marked as evictable (excluding system prompts, etc.) and calculate their scores using the chosen policy. - Sort for Eviction: Sort items in ascending order of their scores. If scores are equal, secondary criteria like oldest-first or largest-first can be used as tie-breakers.
- Iterative Removal: Start removing items from the sorted list, accumulating
freed_tokens, untilfreed_tokens >= tokens_to_free. If the new item is exceptionally large, it might require freeing more tokens than thetokens_to_freeminimum to accommodate it without immediately triggering another eviction. - Edge Case: Unmet Budget: If all evictable items are removed and
freed_tokensis still less thantokens_to_free, the system is in a critical state. Options include:
- Reject new item: The new item is not added, and an error is returned.
- Truncate new item: The new item's content is truncated to fit the remaining space. This can lead to loss of information in the new item itself.
- Force-evict critical items: A dangerous last resort, potentially removing system prompts or other essential context, leading to agent failure.
- Panic/Error: Halt execution and log a severe error, requiring operator intervention.
Robust implementations typically log eviction decisions, including which items were removed and why, to aid in debugging and policy tuning. This audit trail is crucial for understanding agent behavior and optimizing context management over time.
Code Example
import time
import os
import logging
from collections import deque
from typing import List, Dict, Any, Optional
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class ContextItem:
"""Represents a single piece of context in the agent's memory."""
def __init__(
self,
id: str,
content: str,
token_count: int,
timestamp: float,
relevance_score: float = 0.5,
is_critical: bool = False
):
self.id = id
self.content = content
self.token_count = token_count
self.timestamp = timestamp # Creation time or last access time
self.relevance_score = relevance_score # Semantic relevance to current task
self.is_critical = is_critical # E.g., system prompts, immutable instructions
def __repr__(self):
return f"ContextItem(id='{self.id}', tokens={self.token_count}, ts={self.timestamp:.2f}, rel={self.relevance_score:.2f}, critical={self.is_critical})"
class DynamicContextEvictionManager:
"""Manages the context window using dynamic eviction policies."""
def __init__(
self,
max_tokens: int,
eviction_policy: str = "lru",
recency_weight: float = 0.6,
relevance_weight: float = 0.4
):
self.max_tokens = max_tokens
self.context_items: List[ContextItem] = []
self.current_tokens = 0
self.eviction_policy = eviction_policy.lower()
self.recency_weight = recency_weight
self.relevance_weight = relevance_weight
if not (0 <= recency_weight <= 1 and 0 <= relevance_weight <= 1 and (recency_weight + relevance_weight) == 1):
logging.warning("Recency and relevance weights should sum to 1. Adjusting.")
total_weight = recency_weight + relevance_weight
if total_weight > 0:
self.recency_weight /= total_weight
self.relevance_weight /= total_weight
else:
self.recency_weight = 0.5
self.relevance_weight = 0.5
self._validate_policy()
def _validate_policy(self):
valid_policies = ["lru", "recency_weighted_relevance", "token_cost_prioritization"]
if self.eviction_policy not in valid_policies:
raise ValueError(f"Invalid eviction policy: {self.eviction_policy}. Must be one of {valid_policies}")
def _calculate_recency_score(self, item: ContextItem, current_time: float) -> float:
"""Scores an item based on recency (higher is more recent)."""
time_since_access = current_time - item.timestamp
# Using a simple inverse decay: more recent = higher score
return 1.0 / (time_since_access + 1e-6) # Add small epsilon to avoid division by zero
def _calculate_rwr_score(self, item: ContextItem, current_time: float) -> float:
"""Calculates Recency-Weighted Relevance score."""
recency_score = self._calculate_recency_score(item, current_time)
return (self.recency_weight * recency_score) + (self.relevance_weight * item.relevance_score)
def _calculate_tcp_score(self, item: ContextItem) -> float:
"""Calculates Token-Cost Prioritization score (value per token)."""
# Here, relevance_score acts as 'value'. Higher value, lower token_count is better.
# To make it an eviction score (lower = evict first), we want (low value / high tokens)
# So, we invert: (1 - relevance_score) / token_count.
# Or, more simply, just use 1/score if score is 'goodness'.
# For eviction, we want lower score to be evicted first. So, value/token_count directly.
# If relevance_score is 0, this item has no value. If token_count is 0, this is problematic.
if item.token_count == 0:
return float('-inf') if item.relevance_score > 0 else float('inf') # Don't evict valuable zero-token items
return item.relevance_score / item.token_count
def add_item(self, item: ContextItem) -> bool:
"""Adds a context item, performing eviction if necessary."""
if item.is_critical:
# Critical items are added without eviction logic, assuming they fit or are handled externally
if self.current_tokens + item.token_count > self.max_tokens:
logging.error(f"Critical item '{item.id}' cannot fit. Current: {self.current_tokens}, Item: {item.token_count}, Max: {self.max_tokens}")
return False # Or raise an error
self.context_items.append(item)
self.current_tokens += item.token_count
logging.info(f"Added critical item '{item.id}'. Current tokens: {self.current_tokens}")
return True
tokens_needed = item.token_count
tokens_to_free = (self.current_tokens + tokens_needed) - self.max_tokens
if tokens_to_free > 0:
logging.info(f"Context window full. Current: {self.current_tokens}, Needed: {tokens_needed}, Max: {self.max_tokens}. Evicting {tokens_to_free} tokens.")
freed_tokens = self._evict_items(tokens_to_free)
if freed_tokens < tokens_to_free:
logging.warning(f"Could not free enough tokens. Needed {tokens_to_free}, freed {freed_tokens}. Item '{item.id}' might not fit or will cause overflow.")
# Even if not enough, we proceed to add if possible, or fail.
if self.current_tokens + tokens_needed <= self.max_tokens:
self.context_items.append(item)
self.current_tokens += tokens_needed
logging.info(f"Added item '{item.id}'. Current tokens: {self.current_tokens}")
return True
else:
logging.error(f"Failed to add item '{item.id}'. Not enough space after eviction. Current: {self.current_tokens}, Item: {tokens_needed}, Max: {self.max_tokens}")
return False
def _evict_items(self, tokens_to_free: int) -> int:
"""Evicts items based on policy until enough tokens are freed."""
evictable_items = [item for item in self.context_items if not item.is_critical]
if not evictable_items:
logging.warning("No evictable items available.")
return 0
current_time = time.time()
if self.eviction_policy == "lru":
# For LRU, timestamp is last access. Sort by oldest timestamp first.
evictable_items.sort(key=lambda x: x.timestamp)
elif self.eviction_policy == "recency_weighted_relevance":
# Sort by lowest RWR score first (lowest score = evict first)
evictable_items.sort(key=lambda x: self._calculate_rwr_score(x, current_time))
elif self.eviction_policy == "token_cost_prioritization":
# Sort by lowest TCP score first (lowest score = evict first)
evictable_items.sort(key=self._calculate_tcp_score)
else:
# Fallback to LRU if policy is somehow invalid (should be caught by _validate_policy)
evictable_items.sort(key=lambda x: x.timestamp)
logging.warning(f"Unknown eviction policy '{self.eviction_policy}'. Falling back to LRU.")
freed_tokens = 0
items_to_keep = []
evicted_ids = []
for item in evictable_items:
if freed_tokens >= tokens_to_free:
items_to_keep.append(item)
continue
self.current_tokens -= item.token_count
freed_tokens += item.token_count
evicted_ids.append(item.id)
logging.info(f"Evicted item '{item.id}' (tokens: {item.token_count}). Freed: {freed_tokens}/{tokens_to_free}")
# Reconstruct context_items list with remaining critical items and kept evictable items
critical_items = [item for item in self.context_items if item.is_critical]
self.context_items = critical_items + items_to_keep
# Ensure the list is ordered by timestamp for consistency, or by some other logical order
self.context_items.sort(key=lambda x: x.timestamp)
return freed_tokens
def get_context(self) -> List[ContextItem]:
"""Returns the current list of context items."""
return list(self.context_items)
def get_current_token_count(self) -> int:
"""Returns the current total token count."""
return self.current_tokens
# --- Example Usage ---
if __name__ == "__main__":
# Set max tokens for the context window
MAX_LLM_TOKENS = int(os.environ.get("MAX_LLM_TOKENS", "100")) # Default to 100 for small example
print(f"
--- Testing LRU Eviction (MAX_LLM_TOKENS={MAX_LLM_TOKENS}) ---")
lru_manager = DynamicContextEvictionManager(max_tokens=MAX_LLM_TOKENS, eviction_policy="lru")
lru_manager.add_item(ContextItem("sys_prompt", "You are a helpful assistant.", 10, time.time() - 100, is_critical=True))
lru_manager.add_item(ContextItem("item1", "User asked about weather.", 20, time.time() - 50))
lru_manager.add_item(ContextItem("item2", "Agent replied about weather.", 30, time.time() - 40))
lru_manager.add_item(ContextItem("item3", "User asked about news.", 25, time.time() - 30))
print(f"Current tokens: {lru_manager.get_current_token_count()}, Items: {[i.id for i in lru_manager.get_context()]}")
# This item will trigger eviction
lru_manager.add_item(ContextItem("item4", "Agent replied about news.", 35, time.time() - 20))
print(f"Current tokens: {lru_manager.get_current_token_count()}, Items: {[i.id for i in lru_manager.get_context()]}")
print(f"
--- Testing Recency-Weighted Relevance (MAX_LLM_TOKENS={MAX_LLM_TOKENS}) ---")
rwr_manager = DynamicContextEvictionManager(max_tokens=MAX_LLM_TOKENS, eviction_policy="recency_weighted_relevance", recency_weight=0.7, relevance_weight=0.3)
rwr_manager.add_item(ContextItem("sys_prompt", "You are a helpful assistant.", 10, time.time() - 100, is_critical=True))
rwr_manager.add_item(ContextItem("itemA", "Initial user query about project status.", 20, time.time() - 50, relevance_score=0.2))
rwr_manager.add_item(ContextItem("itemB", "Agent retrieved project documentation.", 30, time.time() - 40, relevance_score=0.8))
rwr_manager.add_item(ContextItem("itemC", "User follow-up on specific task.", 25, time.time() - 30, relevance_score=0.6))
print(f"Current tokens: {rwr_manager.get_current_token_count()}, Items: {[i.id for i in rwr_manager.get_context()]}")
# This item will trigger eviction
rwr_manager.add_item(ContextItem("itemD", "Agent provided task update.", 35, time.time() - 20, relevance_score=0.9))
print(f"Current tokens: {rwr_manager.get_current_token_count()}, Items: {[i.id for i in rwr_manager.get_context()]}")
print(f"
--- Testing Token-Cost Prioritization (MAX_LLM_TOKENS={MAX_LLM_TOKENS}) ---")
tcp_manager = DynamicContextEvictionManager(max_tokens=MAX_LLM_TOKENS, eviction_policy="token_cost_prioritization")
tcp_manager.add_item(ContextItem("sys_prompt", "You are a helpful assistant.", 10, time.time() - 100, is_critical=True))
tcp_manager.add_item(ContextItem("itemX", "Long irrelevant log entry.", 40, time.time() - 50, relevance_score=0.1))
tcp_manager.add_item(ContextItem("itemY", "Key instruction from user.", 15, time.time() - 40, relevance_score=0.9))
tcp_manager.add_item(ContextItem("itemZ", "Another irrelevant message.", 20, time.time() - 30, relevance_score=0.2))
print(f"Current tokens: {tcp_manager.get_current_token_count()}, Items: {[i.id for i in tcp_manager.get_context()]}")
# This item will trigger eviction
tcp_manager.add_item(ContextItem("itemW", "New important user query.", 30, time.time() - 20, relevance_score=0.7))
print(f"Current tokens: {tcp_manager.get_current_token_count()}, Items: {[i.id for i in tcp_manager.get_context()]}")--- Testing LRU Eviction (MAX_LLM_TOKENS=100) --- INFO - Added critical item 'sys_prompt'. Current tokens: 10 INFO - Added item 'item1'. Current tokens: 30 INFO - Added item 'item2'. Current tokens: 60 INFO - Added item 'item3'. Current tokens: 85 Current tokens: 85, Items: ['sys_prompt', 'item1', 'item2', 'item3'] INFO - Context window full. Current: 85, Needed: 35, Max: 100. Evicting 20 tokens. INFO - Evicted item 'item1' (tokens: 20). Freed: 20/20 INFO - Added item 'item4'. Current tokens: 100 Current tokens: 100, Items: ['sys_prompt', 'item2', 'item3', 'item4'] --- Testing Recency-Weighted Relevance (MAX_LLM_TOKENS=100) --- INFO - Added critical item 'sys_prompt'. Current tokens: 10 INFO - Added item 'itemA'. Current tokens: 30 INFO - Added item 'itemB'. Current tokens: 60 INFO - Added item 'itemC'. Current tokens: 85 Current tokens: 85, Items: ['sys_prompt', 'itemA', 'itemB', 'itemC'] INFO - Context window full. Current: 85, Needed: 35, Max: 100. Evicting 20 tokens. INFO - Evicted item 'itemA' (tokens: 20). Freed: 20/20 INFO - Added item 'itemD'. Current tokens: 100 Current tokens: 100, Items: ['sys_prompt', 'itemB', 'itemC', 'itemD'] --- Testing Token-Cost Prioritization (MAX_LLM_TOKENS=100) --- INFO - Added critical item 'sys_prompt'. Current tokens: 10 INFO - Added item 'itemX'. Current tokens: 50 INFO - Added item 'itemY'. Current tokens: 65 INFO - Added item 'itemZ'. Current tokens: 85 Current tokens: 85, Items: ['sys_prompt', 'itemX', 'itemY', 'itemZ'] INFO - Context window full. Current: 85, Needed: 30, Max: 100. Evicting 15 tokens. INFO - Evicted item 'itemZ' (tokens: 20). Freed: 20/15 INFO - Added item 'itemW'. Current tokens: 100 Current tokens: 100, Items: ['sys_prompt', 'itemX', 'itemY', 'itemW']