Token 与成本优化：省钱技巧大全

# 示例
"Hello, world!" → ["Hello", ", world", "!"] → 3 tokens
"你好世界" → ["你好", "世界"] → 2 tokens

总费用 = 输入Token数 × 输入单价 + 输出Token数 × 输出单价

import tiktoken

def count_tokens(text, model="gpt-4"):
    encoder = tiktoken.encoding_for_model(model)
    return len(encoder.encode(text))

# 示例
text = "这是一个测试句子"
print(count_tokens(text))  # 约 6-8 tokens

# ❌ 冗余
prompt = """
你是一个专业的 AI 助手。我需要你帮我处理一些任务。
请你认真阅读以下内容，仔细思考，然后给出准确的回答。
谢谢！

任务：总结以下文章：

{article}
"""

# ✅ 精简
prompt = "总结以下文章：\n{article}"

# ❌ 冗余示例
examples = [
    {
        "input": "用户：你好",
        "output": "客服：您好！有什么可以帮助您的？"
    },
    {
        "input": "用户：我想查询订单",
        "output": "客服：好的，请问您的订单号是多少？"
    },
    {
        "input": "用户：我的订单还没到",
        "output": "客服：抱歉给您带来不便，我帮您查一下物流信息。"
    }
]

# ✅ 精简示例（只保留最能体现格式的1个）
examples = [
    {
        "input": "用户：你好",
        "output": "客服：您好！有什么可以帮助您的？"
    }
]

# ❌ 每次都传完整规则
prompt = """
规则：
1. 用户问候要回复问候
2. 用户询问价格要告知价格
3. 用户投诉要道歉并处理
...

用户：你们产品多少钱？
"""

# ✅ 用引用代替
prompt = """
参考规则文档（见附件）：rule_v2.pdf
用户：你们产品多少钱？
"""

def selective_context(messages, max_tokens=8000):
    """只保留最近的关键消息"""
    encoded = tokenizer.encode(messages)
    
    if len(encoded) <= max_tokens:
        return messages
    
    # 保留系统提示 + 最近消息
    system = messages[0]  # 系统提示
    recent = messages[-max_tokens_tokens:]  # 最近的消息
    
    return [system] + recent

def compress_with_summary(conversation, summary_model):
    """定期压缩对话历史"""
    if len(conversation) > 10:
        # 用小模型生成摘要
        summary = summary_model.invoke(
            f"简要总结以下对话要点：{conversation}"
        )
        return [{"role": "system", "content": "之前对话摘要：" + summary}]
    return conversation

class SlidingWindowBuffer:
    def __init__(self, max_tokens=4000):
        self.max_tokens = max_tokens
        self.messages = []
    
    def add(self, message):
        self.messages.append(message)
        self._trim()
    
    def _trim(self):
        while self.total_tokens() > self.max_tokens:
            # 移除最旧的消息（保留系统提示）
            if len(self.messages) > 1:
                self.messages.pop(1)
    
    def total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

import hashlib
import json
from redis import Redis

class LLMCache:
    def __init__(self, redis_url=None, ttl=3600):
        self.cache = Redis.from_url(redis_url) if redis_url else {}
        self.ttl = ttl
    
    def _hash(self, prompt):
        return hashlib.sha256(prompt.encode()).hexdigest()
    
    def get(self, prompt):
        key = self._hash(prompt)
        return self.cache.get(key)
    
    def set(self, prompt, response):
        key = self._hash(prompt)
        self.cache.setex(key, self.ttl, response)

# 使用
cache = LLMCache(redis_url="redis://localhost:6379")

def cached_completion(prompt):
    # 先检查缓存
    cached = cache.get(prompt)
    if cached:
        return cached
    
    # 没有缓存，调用 API
    response = llm.invoke(prompt)
    
    # 存入缓存
    cache.set(prompt, response)
    return response

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

class SemanticCache:
    def __init__(self, threshold=0.9):
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = Chroma("semantic_cache", self.embeddings)
        self.threshold = threshold
    
    def get(self, prompt):
        prompt_embedding = self.embeddings.embed_query(prompt)
        
        # 搜索相似缓存
        results = self.vectorstore.similarity_search_by_vector(
            prompt_embedding, k=1
        )
        
        if results and results[0].metadata.get("score", 0) > self.threshold:
            return results[0].metadata["response"]
        return None
    
    def set(self, prompt, response, score):
        # 存入向量数据库
        self.vectorstore.add_texts(
            texts=[prompt],
            metadatas=[{"response": response, "score": score}]
        )

class TieredCache:
    """本地内存 + Redis + 数据库 多层缓存"""
    
    def __init__(self):
        self.memory = {}           # L1: 内存，最快
        self.redis = Redis()       # L2: Redis
        self.db = Database()       # L3: 数据库，最慢
    
    def get(self, key):
        # L1 检查
        if key in self.memory:
            return self.memory[key]
        
        # L2 检查
        val = self.redis.get(key)
        if val:
            self.memory[key] = val
            return val
        
        # L3 检查
        val = self.db.query(key)
        if val:
            self.redis.setex(key, 3600, val)
            self.memory[key] = val
            return val
        
        return None

def route_task(task):
    """根据任务复杂度选择模型"""
    
    # 简单任务：小模型即可
    simple_patterns = [
        "翻译",
        "格式转换",
        "简单问答",
        "问候"
    ]
    
    if any(p in task for p in simple_patterns):
        return "gpt-4o-mini"
    
    # 中等任务：标准模型
    medium_patterns = [
        "写代码",
        "分析",
        "总结"
    ]
    
    if any(p in task for p in medium_patterns):
        return "gpt-4o"
    
    # 复杂任务：大模型
    return "gpt-4"

# 使用
model_name = route_task(user_input)
response = llm(model_name).invoke(prompt)

from sklearn.linear_model import LogisticRegression
import numpy as np

class TaskClassifier:
    def __init__(self):
        # 训练好的分类器
        self.classifier = LogisticRegression()
        # 特征提取器
        self.vectorizer = TfidfVectorizer(max_features=100)
    
    def classify(self, text):
        features = self.vectorizer.transform([text])
        return self.classifier.predict(features)[0]
    
    def get_model(self, task_type):
        routing = {
            "translation": "gpt-4o-mini",
            "coding": "gpt-4o",
            "reasoning": "claude-4-sonnet",
            "creative": "gpt-4o",
            "analysis": "claude-4-opus",
        }
        return routing.get(task_type, "gpt-4o")

# 使用
classifier = TaskClassifier()
task_type = classifier.classify(user_input)
model = classifier.get_model(task_type)

# 在 Prompt 中限制
prompt = """
用不超过 100 字回答以下问题：
{question}
"""

# 或用参数限制
response = llm.invoke(
    prompt,
    max_tokens=100  # 限制输出最多 100 tokens
)

# 不用等完整响应，边生成边处理
collected = ""
for chunk in llm.stream(prompt):
    collected += chunk.content
    # 实时处理（如打字机效果、进度显示）
    print(chunk.content, end="", flush=True)

# 用 JSON 模式限制输出格式
response = llm.invoke(
    prompt,
    response_format={"type": "json_object"},
    # 或用 Function Calling 限制
    tools=[{
        "type": "function",
        "function": {
            "name": "extract_info",
            "parameters": {
                "type": "object",
                "properties": {
                    "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
                    "confidence": {"type": "number"}
                }
            }
        }
    }]
)

# ❌ 逐个处理
results = []
for item in items:
    results.append(llm.invoke(process(item)))

# ✅ 批量处理
batch_prompt = "处理以下列表：\n" + "\n".join(items)
results = llm.invoke(batch_prompt)
# 解析结果

import asyncio

async def async_invoke(prompt):
    return await llm.ainvoke(prompt)

async def main():
    # 并发调用
    tasks = [async_invoke(p) for p in prompts]
    results = await asyncio.gather(*tasks)
    return results

class RequestBatcher:
    """合并相似请求，减少 API 调用"""
    
    def __init__(self, batch_size=10, wait_time=0.1):
        self.batch_size = batch_size
        self.wait_time = wait_time
        self.pending = []
    
    async def add(self, prompt):
        self.pending.append(prompt)
        
        if len(self.pending) >= self.batch_size:
            return await self._process_batch()
        
        # 等待一段时间，合并更多请求
        await asyncio.sleep(self.wait_time)
        
        if self.pending:
            return await self._process_batch()
    
    async def _process_batch(self):
        # 批量调用
        combined = "\n---\n".join(self.pending)
        response = await llm.ainvoke(combined)
        
        # 拆分结果
        results = response.split("\n---\n")
        self.pending = []
        return results

class CostTracker:
    def __init__(self):
        self.total_input = 0
        self.total_output = 0
        self.costs = []
    
    def track(self, model, input_tokens, output_tokens):
        prices = {
            "gpt-4o": (2.5, 10.0),
            "gpt-4o-mini": (0.15, 0.6),
            "claude-4-sonnet": (3.0, 15.0),
        }
        
        input_price, output_price = prices.get(model, (0, 0))
        
        input_cost = input_tokens * input_price / 1_000_000
        output_cost = output_tokens * output_price / 1_000_000
        total = input_cost + output_cost
        
        self.total_input += input_tokens
        self.total_output += output_tokens
        self.costs.append(total)
        
        return {
            "input_cost": input_cost,
            "output_cost": output_cost,
            "total_cost": total,
            "cumulative": sum(self.costs)
        }
    
    def get_report(self):
        return {
            "total_input_tokens": self.total_input,
            "total_output_tokens:": self.total_output,
            "total_cost": sum(self.costs),
            "avg_cost_per_call": sum(self.costs) / len(self.costs) if self.costs else 0
        }

class CostAlert:
    def __init__(self, threshold=100):
        self.threshold = threshold
        self.daily_cost = 0
    
    def check(self, cost):
        self.daily_cost += cost
        
        if self.daily_cost > self.threshold:
            send_alert(
                f"今日 API 成本 ${self.daily_cost:.2f} 超过阈值 ${self.threshold}"
            )