上下文窗口与Token完全指南：LLM分词原理、计数方法与成本优化策略

import tiktoken

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

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

print(f"原始文本: {text}")
print(f"Token数量: {len(tokens)}")
print(f"Token ID: {tokens}")
print(f"解码后: {[enc.decode([t]) for t in tokens]}")

原始文本: Hello, world! 你好，世界！
Token数量: 11
Token ID: [9906, 11, 1917, 0, 220, 57668, 53901, 3922, 244, 98220, 6447]
解码后: ['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}")
        print(f"字符数: {chars}, Token数: {len(tokens)}, 效率: {ratio:.2f}字符/Token\n")

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

def simple_bpe_demo():
    """简化的BPE演示"""
    corpus = ["low", "lower", "newest", "widest"]
    
    vocab = set()
    for word in corpus:
        vocab.update(list(word))
    vocab.add("</w>")
    
    print(f"初始词汇表: {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"初始分词: {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分词: {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分词: {tokens}")

def calculate_available_context(total_window, system_prompt_tokens, 
                                 history_tokens, max_output_tokens):
    """计算实际可用的输入Token数"""
    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}")
print(f"系统提示: {system}")
print(f"历史记录: {history}")
print(f"预留输出: {max_output}")
print(f"可用输入: {available}")

import numpy as np

def rope_embedding(x, position, d_model):
    """
    RoPE位置编码
    x: 输入向量
    position: 位置索引
    d_model: 模型维度
    """
    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):
    """
    计算ALiBi偏置矩阵
    """
    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"):
    """
    计算文本的Token数量
    """
    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"):
    """
    估算API调用成本（美元）
    """
    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 Model，LLM）是一种基于深度学习的自然语言处理模型，
通过在大规模文本数据上进行预训练，学习语言的统计规律和语义知识。
"""

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

print(f"输入Token数: {tokens}")
print(f"预估输出Token: 500")
print(f"预估成本: ${cost:.4f}")

class TokenCounter:
    """Token计数和成本估算工具"""
    
    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):
        """计算Token数量"""
        return len(self.encoding.encode(text))
    
    def count_messages(self, messages):
        """计算对话消息的Token数量"""
        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):
        """截断文本到指定Token数量"""
        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):
        """按Token数量分割文本"""
        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 = "这是一段很长的文本..." * 100
chunks = counter.split_by_tokens(long_text, chunk_size=100, overlap=20)
print(f"分割成 {len(chunks)} 个块")

verbose_prompt = """
你是一个非常有帮助的AI助手。你的任务是帮助用户解答各种问题。
你应该始终保持友好、专业的态度。当用户提问时，你需要仔细分析问题，
然后给出详细、准确的回答。如果你不确定答案，请诚实地告诉用户。
"""

concise_prompt = """
你是专业的AI助手。提供准确、简洁的回答。不确定时如实说明。
"""

counter = TokenCounter()
print(f"冗长提示: {counter.count(verbose_prompt)} tokens")
print(f"精简提示: {counter.count(concise_prompt)} tokens")

def manage_conversation_history(messages, max_tokens, counter):
    """
    管理对话历史，保持在Token限制内
    策略：保留系统消息和最近的对话
    """
    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):
    """
    智能文档分块：在句子边界处分割
    """
    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):
    """
    当历史过长时，使用摘要压缩早期对话
    """
    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"之前对话摘要：{summary}"}
    
    return system_msgs + [summary_msg] + recent_msgs