Fallback, Handoff, Overflow & Flow Strategies

# Define ContextPriority, ContextBlock, ContextBudgetManager; implement __init__, add_block, build_context
import tiktoken
from dataclasses import dataclass
from enum import IntEnum
class ContextPriority(IntEnum):
    """Priority levels for context window content.
    Higher values are evicted last."""
    SYSTEM_PROMPT = 100 # Never evict
    SAFETY_RULES = 95 # Almost never evict
    USER_PROFILE = 80 # High value, compact
    ACTIVE_TASK_STATE = 75 # Critical for current task
    KEY_FACTS = 70 # Important remembered facts
    RETRIEVED_CONTEXT = 60 # RAG results
    RECENT_TURNS = 50 # Last few conversation turns
    SUMMARY = 40 # Compressed conversation history
    OLDER_TURNS = 20 # Older conversation messages
    EXAMPLES = 10 # Few-shot examples (first to go)
@dataclass
class ContextBlock:
    """A block of content in the context window."""
    content: str
    priority: ContextPriority
    token_count: int
    is_evictable: bool = True
    label: str = ""
    class ContextBudgetManager:
        """Manages context window allocation with priority-based eviction."""
        def __init__(self, max_tokens: int = 128000,
            reserve_for_output: int = 4096):
            self.max_tokens = max_tokens - reserve_for_output
            self.encoder = tiktoken.encoding_for_model("gpt-4o")
            self.blocks: list[ContextBlock] = []
            def add_block(self, content: str, priority: ContextPriority,
                label: str = "", evictable: bool = True) -> None:
                """Add a content block to the context."""
                tokens = len(self.encoder.encode(content))
                self.blocks.append(ContextBlock(
                    content=content, priority=priority,
                    token_count=tokens, is_evictable=evictable,
                    label=label
                    ))
    def build_context(self) -> list[dict]:
        """Build the final context, evicting low-priority content if needed."""
        total = sum(b.token_count for b in self.blocks)
        if total <= self.max_tokens:
            # Everything fits
            return self._blocks_to_messages()
            # Need to evict. Sort evictable blocks by priority (ascending)
            evictable = [b for b in self.blocks if b.is_evictable]
            evictable.sort(key=lambda b: b.priority)
            tokens_to_free = total - self.max_tokens
            freed = 0
            evicted_labels = []
            for block in evictable:
                if freed >= tokens_to_free:
                    break
                    self.blocks.remove(block)
                    freed += block.token_count
                    evicted_labels.append(
                        f"{block.label} ({block.token_count} tokens)"
                        )
                    print(f"Evicted {len(evicted_labels)} blocks: "
                        f"{', '.join(evicted_labels)}")
                    return self._blocks_to_messages()
            def get_budget_report(self) -> dict:
                """Report on how the context budget is allocated."""
                total = sum(b.token_count for b in self.blocks)
                by_priority = {}
                for b in self.blocks:
                    name = b.priority.name
                    by_priority[name] = by_priority.get(name, 0) + b.token_count
                    return {
                        "total_tokens": total,
                        "max_tokens": self.max_tokens,
                        "utilization": total / self.max_tokens,
                        "allocation": by_priority,
                        "blocks": len(self.blocks)
                        }
                    def _blocks_to_messages(self) -> list[dict]:
                        """Convert blocks to chat message format."""
                        # Sort by priority (highest first) for message ordering
                        sorted_blocks = sorted(
                            self.blocks, key=lambda b: b.priority, reverse=True
                            )
                        messages = []
                        for block in sorted_blocks:
                            role = "system" if block.priority >= 70 else "user"
                            messages.append({"role": role, "content": block.content})
                            return messages