Context Window and Token Complete Guide: LLM Tokenization, Counting Methods, and Cost Optimization

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4")

text = "Hello, world! 你好，世界！"
tokens = enc.encode(text)

print(f"Original text: {text}")
print(f"Token count: {len(tokens)}")
print(f"Token IDs: {tokens}")
print(f"Decoded: {[enc.decode([t]) for t in tokens]}")

Original text: Hello, world! 你好，世界！
Token count: 11
Token IDs: [9906, 11, 1917, 0, 220, 57668, 53901, 3922, 244, 98220, 6447]
Decoded: ['Hello', ',', ' world', '!', ' ', '你好', '，', '世', '界', '！']

def compare_token_efficiency(texts):
    enc = tiktoken.encoding_for_model("gpt-4")
    for text in texts:
        tokens = enc.encode(text)
        chars = len(text)
        ratio = chars / len(tokens)
        print(f"Text: {text}")
        print(f"Characters: {chars}, Tokens: {len(tokens)}, Efficiency: {ratio:.2f} chars/token\n")

compare_token_efficiency([
    "The quick brown fox jumps over the lazy dog.",
    "敏捷的棕色狐狸跳过了懒惰的狗。",
    "Transformer architecture revolutionized NLP.",
    "Transformer架构彻底改变了自然语言处理领域。"
])

def simple_bpe_demo():
    """Simplified BPE demonstration"""
    corpus = ["low", "lower", "newest", "widest"]
    
    vocab = set()
    for word in corpus:
        vocab.update(list(word))
    vocab.add("</w>")
    
    print(f"Initial vocabulary: {sorted(vocab)}")
    
    word_freqs = {}
    for word in corpus:
        chars = list(word) + ["</w>"]
        word_freqs[tuple(chars)] = word_freqs.get(tuple(chars), 0) + 1
    
    print(f"Initial tokenization: {list(word_freqs.keys())}")
    
simple_bpe_demo()

from transformers import BertTokenizer

tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')

text = "unbelievable"
tokens = tokenizer.tokenize(text)
print(f"WordPiece tokenization: {tokens}")

import sentencepiece as spm

sp = spm.SentencePieceProcessor()
sp.load('model.model')

text = "This is a test."
tokens = sp.encode_as_pieces(text)
print(f"SentencePiece tokenization: {tokens}")

def calculate_available_context(total_window, system_prompt_tokens, 
                                 history_tokens, max_output_tokens):
    """Calculate actually available input tokens"""
    available = total_window - system_prompt_tokens - history_tokens - max_output_tokens
    return max(0, available)

total = 128000
system = 500
history = 10000
max_output = 4096

available = calculate_available_context(total, system, history, max_output)
print(f"Total window: {total}")
print(f"System prompt: {system}")
print(f"History: {history}")
print(f"Reserved output: {max_output}")
print(f"Available input: {available}")

import numpy as np

def rope_embedding(x, position, d_model):
    """
    RoPE positional encoding
    x: Input vector
    position: Position index
    d_model: Model dimension
    """
    freqs = 1.0 / (10000 ** (np.arange(0, d_model, 2) / d_model))
    
    angles = position * freqs
    
    cos_vals = np.cos(angles)
    sin_vals = np.sin(angles)
    
    x_even = x[0::2]
    x_odd = x[1::2]
    
    rotated_even = x_even * cos_vals - x_odd * sin_vals
    rotated_odd = x_even * sin_vals + x_odd * cos_vals
    
    result = np.zeros_like(x)
    result[0::2] = rotated_even
    result[1::2] = rotated_odd
    
    return result

def alibi_bias(seq_len, num_heads):
    """
    Calculate ALiBi bias matrix
    """
    slopes = np.array([2 ** (-8 * i / num_heads) for i in range(1, num_heads + 1)])
    
    positions = np.arange(seq_len)
    bias = -np.abs(positions[:, None] - positions[None, :])
    
    alibi = slopes[:, None, None] * bias[None, :, :]
    
    return alibi

import tiktoken

def count_tokens(text, model="gpt-4"):
    """
    Count tokens in text
    """
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    
    return len(encoding.encode(text))

def estimate_cost(input_tokens, output_tokens, model="gpt-4"):
    """
    Estimate API call cost (USD)
    """
    pricing = {
        "gpt-4": {"input": 0.03, "output": 0.06},
        "gpt-4-turbo": {"input": 0.01, "output": 0.03},
        "gpt-4o": {"input": 0.005, "output": 0.015},
        "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
        "claude-3-opus": {"input": 0.015, "output": 0.075},
        "claude-3-sonnet": {"input": 0.003, "output": 0.015},
    }
    
    if model not in pricing:
        return None
    
    input_cost = (input_tokens / 1000) * pricing[model]["input"]
    output_cost = (output_tokens / 1000) * pricing[model]["output"]
    
    return input_cost + output_cost

text = """
Large Language Models (LLMs) are deep learning-based natural language processing models
that learn statistical patterns and semantic knowledge from large-scale text data through pre-training.
"""

tokens = count_tokens(text)
cost = estimate_cost(tokens, 500, "gpt-4o")

print(f"Input tokens: {tokens}")
print(f"Estimated output tokens: 500")
print(f"Estimated cost: ${cost:.4f}")

class TokenCounter:
    """Token counting and cost estimation utility"""
    
    def __init__(self, model="gpt-4"):
        self.model = model
        try:
            self.encoding = tiktoken.encoding_for_model(model)
        except KeyError:
            self.encoding = tiktoken.get_encoding("cl100k_base")
    
    def count(self, text):
        """Count tokens"""
        return len(self.encoding.encode(text))
    
    def count_messages(self, messages):
        """Count tokens in conversation messages"""
        total = 0
        for message in messages:
            total += 4
            for key, value in message.items():
                total += self.count(value)
                if key == "name":
                    total += -1
        total += 2
        return total
    
    def truncate_to_limit(self, text, max_tokens):
        """Truncate text to specified token count"""
        tokens = self.encoding.encode(text)
        if len(tokens) <= max_tokens:
            return text
        return self.encoding.decode(tokens[:max_tokens])
    
    def split_by_tokens(self, text, chunk_size, overlap=0):
        """Split text by token count"""
        tokens = self.encoding.encode(text)
        chunks = []
        start = 0
        while start < len(tokens):
            end = min(start + chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunks.append(self.encoding.decode(chunk_tokens))
            start = end - overlap
        return chunks

counter = TokenCounter("gpt-4")
long_text = "This is a very long text..." * 100
chunks = counter.split_by_tokens(long_text, chunk_size=100, overlap=20)
print(f"Split into {len(chunks)} chunks")

verbose_prompt = """
You are a very helpful AI assistant. Your task is to help users answer various questions.
You should always maintain a friendly and professional attitude. When users ask questions,
you need to carefully analyze the problem and then provide detailed, accurate answers.
If you're not sure about an answer, please honestly tell the user.
"""

concise_prompt = """
You are a professional AI assistant. Provide accurate, concise answers. Be honest when uncertain.
"""

counter = TokenCounter()
print(f"Verbose prompt: {counter.count(verbose_prompt)} tokens")
print(f"Concise prompt: {counter.count(concise_prompt)} tokens")

def manage_conversation_history(messages, max_tokens, counter):
    """
    Manage conversation history within token limits
    Strategy: Keep system messages and most recent conversations
    """
    system_messages = [m for m in messages if m.get("role") == "system"]
    other_messages = [m for m in messages if m.get("role") != "system"]
    
    system_tokens = counter.count_messages(system_messages)
    available_tokens = max_tokens - system_tokens
    
    kept_messages = []
    current_tokens = 0
    
    for message in reversed(other_messages):
        msg_tokens = counter.count_messages([message])
        if current_tokens + msg_tokens <= available_tokens:
            kept_messages.insert(0, message)
            current_tokens += msg_tokens
        else:
            break
    
    return system_messages + kept_messages

def smart_chunk_document(text, max_chunk_tokens=1000, overlap_tokens=100):
    """
    Smart document chunking: split at sentence boundaries
    """
    import re
    
    counter = TokenCounter()
    sentences = re.split(r'(?<=[.!?])\s+', text)
    
    chunks = []
    current_chunk = []
    current_tokens = 0
    
    for sentence in sentences:
        sentence_tokens = counter.count(sentence)
        
        if current_tokens + sentence_tokens > max_chunk_tokens:
            if current_chunk:
                chunks.append(' '.join(current_chunk))
            
            overlap_text = ' '.join(current_chunk[-2:]) if len(current_chunk) >= 2 else ''
            current_chunk = [overlap_text, sentence] if overlap_text else [sentence]
            current_tokens = counter.count(' '.join(current_chunk))
        else:
            current_chunk.append(sentence)
            current_tokens += sentence_tokens
    
    if current_chunk:
        chunks.append(' '.join(current_chunk))
    
    return chunks

def compress_history_with_summary(messages, summarizer_fn, threshold_tokens=2000):
    """
    When history is too long, use summaries to compress earlier conversations
    """
    counter = TokenCounter()
    total_tokens = counter.count_messages(messages)
    
    if total_tokens <= threshold_tokens:
        return messages
    
    system_msgs = [m for m in messages if m["role"] == "system"]
    other_msgs = [m for m in messages if m["role"] != "system"]
    
    split_point = len(other_msgs) // 2
    old_msgs = other_msgs[:split_point]
    recent_msgs = other_msgs[split_point:]
    
    old_text = "\n".join([f"{m['role']}: {m['content']}" for m in old_msgs])
    summary = summarizer_fn(old_text)
    
    summary_msg = {"role": "system", "content": f"Previous conversation summary: {summary}"}
    
    return system_msgs + [summary_msg] + recent_msgs