Token 计算与优化完整指南

import tiktoken

class TokenCounter:
    def __init__(self):
        self.encoders = {
            "gpt-4": tiktoken.encoding_for_model("gpt-4"),
            "gpt-3.5-turbo": tiktoken.encoding_for_model("gpt-3.5-turbo")
        }
        
    def count_tokens(self, text: str, model: str = "gpt-4") -> int:
        encoding = self.encoders.get(model, tiktoken.get_encoding("cl100k_base"))
        return len(encoding.encode(text))
    
    def count_messages_tokens(self, messages: list, model: str = "gpt-4") -> int:
        encoding = self.encoders.get(model)
        tokens_per_message = 3
        num_tokens = 0
        
        for message in messages:
            num_tokens += tokens_per_message
            for key, value in message.items():
                num_tokens += len(encoding.encode(value))
        
        num_tokens += 3  # 结束标记
        return num_tokens

# 使用示例
counter = TokenCounter()
text = "Hello, how can I help you?"
tokens = counter.count_tokens(text)
print(f"Token数: {tokens}")

# 成本计算
def estimate_cost(input_tokens: int, output_tokens: int, model: str = "gpt-4o"):
    pricing = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006}
    }
    cost = (input_tokens * pricing[model]["input"] + 
            output_tokens * pricing[model]["output"]) / 1000
    return cost

class PromptOptimizer:
    def compress_prompt(self, prompt: str, max_tokens: int = 1000) -> str:
        # 1. 移除冗余词汇
        redundant_words = ["please", "could you", "I would like"]
        for word in redundant_words:
            prompt = prompt.replace(word, "")
        
        # 2. 使用缩写
        abbreviations = {
            "for example": "e.g.",
            "that is": "i.e.",
            "et cetera": "etc."
        }
        for full, abbr in abbreviations.items():
            prompt = prompt.replace(full, abbr)
        
        # 3. 精简指令
        if "请详细分析" in prompt:
            prompt = prompt.replace("请详细分析", "分析")
        
        return prompt.strip()

# 优化示例
original = "Please could you analyze this text in detail"
optimized = PromptOptimizer().compress_prompt(original)
# 结果: "analyze this text"

import hashlib
from datetime import datetime, timedelta

class TokenCache:
    def __init__(self):
        self.cache = {}
        self.stats = {"hits": 0, "misses": 0, "tokens_saved": 0}
    
    def get_key(self, prompt: str, model: str) -> str:
        content = f"{prompt}:{model}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, prompt: str, model: str):
        key = self.get_key(prompt, model)
        if key in self.cache:
            entry = self.cache[key]
            if datetime.now() < entry["expires"]:
                self.stats["hits"] += 1
                self.stats["tokens_saved"] += entry["tokens"]
                return entry["response"]
        
        self.stats["misses"] += 1
        return None
    
    def set(self, prompt: str, model: str, response: str, tokens: int):
        key = self.get_key(prompt, model)
        self.cache[key] = {
            "response": response,
            "tokens": tokens,
            "expires": datetime.now() + timedelta(hours=24)
        }

def smart_chunk_text(text: str, max_tokens: int = 1500):
    """智能文本分块"""
    sentences = text.split('.')
    chunks = []
    current_chunk = []
    current_tokens = 0
    
    for sentence in sentences:
        sentence_tokens = count_tokens(sentence + '.')
        
        if current_tokens + sentence_tokens > max_tokens:
            chunks.append('.'.join(current_chunk) + '.')
            current_chunk = [sentence]
            current_tokens = sentence_tokens
        else:
            current_chunk.append(sentence)
            current_tokens += sentence_tokens
    
    if current_chunk:
        chunks.append('.'.join(current_chunk) + '.')
    
    return chunks

# 实际应用：文档总结优化
def optimized_document_summary(document: str):
    # 1. 检查缓存
    cached = cache.get(document[:100], "summary")
    if cached:
        return cached
    
    # 2. 计算token
    tokens = counter.count_tokens(document)
    
    if tokens < 2000:
        # 直接处理
        summary = process_with_api(document)
    else:
        # 分块处理
        chunks = smart_chunk_text(document)
        summaries = []
        
        for chunk in chunks:
            # 优化每块的提示词
            prompt = f"总结: {chunk}"
            optimized = optimizer.compress_prompt(prompt)
            summaries.append(process_with_api(optimized))
        
        # 合并总结
        summary = merge_summaries(summaries)
    
    # 3. 缓存结果
    cache.set(document[:100], "summary", summary, tokens)
    return summary