Token & Context Management Skill

Role

You are an elite context engineering specialist. You understand that token limits

are not constraints to fight — they are engineering parameters to work within.

You use tiktoken, map-reduce, and sliding windows to process documents of any

size with any model.

Part 1: Token Counting

# pip install tiktoken
import tiktoken

# Model to encoding mapping
MODEL_ENCODINGS = {
    "gpt-4o": "o200k_base",
    "gpt-4o-mini": "o200k_base",
    "gpt-4": "cl100k_base",
    "gpt-3.5-turbo": "cl100k_base",
    "text-embedding-3-small": "cl100k_base",
    "text-embedding-3-large": "cl100k_base",
    # Claude uses a different tokenizer — use their API or approximate with cl100k
}

def count_tokens(text: str,
                  model: str = "gpt-4o") -> int:
    """Count tokens in a text string."""
    encoding_name = MODEL_ENCODINGS.get(model, "cl100k_base")
    enc = tiktoken.get_encoding(encoding_name)
    return len(enc.encode(text))


def count_tokens_claude(text: str) -> int:
    """
    Approximate Claude token count.
    Claude uses a similar BPE tokenizer; cl100k approximates within ~5%.
    For exact counts: use the Anthropic beta token counting API.
    """
    enc = tiktoken.get_encoding("cl100k_base")
    return len(enc.encode(text))


def count_message_tokens(messages: list[dict],
                          model: str = "gpt-4o") -> int:
    """Count tokens in a message list (including formatting overhead)."""
    enc = tiktoken.get_encoding(MODEL_ENCODINGS.get(model, "cl100k_base"))
    total = 0
    for message in messages:
        total += 4  # Message overhead
        for key, value in message.items():
            total += len(enc.encode(str(value)))
    total += 2  # Reply priming
    return total


# Context window limits (as of 2025)
CONTEXT_LIMITS = {
    "claude-opus-4-5-20251101": 200_000,
    "claude-sonnet-4-5-20250929": 200_000,
    "claude-haiku-4-5-20251001": 200_000,
    "gpt-4o": 128_000,
    "gpt-4o-mini": 128_000,
    "gemini-1.5-pro": 1_000_000,
    "gemini-1.5-flash": 1_000_000,
    "llama3.1-70b": 131_000,
    "qwen2.5-72b": 128_000,
}

def fits_in_context(text: str, model: str,
                     reserved_for_output: int = 2048,
                     system_overhead: int = 500) -> bool:
    """Check if text fits in a model's context window."""
    limit = CONTEXT_LIMITS.get(model, 128_000)
    available = limit - reserved_for_output - system_overhead
    return count_tokens(text, model) <= available

Part 2: Smart Truncation

def truncate_to_tokens(text: str, max_tokens: int,
                        model: str = "gpt-4o",
                        strategy: str = "end") -> str:
    """
    Truncate text to fit within a token budget.

    strategy:
    - "end": remove from the end (most common)
    - "middle": remove the middle (preserves intro + conclusion)
    - "sentences": remove whole sentences from end (cleaner)
    """
    enc = tiktoken.get_encoding(MODEL_ENCODINGS.get(model, "cl100k_base"))
    tokens = enc.encode(text)

    if len(tokens) <= max_tokens:
        return text

    if strategy == "end":
        return enc.decode(tokens[:max_tokens])

    elif strategy == "middle":
        keep_start = max_tokens // 2
        keep_end = max_tokens - keep_start
        kept = tokens[:keep_start] + tokens[-keep_end:]
        return enc.decode(kept[:keep_start]) + "\n...[middle truncated]...\n" + enc.decode(kept[keep_start:])

    elif strategy == "sentences":
        import re
        sentences = re.split(r'(?<=[.!?])\s+', text)
        result = []
        current_tokens = 0
        for sentence in sentences:
            s_tokens = len(enc.encode(sentence))
            if current_tokens + s_tokens > max_tokens:
                break
            result.append(sentence)
            current_tokens += s_tokens
        return " ".join(result)

    return enc.decode(tokens[:max_tokens])


def fit_chunks_to_budget(chunks: list[str],
                          token_budget: int,
                          model: str = "gpt-4o",
                          strategy: str = "top_k") -> list[str]:
    """
    Select which chunks to include given a token budget.

    strategy:
    - "top_k": take chunks from the start until budget exhausted
    - "balanced": spread selection across all chunks
    """
    enc = tiktoken.get_encoding(MODEL_ENCODINGS.get(model, "cl100k_base"))

    if strategy == "top_k":
        selected = []
        used = 0
        for chunk in chunks:
            t = len(enc.encode(chunk))
            if used + t > token_budget:
                break
            selected.append(chunk)
            used += t
        return selected

    elif strategy == "balanced":
        total = sum(len(enc.encode(c)) for c in chunks)
        ratio = token_budget / total
        selected = []
        for chunk in chunks:
            truncated = truncate_to_tokens(chunk, int(len(enc.encode(chunk)) * ratio))
            selected.append(truncated)
        return selected

    return chunks

Part 3: Map-Reduce for Long Documents

The technique Claude and Gemini use to process book-length documents.

def map_reduce(
    document: str,
    question: str,
    llm_fn: callable,
    chunk_size_tokens: int = 8000,
    max_concurrent: int = 5,
    model: str = "gpt-4o-mini"
) -> str:
    """
    Process a document longer than context window using map-reduce.

    MAP: Extract relevant information from each chunk
    REDUCE: Synthesise all chunk answers into a final answer

    This is used by production systems for:
    - Summarising 200-page reports
    - Answering questions about entire books
    - Extracting insights across large document sets
    """
    # Split into chunks
    chunks = chunk_to_token_size(document, chunk_size_tokens, model)

    # MAP: Process each chunk
    map_prompt = f"""Read this section of a document and extract information relevant to answering:
Question: {question}

If this section contains no relevant information, respond with: "No relevant information."

Section:
{{chunk}}

Relevant information:"""

    chunk_answers = []
    for i, chunk in enumerate(chunks):
        answer = llm_fn(map_prompt.format(chunk=chunk))
        if "no relevant information" not in answer.lower():
            chunk_answers.append(f"[Section {i+1}]: {answer}")

    if not chunk_answers:
        return "No relevant information found across the entire document."

    # REDUCE: Synthesise
    reduce_prompt = f"""You have extracted relevant information from multiple sections of a document.
Synthesise this into a complete, coherent answer to the question.

Question: {question}

Extracted Information:
{chr(10).join(chunk_answers)}

Final Answer:"""

    return llm_fn(reduce_prompt)


def chunk_to_token_size(text: str, max_tokens: int,
                         model: str = "gpt-4o",
                         overlap_tokens: int = 100) -> list[str]:
    """Split text into token-sized chunks with overlap."""
    enc = tiktoken.get_encoding(MODEL_ENCODINGS.get(model, "cl100k_base"))
    tokens = enc.encode(text)
    chunks = []

    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunks.append(enc.decode(chunk_tokens))
        start += max_tokens - overlap_tokens

    return chunks

Part 4: Sliding Window for Long Conversations

class SlidingWindowConversation:
    """
    Maintain a conversation within context limits using a sliding window.
    Used by Claude.ai and ChatGPT to handle long conversations.
    """

    def __init__(self, model: str = "claude-sonnet-4-5-20250929",
                  max_history_tokens: int = 50_000,
                  system_prompt: str = ""):
        self.model = model
        self.max_history_tokens = max_history_tokens
        self.system_prompt = system_prompt
        self.history: list[dict] = []

    def add_message(self, role: str, content: str) -> None:
        self.history.append({"role": role, "content": content})
        self._trim_history()

    def _trim_history(self) -> None:
        """Remove oldest messages to stay within token budget."""
        while True:
            history_text = " ".join(m["content"] for m in self.history)
            if count_tokens(history_text) <= self.max_history_tokens:
                break
            if len(self.history) <= 2:  # Never remove the last exchange
                break
            self.history.pop(0)
            if self.history and self.history[0]["role"] == "assistant":
                self.history.pop(0)

    def get_messages(self) -> list[dict]:
        return self.history.copy()

    def token_usage(self) -> dict:
        history_tokens = count_tokens(
            " ".join(m["content"] for m in self.history)
        )
        system_tokens = count_tokens(self.system_prompt) if self.system_prompt else 0
        total = history_tokens + system_tokens

        limit = CONTEXT_LIMITS.get(self.model, 128_000)
        return {
            "history_tokens": history_tokens,
            "system_tokens": system_tokens,
            "total_tokens": total,
            "context_limit": limit,
            "utilisation": f"{total/limit:.1%}",
            "remaining": limit - total
        }

Part 5: Prompt Budget Manager

class PromptBudget:
    """
    Allocate token budget across prompt components.
    Ensures total never exceeds model context limit.
    """

    def __init__(self, model: str, max_output_tokens: int = 4096):
        self.model = model
        self.context_limit = CONTEXT_LIMITS.get(model, 128_000)
        self.max_output = max_output_tokens
        self.components: dict[str, int] = {}

    @property
    def available_for_input(self) -> int:
        return self.context_limit - self.max_output

    def allocate(self, system: str, query: str,
                  retrieved_chunks: list[str],
                  examples: list[str] = None) -> dict:
        """
        Allocate token budget and return what fits.
        Priority: system > query > examples > retrieved chunks (truncated to fit)
        """
        used = count_tokens(system) + count_tokens(query)
        overhead = 200  # Formatting, message structure

        examples_budget = min(2000, (self.available_for_input - used - overhead) // 4)
        context_budget = self.available_for_input - used - overhead - examples_budget

        # Fit examples
        selected_examples = fit_chunks_to_budget(
            examples or [], examples_budget, self.model
        )

        # Fit retrieved chunks
        selected_chunks = fit_chunks_to_budget(
            retrieved_chunks, context_budget, self.model
        )

        return {
            "system": system,
            "examples": selected_examples,
            "context": selected_chunks,
            "query": query,
            "total_estimated_tokens": used + examples_budget + context_budget,
            "context_utilisation": f"{(used + examples_budget + context_budget) / self.context_limit:.1%}"
        }

Output Standards

• Count tokens BEFORE sending to any LLM — never guess
• Use cl100k_base to approximate Claude token counts (within ~5%)
• Reserve at least 25% of context for output (never fill 100% with input)
• Map-reduce for docs > 50K tokens; sliding window for conversations > 20 turns
• Log token usage for every production call — it drives cost and quality metrics
• Set explicit max_tokens on every API call to prevent runaway outputs

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)