AI成本优化2026实战

class ModelTierRouter:
    """根据任务复杂度路由到不同级别的模型"""
    
    def __init__(self):
        self.tiers = {
            "simple": {"model": "qwen3-7b", "cost_per_1k": 0.0005},
            "medium": {"model": "qwen3-32b", "cost_per_1k": 0.002},
            "complex": {"model": "qwen3-72b", "cost_per_1k": 0.008},
        }
    
    def route(self, query, context=None):
        # 简单规则路由
        if len(query) < 50 and "?" in query:
            return self.tiers["simple"]
        
        # 基于意图分类
        complexity = self.estimate_complexity(query, context)
        
        if complexity < 0.3:
            return self.tiers["simple"]
        elif complexity < 0.7:
            return self.tiers["medium"]
        else:
            return self.tiers["complex"]
    
    def estimate_complexity(self, query, context):
        """估计查询复杂度"""
        score = 0
        # 长查询更复杂
        score += min(len(query) / 1000, 0.3)
        # 需要推理的关键词
        reasoning_words = ["分析", "比较", "为什么", "推导", "计算"]
        score += 0.2 * any(w in query for w in reasoning_words)
        # 多轮上下文更复杂
        if context and len(context) > 5:
            score += 0.2
        # 代码生成更复杂
        if "代码" in query or "函数" in query:
            score += 0.3
        return min(score, 1.0)

class SemanticCache:
    """语义缓存：相似查询命中缓存"""
    
    def __init__(self, vector_store, similarity_threshold=0.95):
        self.store = vector_store
        self.threshold = similarity_threshold
    
    async def get(self, query):
        """查询缓存"""
        query_embedding = await self.embed(query)
        
        results = await self.store.search(
            query_embedding, top_k=1
        )
        
        if results and results[0].score > self.threshold:
            # 缓存命中
            return json.loads(results[0].document)
        
        return None
    
    async def set(self, query, response, ttl=3600):
        """写入缓存"""
        embedding = await self.embed(query)
        await self.store.add(
            id=generate_uuid(),
            embedding=embedding,
            document=json.dumps({
                "query": query,
                "response": response,
                "timestamp": time.time(),
            }),
            ttl=ttl
        )

class TokenOptimizer:
    """减少token消耗的各种策略"""
    
    def compress_history(self, messages):
        """压缩对话历史"""
        if len(messages) <= 4:
            return messages
        
        # 保留最近2轮 + 摘要
        recent = messages[-4:]
        old = messages[:-4]
        summary = self.summarize(old)
        
        return [{"role": "system", "content": f"历史摘要：{summary}"}] + recent
    
    def optimize_prompt(self, prompt):
        """精简提示词"""
        # 移除冗余空格和换行
        prompt = re.sub(r'\n{3,}', '\n\n', prompt)
        prompt = re.sub(r' {2,}', ' ', prompt)
        return prompt.strip()
    
    def truncate_context(self, documents, max_tokens=2000):
        """智能截断文档"""
        result = []
        current_tokens = 0
        
        for doc in documents:
            doc_tokens = len(doc) // 4
            if current_tokens + doc_tokens > max_tokens:
                # 截断最后一个文档而非完全丢弃
                remaining = max_tokens - current_tokens
                if remaining > 100:
                    result.append(doc[:remaining * 4])
                break
            result.append(doc)
            current_tokens += doc_tokens
        
        return result

class BatchProcessor:
    """合并多个请求降低API调用次数"""
    
    async def batch_generate(self, requests):
        """将多个独立请求合并为一个"""
        # 合并多个查询为一个prompt
        combined_prompt = "请依次回答以下问题：\n\n"
        for i, req in enumerate(requests):
            combined_prompt += f"问题{i+1}：{req['query']}\n"
        combined_prompt += "\n请按问题编号分别回答。"
        
        response = await self.llm.generate(combined_prompt)
        
        # 解析响应
        answers = self.parse_batch_response(response, len(requests))
        return answers

class CostMonitor:
    def __init__(self):
        self.daily_costs = defaultdict(float)
        self.user_costs = defaultdict(float)
        self.model_costs = defaultdict(float)
    
    def record(self, user_id, model, input_tokens, output_tokens):
        pricing = {
            "qwen3-7b": {"input": 0.0003, "output": 0.0006},
            "qwen3-32b": {"input": 0.001, "output": 0.002},
            "qwen3-72b": {"input": 0.004, "output": 0.008},
        }
        
        rates = pricing.get(model, {"input": 0.001, "output": 0.002})
        cost = (input_tokens / 1000 * rates["input"] + 
                output_tokens / 1000 * rates["output"])
        
        today = datetime.now().strftime("%Y-%m-%d")
        self.daily_costs[today] += cost
        self.user_costs[user_id] += cost
        self.model_costs[model] += cost
    
    def alert_if_over_budget(self, daily_budget=100):
        today = datetime.now().strftime("%Y-%m-%d")
        if self.daily_costs[today] > daily_budget:
            self.send_alert(
                f"日成本超预算：${self.daily_costs[today]:.2f} / ${daily_budget}"
            )