Dynamic Context Pruning During Long Agentic Tasks

What Is Dynamic Context Pruning

Dynamic context pruning is the real-time removal of context items during an ongoing agent task based on their current relevance. Unlike static compression (which summarizes all old content uniformly), pruning is selective: it identifies specific items that are no longer useful to the current task phase and removes them individually while leaving the rest intact.

The key insight is that relevance is time-dependent. A database schema retrieved in turn 3 was essential then; by turn 15, the agent has finished the data modeling step and that schema is now consuming tokens without providing value. Pruning recognizes this shift and removes the schema, freeing up budget for the current phase's context needs.

Pruning vs Compression: When to Use Which

• Use pruning when: specific items have become irrelevant but most context is still needed. Pruning is surgical and fast, no LLM required.
• Use compression when: overall context volume is too high regardless of relevance. Compression reduces token count for every item by summarizing.
• Use both in sequence: First prune clearly irrelevant items (fast), then compress remaining content if budget is still insufficient (slow but necessary).
• Pruning latency is near-zero; compression adds 500ms–2s. Prefer pruning in time-sensitive paths.

Detecting Stale Context

Stale context is any information that was relevant at some point but no longer influences the current or near-future reasoning steps. Detecting staleness requires understanding the task's progress and the purpose each context item serves.

Staleness Signals

• Task phase completion: When the agent transitions between phases (research → planning → execution), context from the completed phase becomes a candidate for pruning.
• Superseded information: A tool returned updated data (e.g., refreshed pricing). The previous pricing data is now stale.
• Resolved errors: An error state that the agent successfully recovered from. The error detail is no longer useful for forward progress.
• Age threshold: Items older than N turns that haven't been referenced in subsequent turns. Reference tracking identifies which items were actually used.
• Explicit task completion markers: When the agent outputs a completion signal for a sub-task, its supporting context can be pruned.

Tracking Context Item References

from dataclasses import dataclass, field
from typing import Optional
import re

@dataclass
class TrackedContextItem:
    item_id: str
    content: str
    added_at_turn: int
    item_type: str           # "tool_result", "user_message", "rag_chunk", etc.
    token_count: int
    last_referenced_at: Optional[int] = None
    reference_count: int = 0
    protected: bool = False

    @property
    def turns_since_last_reference(self, current_turn: int) -> int:
        if self.last_referenced_at is None:
            return current_turn - self.added_at_turn
        return current_turn - self.last_referenced_at

class ReferenceTracker:
    def __init__(self):
        self._items: dict[str, TrackedContextItem] = {}

    def add_item(self, item: TrackedContextItem) -> None:
        self._items[item.item_id] = item

    def mark_referenced(self, item_id: str, at_turn: int) -> None:
        if item_id in self._items:
            self._items[item_id].last_referenced_at = at_turn
            self._items[item_id].reference_count += 1

    def scan_response_for_references(
        self, response_text: str, at_turn: int
    ) -> None:
        """Check each tracked item's content against the response."""
        for item in self._items.values():
            # Extract key phrases from item (first 20 chars of sentences)
            key_phrases = re.findall(r'[A-Z][^.!?]{10,40}', item.content)
            for phrase in key_phrases[:5]:
                if phrase[:15] in response_text:
                    self.mark_referenced(item.item_id, at_turn)
                    break

    def get_stale_items(
        self, current_turn: int, staleness_threshold: int = 5
    ) -> list[TrackedContextItem]:
        return [
            item for item in self._items.values()
            if not item.protected
            and item.turns_since_last_reference(current_turn) >= staleness_threshold
        ]

Relevance Scoring for Context Items

Relevance scoring assigns a numeric value to each context item representing how useful it is for the current task state. Items below a threshold become pruning candidates. The scoring function should combine multiple signals, not just recency.

Multi-Factor Relevance Score

import numpy as np
from dataclasses import dataclass

@dataclass
class ScoringWeights:
    recency: float = 0.30
    task_relevance: float = 0.35
    type_weight: float = 0.20
    reference_count: float = 0.15

TYPE_BASE_SCORES = {
    "system_prompt": 1.0,       # never prune
    "task_objective": 1.0,      # never prune
    "active_tool_result": 0.9,
    "user_constraint": 0.9,
    "recent_tool_result": 0.7,
    "rag_chunk": 0.5,
    "intermediate_reasoning": 0.3,
    "background_context": 0.2,
    "filler_message": 0.05,
}

def compute_relevance_score(
    item: TrackedContextItem,
    current_turn: int,
    total_turns: int,
    task_query_embedding: list[float],
    item_embedding: list[float],
    weights: ScoringWeights = ScoringWeights(),
) -> float:
    # Recency: newer = higher score (0–1)
    recency = item.added_at_turn / max(total_turns, 1)

    # Task relevance: cosine similarity to current task query
    q = np.array(task_query_embedding)
    e = np.array(item_embedding)
    task_rel = float(np.dot(q, e) / (np.linalg.norm(q) * np.linalg.norm(e) + 1e-9))

    # Type weight: based on item type
    type_w = TYPE_BASE_SCORES.get(item.item_type, 0.4)

    # Reference count: normalized log scale (0–1)
    ref_norm = min(1.0, np.log1p(item.reference_count) / np.log1p(10))

    return (
        weights.recency * recency
        + weights.task_relevance * task_rel
        + weights.type_weight * type_w
        + weights.reference_count * ref_norm
    )

Task-Phase-Aware Pruning

The most powerful form of dynamic pruning is aware of the agent's current task phase. Different phases have different context requirements. A research phase needs retrieved documents; an execution phase needs plans and constraints. Pruning guided by phase transitions dramatically improves efficiency.

Defining Task Phases

• RESEARCH phase: Retrieve and analyze information. Context needs: retrieved documents, search results, tool outputs. Prunable: previous phase plans, user onboarding messages.
• PLANNING phase: Synthesize research into a plan. Context needs: research summaries, constraints, objectives. Prunable: raw retrieved documents (keep summaries), redundant search results.
• EXECUTION phase: Carry out the plan. Context needs: plan details, execution tool results, constraints. Prunable: most research documents, planning reasoning artifacts.
• VALIDATION phase: Verify outcomes. Context needs: original objectives, execution results, acceptance criteria. Prunable: intermediate execution details already validated.
• COMPLETION phase: Generate final output. Context needs: task objective, key results. Prunable: almost everything except the task specification and final results.

Phase Transition Pruning Hook

from enum import Enum, auto
from typing import Callable

class TaskPhase(Enum):
    RESEARCH = auto()
    PLANNING = auto()
    EXECUTION = auto()
    VALIDATION = auto()
    COMPLETION = auto()

# Which item types to prune when transitioning INTO each phase
PHASE_PRUNE_MAP: dict[TaskPhase, list[str]] = {
    TaskPhase.PLANNING: ["raw_search_result", "duplicate_rag_chunk"],
    TaskPhase.EXECUTION: ["research_doc", "planning_reasoning", "draft_plan"],
    TaskPhase.VALIDATION: ["intermediate_execution", "discarded_approach"],
    TaskPhase.COMPLETION: [
        "research_doc", "planning_reasoning",
        "intermediate_execution", "background_context",
    ],
}

class PhaseAwarePruner:
    def __init__(self, tracker: ReferenceTracker):
        self.tracker = tracker
        self._current_phase = TaskPhase.RESEARCH

    def transition_to(
        self, new_phase: TaskPhase, token_counter: Callable[[str], int]
    ) -> tuple[list[TrackedContextItem], int]:
        """Prune items that are no longer needed in the new phase."""
        pruneable_types = PHASE_PRUNE_MAP.get(new_phase, [])
        pruned = []
        tokens_freed = 0

        for item in list(self.tracker._items.values()):
            if item.item_type in pruneable_types and not item.protected:
                pruned.append(item)
                tokens_freed += item.token_count
                del self.tracker._items[item.item_id]

        self._current_phase = new_phase
        return pruned, tokens_freed

Identifying and Protecting Critical Context

A robust pruning system must identify context that is absolutely critical and mark it as protected from any pruning or compression. Getting this wrong is catastrophic, an agent that loses its task objective mid-task will complete the wrong task.

Critical Context Categories

• Task objective and success criteria: The original user request and what 'done' looks like. Protected forever.
• Hard constraints: Budget limits, prohibited approaches, time constraints, legal requirements. Protected forever.
• In-progress tool calls: Any tool call that has been sent but not yet received a response. Removing the call context breaks the response association.
• User-confirmed decisions: Any decision the user explicitly approved. These represent human-in-the-loop approvals that cannot be silently undone.
• Active plan steps: The current execution plan while the agent is in EXECUTION phase. Pruning the plan mid-execution causes the agent to lose its roadmap.

Runtime Pruning Triggers

Pruning should be triggered proactively before the context is full, not reactively when an overflow occurs. Reactive pruning under time pressure tends to be too aggressive and poorly calibrated.

Trigger Types and Thresholds

• Budget threshold trigger: Fire when context utilization exceeds 70%. Run a light pruning pass to remove clearly stale items. Goal: bring utilization below 60%.
• Turn-count trigger: Fire every N turns (e.g., every 5 turns) regardless of budget. Removes items that have aged out even if budget is healthy.
• Phase transition trigger: Fire when the task transitions to a new phase. Remove items that belong to the completed phase.
• Tool output trigger: Fire after receiving a large tool output (> 500 tokens). Immediately assess whether any existing context is superseded by the new output.
• Explicit signal trigger: The agent explicitly signals that a sub-task is complete. Prune all context items associated with that sub-task.

Recency vs Relevance Tradeoffs

Pure recency-based pruning (remove the oldest items first) is simple but wrong for agentic tasks. A constraint stated in turn 1 is old but critical. A verbose tool output from turn 7 is newer but may be fully redundant with what the agent already knows.

When Recency Dominates

Recency should dominate for conversational content with no task-critical information. Pleasantries, redundant acknowledgments, and verbose intermediate reasoning are safe to prune based on age alone. Set a maximum age for these content types and prune automatically.

When Relevance Dominates

Relevance should dominate for task-critical content: retrieved facts, constraints, and plans. Age alone should never trigger pruning of these items. They should only be pruned when the task phase they belong to is complete or when a more recent piece of information supersedes them explicitly.

Pruning Algorithms in Practice

Three practical pruning algorithms cover most production use cases. Each makes a different tradeoff between simplicity, quality, and compute cost.

Algorithm 1: Threshold-Based Score Pruning

Compute a relevance score for every context item. Remove all items below a threshold score until the budget is satisfied. Simple and fast, but requires accurate scoring. Works best when the scoring function has been calibrated on your specific task domain.

Algorithm 2: Greedy Token-Efficient Pruning

Rank items by (tokens_to_free / relevance_loss) ratio, like a token efficiency score. Prune the item that frees the most tokens for the least relevance sacrifice first. This greedy approach often outperforms pure score thresholding when there's a mix of small high-relevance items and large low-relevance items.

Algorithm 3: Window-Based Structural Pruning

Maintain a fixed-size rolling window of K turns plus a summary of everything before the window. Never score individual items, just roll the window forward each turn. Simplest to implement, most predictable behavior, but less adaptive than score-based approaches.

Testing Pruning Safety

Pruning introduces a new failure mode: the agent loses critical context and makes decisions it wouldn't have made with the full context. Testing pruning safety is as important as testing pruning efficiency.

Safety Test Suite Design

• Constraint preservation test: Set a hard constraint in turn 1 (e.g., 'never use API X'). Run for 20 turns with pruning active. Verify the constraint is still honored in turn 20.
• Objective drift test: State a specific task objective. Run with pruning through multiple phase transitions. Compare the final output against the original objective.
• Factual consistency test: Introduce a specific fact in turn 3. Query the agent about that fact in turn 15. The answer should match the turn-3 fact exactly.
• Recovery from pruning test: Intentionally over-prune (remove 80% of context). Verify the agent detects it's missing context and either requests clarification or fails gracefully rather than hallucinating.

Observability for Pruning Events

Every pruning event is a potential information loss event. Log them with enough detail to diagnose downstream failures caused by over-pruning.

Pruning Event Log Schema

import structlog
from dataclasses import asdict

log = structlog.get_logger()

def log_pruning_event(
    session_id: str,
    turn: int,
    trigger: str,
    pruned_items: list[TrackedContextItem],
    tokens_freed: int,
    budget_before: int,
    budget_after: int,
    total_budget: int,
) -> None:
    log.info(
        "context_pruning_event",
        session_id=session_id,
        turn=turn,
        trigger=trigger,
        items_pruned=len(pruned_items),
        tokens_freed=tokens_freed,
        budget_before=budget_before,
        budget_after=budget_after,
        utilization_before=round(budget_before / total_budget, 3),
        utilization_after=round(budget_after / total_budget, 3),
        pruned_item_types=[i.item_type for i in pruned_items],
        pruned_item_ages=[turn - i.added_at_turn for i in pruned_items],
        pruned_item_reference_counts=[i.reference_count for i in pruned_items],
    )