LLM Token 经济学：分词原理、成本计算与优化实战

# 安装所需库
# pip install tiktoken transformers openai anthropic

import tiktoken
from transformers import AutoTokenizer
import json

class TokenCounter:
    """多模型 Token 计数器"""
    
    def __init__(self):
        self.encoders = {}
        self._init_encoders()
    
    def _init_encoders(self):
        """初始化各模型的 tokenizer"""
        # OpenAI 系列 (tiktoken)
        self.encoders["gpt-4o"] = tiktoken.encoding_for_model("gpt-4o")
        self.encoders["gpt-4"] = tiktoken.encoding_for_model("gpt-4")
        self.encoders["gpt-3.5-turbo"] = tiktoken.encoding_for_model("gpt-3.5-turbo")
        
        # Claude 系列 (通过 Anthropic 官方库)
        try:
            import anthropic
            self.encoders["claude-opus-4.7"] = anthropic.Client()
            self.has_claude = True
        except:
            self.has_claude = False
    
    def count_tiktoken(self, text: str, model: str) -> int:
        """使用 tiktoken 计算 token 数"""
        if model not in self.encoders:
            raise ValueError(f"不支持的模型: {model}")
        return len(self.encoders[model].encode(text))
    
    def count_transformers(self, text: str, model_name: str) -> int:
        """使用 HuggingFace transformers 计算 token 数"""
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        return len(tokenizer.encode(text))
    
    def compare_models(self, text: str) -> dict:
        """对比多个模型的 token 数量"""
        results = {}
        
        tiktoken_models = ["gpt-4o", "gpt-4", "gpt-3.5-turbo"]
        for model in tiktoken_models:
            count = self.count_tiktoken(text, model)
            results[model] = {
                "tokens": count,
                "chars_per_token": len(text) / count if count > 0 else 0
            }
        
        return results
    
    def estimate_cost(self, text: str, model: str, 
                      input_price: float, output_price: float) -> dict:
        """估算 API 调用成本"""
        counter = TokenCounter()
        input_tokens = counter.count_tiktoken(text, model)
        
        # 假设输出 token 数为输入的 50%
        output_tokens = int(input_tokens * 0.5)
        
        input_cost = (input_tokens / 1_000_000) * input_price
        output_cost = (output_tokens / 1_000_000) * output_price
        
        return {
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "input_cost_usd": round(input_cost, 6),
            "output_cost_usd": round(output_cost, 6),
            "total_cost_usd": round(input_cost + output_cost, 6)
        }

# 使用示例
if __name__ == "__main__":
    counter = TokenCounter()
    
    test_text = """
    LLM Token Economics is a critical concept for AI engineers.
    Understanding how different models tokenize text can save 
    thousands of dollars in API costs.
    """
    
    results = counter.compare_models(test_text)
    print(json.dumps(results, indent=2))
    
    # 估算成本
    cost = counter.estimate_cost(
        test_text * 1000,  # 模拟 1000 倍文本
        "gpt-4o",
        input_price=5.0,   # $5/M input tokens
        output_price=15.0  # $15/M output tokens
    )
    print(f"\n估算成本: ${cost['total_cost_usd']:.4f}")

import base64
import anthropic

def estimate_image_tokens(image_path: str, 
                          model: str = "claude-opus-4-7-20260414") -> dict:
    """估算图像输入的实际 token 消耗"""
    client = anthropic.Anthropic()
    
    # 读取图像
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
    
    # 确定媒体类型
    media_type = "image/png" if image_path.endswith(".png") else "image/jpeg"
    
    # 使用 Claude 的 token counting API
    message = client.beta.messages.count_tokens(
        model=model,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": media_type,
                        "data": image_data
                    }
                },
                {
                    "type": "text",
                    "text": "请描述这张图片"
                }
            ]
        }]
    )
    
    return {
        "model": model,
        "input_tokens": message.input_tokens,
        "estimated_cost_usd": round(
            message.input_tokens / 1_000_000 * 5.0 + 
            1000 / 1_000_000 * 25.0,  # 假设输出 1000 tokens
            4
        )
    }

def compare_image_token_costs(image_path: str) -> dict:
    """对比同一图像在不同模型下的 token 消耗"""
    results = {}
    
    models = {
        "claude-opus-4-6-20251022": {"input_price": 5.0, "output_price": 25.0},
        "claude-opus-4-7-20260414": {"input_price": 5.0, "output_price": 25.0},
    }
    
    for model, prices in models.items():
        try:
            token_info = estimate_image_tokens(image_path, model)
            results[model] = token_info
        except Exception as e:
            results[model] = {"error": str(e)}
    
    return results

# 使用示例
# results = compare_image_token_costs("test_image.png")
# for model, info in results.items():
#     print(f"{model}: {info.get('input_tokens', 'N/A')} tokens, "
#           f"${info.get('estimated_cost_usd', 'N/A')}")

def find_cheapest_model(text: str, models: list) -> dict:
    """找出处理给定文本最便宜的模型
    
    Args:
        text: 输入文本
        models: 模型列表，每项包含:
            - name: 模型名称
            - encoder: tiktoken encoder
            - input_price: 输入价格 ($/M tokens)
            - output_price: 输出价格 ($/M tokens)
            - output_ratio: 预计输出/输入 token 比例
    
    Returns:
        最便宜的模型及其成本信息
    """
    results = []
    
    for model in models:
        input_tokens = len(model["encoder"].encode(text))
        output_tokens = int(input_tokens * model.get("output_ratio", 0.5))
        
        input_cost = (input_tokens / 1_000_000) * model["input_price"]
        output_cost = (output_tokens / 1_000_000) * model["output_price"]
        
        results.append({
            "name": model["name"],
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "total_cost_usd": round(input_cost + output_cost, 6),
            "chars_per_token": round(len(text) / input_tokens, 2)
        })
    
    # 按总成本排序
    results.sort(key=lambda x: x["total_cost_usd"])
    return results[0]

# 使用示例
import tiktoken

models = [
    {
        "name": "gpt-4o",
        "encoder": tiktoken.encoding_for_model("gpt-4o"),
        "input_price": 2.50,
        "output_price": 10.00,
        "output_ratio": 0.5
    },
    {
        "name": "gpt-4",
        "encoder": tiktoken.encoding_for_model("gpt-4"),
        "input_price": 10.00,
        "output_price": 30.00,
        "output_ratio": 0.5
    },
]

text = "Your long input text here..." * 100
best = find_cheapest_model(text, models)
print(f"最便宜: {best['name']}, 成本: ${best['total_cost_usd']:.4f}")

import anthropic

def call_claude_with_caching(system_prompt: str, user_message: str) -> str:
    """使用 Claude 系统提示缓存降低重复成本
    
    原理: Claude 会缓存系统提示的 token 表示，
    同一系统提示的重复调用只需支付一次编码成本。
    """
    client = anthropic.Anthropic()
    
    message = client.messages.create(
        model="claude-sonnet-4-6-20260314",
        max_tokens=4096,
        system=[{
            "type": "text",
            "text": system_prompt,
            # 启用缓存锚点
            "cache_control": {"type": "ephemeral"}
        }],
        messages=[{
            "role": "user",
            "content": user_message
        }]
    )
    
    # 检查缓存命中情况
    usage = message.usage
    print(f"输入 tokens: {usage.input_tokens}")
    print(f"缓存读取 tokens: {usage.cache_read_input_tokens}")
    print(f"创建 tokens: {usage.cache_creation_input_tokens}")
    
    return message.content[0].text

# 实际效果：
# 第一次调用：完整编码系统提示
# 后续调用：缓存读取，成本仅为原来的 10%
# 
# 典型场景：
# - 长系统提示（2000+ tokens）+ 高频率调用
# - 月节省可达 $2,000-$5,000

from PIL import Image
import os

def optimize_image_for_llm(image_path: str, 
                          max_long_edge: int = 1024,
                          quality: int = 85) -> str:
    """优化图像尺寸以减少 LLM token 消耗
    
    Args:
        image_path: 原始图像路径
        max_long_edge: 长边最大像素数（Claude 4.7 支持到 2576）
        quality: JPEG 质量 (1-100)
    
    Returns:
        优化后的图像路径
    """
    img = Image.open(image_path)
    w, h = img.size
    long_edge = max(w, h)
    
    if long_edge > max_long_edge:
        scale = max_long_edge / long_edge
        new_w = int(w * scale)
        new_h = int(h * scale)
        img = img.resize((new_w, new_h), Image.LANCZOS)
    
    # 输出路径
    base, ext = os.path.splitext(image_path)
    output_path = f"{base}_optimized.jpg"
    
    # 转为 JPEG 以进一步压缩
    if img.mode in ('RGBA', 'P'):
        img = img.convert('RGB')
    
    img.save(output_path, 'JPEG', quality=quality, optimize=True)
    
    original_size = os.path.getsize(image_path)
    new_size = os.path.getsize(output_path)
    
    print(f"原始: {w}x{h}, {original_size/1024:.1f} KB")
    print(f"优化: {new_w if long_edge > max_long_edge else w}x"
          f"{new_h if long_edge > max_long_edge else h}, "
          f"{new_size/1024:.1f} KB")
    print(f"压缩率: {(1 - new_size/original_size)*100:.1f}%")
    
    return output_path

# 实测效果（基于 Simon Willison 的数据）：
# 3456x2234 PNG (3.7MB) → 优化为 1024x664 JPEG (120KB)
# Token 消耗从 3.01× 降至 ~1.2×
# 成本降低约 60%

import openai
from typing import Generator

def stream_with_budget(client: openai.OpenAI,
                       messages: list,
                       max_tokens: int = 2000,
                       budget_warning: int = 1500) -> Generator[str, None, None]:
    """流式输出 + token 预算控制
    
    当输出接近预算上限时发出警告，避免意外成本。
    """
    token_count = 0
    
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=max_tokens,
        temperature=0.7,
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            token_count += 1
            
            if token_count == budget_warning:
                print(f"\n⚠️ 警告: 已使用 {token_count} tokens，接近预算上限")
            
            yield content
    
    print(f"\n✅ 完成: 总输出 {token_count} tokens")
    # 估算成本
    cost = (token_count / 1_000_000) * 10.0  # GPT-4o 输出价格
    print(f"💰 输出成本: ${cost:.4f}")

# 使用示例
# client = openai.OpenAI()
# messages = [{"role": "user", "content": "写一篇关于 AI 的文章"}]
# for text in stream_with_budget(client, messages, max_tokens=2000):
#     print(text, end="", flush=True)

class ModelRouter:
    """智能模型路由器 —— 根据任务复杂度选择最优模型"""
    
    ROUTING_RULES = [
        {
            "name": "simple_classification",
            "condition": lambda t: len(t.split()) < 50,
            "model": "claude-haiku-4-5",
            "reason": "短文本分类不需要强大推理"
        },
        {
            "name": "summarization",
            "condition": lambda t: "summarize" in t.lower() or "总结" in t,
            "model": "gpt-4o-mini",
            "reason": "摘要任务中等模型即可"
        },
        {
            "name": "code_generation",
            "condition": lambda t: any(k in t.lower() 
                                       for k in ["code", "function", "实现", "写代码"]),
            "model": "claude-sonnet-4-6",
            "reason": "代码生成需要较强推理"
        },
        {
            "name": "complex_analysis",
            "condition": lambda t: len(t.split()) > 500 or 
                                   any(k in t.lower() 
                                       for k in ["analyze", "分析", "compare", "对比"]),
            "model": "claude-opus-4-7",
            "reason": "复杂分析需要最强模型"
        },
        {
            "name": "default",
            "condition": lambda t: True,
            "model": "claude-sonnet-4-6",
            "reason": "默认使用性价比最优模型"
        }
    ]
    
    @classmethod
    def route(cls, task_description: str) -> dict:
        """根据任务描述路由到最优模型"""
        for rule in cls.ROUTING_RULES:
            if rule["condition"](task_description):
                return {
                    "model": rule["model"],
                    "rule": rule["name"],
                    "reason": rule["reason"]
                }
        return cls.ROUTING_RULES[-1]  # fallback

# 使用示例
# router = ModelRouter()
# result = router.route("请分析这份财报的关键财务指标变化")
# print(f"路由到: {result['model']} ({result['reason']})")
# → 路由到: claude-opus-4-7 (复杂分析需要最强模型)
#
# result = router.route("这是正面还是负面评价？")
# print(f"路由到: {result['model']} ({result['reason']})")
# → 路由到: claude-haiku-4-5 (短文本分类不需要强大推理)
#
# 实际效果：月度 API 成本降低 35%-50%