月成本 = (输入Token数 × 输入单价 + 输出Token数 × 输出单价) × 日均调用量 × 30天

单次调用成本 = (输入Token数 × 输入单价) + (输出Token数 × 输出单价)

实际成本 = (缓存命中文本 × 缓存单价) + (缓存未命中文本 × 输入单价) + (输出Token数 × 输出单价)

不使用缓存：
成本 = (50,000 × $4/1M) + (500 × $4/1M) + (1,000 × $20/1M)
     = $0.20 + $0.002 + $0.02
     = $0.222/次

使用缓存（50K系统提示词命中）：
成本 = (50,000 × $0.40/1M) + (500 × $4/1M) + (1,000 × $20/1M)
     = $0.02 + $0.002 + $0.02
     = $0.042/次

节省比例 = (0.222 - 0.042) / 0.222 = 81%

CPI = 性能评分 / (每百万Token综合成本)

每百万Token综合成本 = 0.8 × 输入单价 + 0.2 × 输出单价

RAG单次成本 = (系统Prompt × 缓存单价) + (检索结果 × 输入单价) + (用户问题 × 输入单价) + (回答 × 输出单价)

开始
  │
  ├─► 任务是否涉及敏感数据/合规要求？
  │     ├─ 是 ──► 国产模型路线
  │     │        ├─ 需要旗舰能力？ ──► Qwen3.7 Max / DeepSeek V4 Pro
  │     │        └─ 普通任务？ ──────► Qwen3.5 Plus / DeepSeek V4-Flash
  │     │
  │     └─ 否 ──► 海外模型路线
  │              │
  │              ├─► 任务复杂度如何？
  │              │     ├─ 极高（复杂推理/高质量代码）
  │              │     │    ├─ 预算充足？ ──► GPT-5.6 Sol（优惠期）
  │              │     │    └─ 追求性价比？ ──► Grok 4.6
  │              │     │
  │              │     ├─ 中等（日常开发/Agent）
  │              │     │    ├─ 依赖OpenAI生态？ ──► GPT-5.6 Terra
  │              │     │    ├─ 编码为主？ ────────► Gemini 3.7 Flash
  │              │     │    └─ 稳定性优先？ ──────► Claude Sonnet 5
  │              │     │
  │              │     └─ 简单（分类/摘要/高吞吐）
  │              │          ├─ OpenAI生态？ ──► GPT-5.6 Luna
  │              │          └─ 极致便宜？ ────► DeepSeek V4-Flash / Qwen3.5 Plus
  │              │
  │              └─► 上下文需求？
  │                    ├─ ≤128K ──► 按复杂度选择（同上）
  │                    ├─ ≤200K ──► 注意Grok的200K阈值定价
  │                    └─ >200K ──► Gemini 3.7 Flash（1M）/ Claude Fable 5（1M）
  │
  └─► 最终验证：用你的真实数据做AB测试，确认效果达标

from typing import Optional, Dict, Any
import tiktoken

class ModelRouter:
    """智能模型路由器 - 根据任务复杂度自动选择最优模型"""
    
    MODEL_TIERS = {
        "tier1_simple": {
            "model": "gpt-5.6-luna",
            "input_price": 0.20,   # $/M tokens
            "output_price": 1.20,
            "max_tokens": 800,
            "description": "简单分类、摘要、常规问答"
        },
        "tier2_standard": {
            "model": "gpt-5.6-terra",
            "input_price": 2.00,
            "output_price": 12.00,
            "max_tokens": 2000,
            "description": "日常代码、中等推理、Agent任务"
        },
        "tier3_complex": {
            "model": "gpt-5.6-sol",  # 优惠期使用sol
            "input_price": 4.00,
            "output_price": 20.00,
            "max_tokens": 4000,
            "description": "复杂推理、高质量代码、多步Agent"
        }
    }
    
    def __init__(self):
        self.encoder = tiktoken.encoding_for_model("gpt-4o")
    
    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))
    
    def assess_complexity(self, prompt: str, task_type: str = "general") -> str:
        """
        评估任务复杂度，返回对应的层级
        可以基于关键词、输入长度、历史准确率等多维度判断
        """
        token_count = self.count_tokens(prompt)
        
        # 基于任务类型的初步判断
        complex_keywords = [
            "优化", "重构", "设计", "架构", "算法", "证明",
            "分析", "debug", "解释为什么", "解决方案",
            "compare", "optimize", "refactor", "architect"
        ]
        
        prompt_lower = prompt.lower()
        complex_score = sum(1 for kw in complex_keywords if kw in prompt_lower)
        
        # 长输入通常意味着更复杂的处理需求
        if token_count > 8000:
            complex_score += 2
        elif token_count > 4000:
            complex_score += 1
        
        # 决策逻辑
        if complex_score >= 3 or token_count > 12000:
            return "tier3_complex"
        elif complex_score >= 1 or token_count > 2000:
            return "tier2_standard"
        else:
            return "tier1_simple"
    
    def route(self, prompt: str, task_type: str = "general") -> Dict[str, Any]:
        """返回路由决策：选择哪个模型及预估成本"""
        tier = self.assess_complexity(prompt, task_type)
        model_info = self.MODEL_TIERS[tier].copy()
        
        # 预估成本（假设输出为输入的1/4）
        input_tokens = self.count_tokens(prompt)
        estimated_output = min(input_tokens // 4, model_info["max_tokens"])
        estimated_cost = (
            input_tokens * model_info["input_price"] / 1_000_000 +
            estimated_output * model_info["output_price"] / 1_000_000
        )
        
        return {
            "tier": tier,
            "model": model_info["model"],
            "input_tokens": input_tokens,
            "estimated_output_tokens": estimated_output,
            "estimated_cost": round(estimated_cost, 4),
            "description": model_info["description"]
        }


# 使用示例
if __name__ == "__main__":
    router = ModelRouter()
    
    # 简单任务
    simple_prompt = "把这句话翻译成英文：你好，世界"
    result = router.route(simple_prompt)
    print(f"简单任务路由结果: {result['model']}, 预估成本: ${result['estimated_cost']}")
    # 输出: gpt-5.6-luna, 预估成本: $0.0001
    
    # 复杂任务
    complex_prompt = """请帮我优化以下微服务架构的设计方案，
    要求：1. 分析当前架构的性能瓶颈 2. 提出三种优化方案并对比
    3. 给出推荐方案的具体实现路径和成本估算
    [此处省略2000字的架构描述...]"""
    result = router.route(complex_prompt, task_type="code")
    print(f"复杂任务路由结果: {result['model']}, 预估成本: ${result['estimated_cost']}")

class FeedbackModelRouter(ModelRouter):
    """带质量反馈的动态路由器"""
    
    def __init__(self):
        super().__init__()
        self.fallback_history = []  # 记录降级/升级的情况
        self.quality_threshold = 0.85  # 质量达标率阈值
    
    def record_result(self, tier: str, quality_score: float, was_upgraded: bool = False):
        """记录调用结果，用于动态调整路由策略"""
        self.fallback_history.append({
            "tier": tier,
            "quality_score": quality_score,
            "was_upgraded": was_upgraded
        })
    
    def should_upgrade(self, current_tier: str, quality_score: float) -> bool:
        """判断是否需要升级到更高层级"""
        if current_tier == "tier3_complex":
            return False  # 已经是最高级
        return quality_score < self.quality_threshold
    
    def get_savings_stats(self) -> Dict[str, float]:
        """计算路由带来的成本节省"""
        # 统计各层级的调用比例
        # 对比全部使用旗舰模型的成本
        ...

import hashlib
import json
from typing import List, Dict

class PromptCacheOptimizer:
    """Prompt缓存优化器 - 最大化缓存命中率"""
    
    def __init__(self):
        # 记录哪些内容应该放在前面以获得缓存
        self.cacheable_prefixes = {}
    
    def optimize_message_order(self, messages: List[Dict]) -> List[Dict]:
        """
        优化消息顺序，将可缓存的内容放在最前面
        注意：这需要根据具体模型的缓存机制调整
        """
        system_msgs = [m for m in messages if m["role"] == "system"]
        other_msgs = [m for m in messages if m["role"] != "system"]
        return system_msgs + other_msgs  # 系统消息放最前面，最大化缓存
    
    def build_cached_system_prompt(
        self,
        base_instructions: str,
        knowledge_base: str,
        dynamic_context: str = None
    ) -> str:
        """
        构建分层的系统提示词
        - 第一层（最前面）：固定不变的指令 → 100%可缓存
        - 第二层：半固定的知识库 → 高缓存率
        - 第三层：动态上下文 → 不可缓存
        """
        parts = []
        
        # 固定指令部分（永远不变，完全缓存）
        parts.append(base_instructions)
        
        # 知识库部分（相对稳定，高缓存率）
        if knowledge_base:
            parts.append(f"

=== 参考知识库 ===
{knowledge_base}")
        
        # 动态上下文（每次变化，不缓存）
        if dynamic_context:
            parts.append(f"

=== 当前上下文 ===
{dynamic_context}")
        
        return "
".join(parts)
    
    def estimate_cache_savings(
        self,
        total_input_tokens: int,
        cached_tokens: int,
        full_price_per_m: float,
        cache_discount_rate: float = 0.1  # 缓存部分按10%计费
    ) -> Dict:
        """
        估算缓存节省的成本
        cache_discount_rate: 缓存单价相对于原价的比例，OpenAI为10%
        """
        full_cost = total_input_tokens * full_price_per_m / 1_000_000
        cached_cost = cached_tokens * full_price_per_m * cache_discount_rate / 1_000_000
        non_cached_cost = (total_input_tokens - cached_tokens) * full_price_per_m / 1_000_000
        actual_cost = cached_cost + non_cached_cost
        
        return {
            "full_cost": round(full_cost, 4),
            "actual_cost": round(actual_cost, 4),
            "savings": round(full_cost - actual_cost, 4),
            "savings_rate": round((1 - actual_cost / full_cost) * 100, 1)
        }


# 使用示例
if __name__ == "__main__":
    optimizer = PromptCacheOptimizer()
    
    # RAG场景：50K知识库 + 1K用户问题
    result = optimizer.estimate_cache_savings(
        total_input_tokens=51000,  # 50K知识库 + 1K用户输入
        cached_tokens=50000,       # 知识库部分可缓存
        full_price_per_m=4.00,     # GPT-5.6 Sol优惠价
        cache_discount_rate=0.5    # 假设50%折扣（实际以官方为准）
    )
    print(f"原价: ${result['full_cost']}")
    print(f"实际: ${result['actual_cost']}")
    print(f"节省: ${result['savings']} ({result['savings_rate']}%)")
    # 输出: 节省约49%

class OutputLengthController:
    """输出长度控制器 - 根据场景动态限制输出Token"""
    
    SCENARIO_CONFIGS = {
        "classification": {
            "max_tokens": 50,
            "system_hint": "请只用一个词回答分类结果，不要添加任何解释。",
            "expected_ratio": 0.05  # 输出/输入比
        },
        "qa_short": {
            "max_tokens": 200,
            "system_hint": "请简洁回答，控制在3句话以内。",
            "expected_ratio": 0.1
        },
        "qa_detailed": {
            "max_tokens": 800,
            "system_hint": "请详细回答，分点说明。",
            "expected_ratio": 0.3
        },
        "code_generation": {
            "max_tokens": 2000,
            "system_hint": "请生成完整可运行的代码，包含必要的注释。",
            "expected_ratio": 0.5
        },
        "creative_writing": {
            "max_tokens": 4000,
            "system_hint": "请展开详细描述，注重内容质量和表达。",
            "expected_ratio": 1.0
        }
    }
    
    def get_config(self, scenario: str) -> Dict:
        return self.SCENARIO_CONFIGS.get(scenario, self.SCENARIO_CONFIGS["qa_short"])
    
    def calculate_output_cost(
        self,
        input_tokens: int,
        scenario: str,
        output_price_per_m: float
    ) -> float:
        """预估输出成本"""
        config = self.get_config(scenario)
        estimated_output = min(
            int(input_tokens * config["expected_ratio"]),
            config["max_tokens"]
        )
        return estimated_output * output_price_per_m / 1_000_000

import numpy as np
from typing import Optional, Tuple

class SemanticCache:
    """语义缓存 - 避免重复调用LLM处理相似问题"""
    
    def __init__(self, embedding_model, similarity_threshold: float = 0.95):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.cache = {}  # {question_embedding.tobytes(): answer}
        self.embeddings = []  # 存储所有embedding用于快速检索
        self.answers = []
    
    def _get_embedding(self, text: str) -> np.ndarray:
        """获取文本的向量表示"""
        return self.embedding_model.encode(text)
    
    def _cosine_similarity(self, v1: np.ndarray, v2: np.ndarray) -> float:
        return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
    
    def get(self, query: str) -> Optional[Tuple[str, float]]:
        """查询缓存，返回(答案, 相似度)或None"""
        if not self.embeddings:
            return None
        
        query_emb = self._get_embedding(query)
        
        # 计算与所有缓存条目的相似度
        similarities = [self._cosine_similarity(query_emb, emb) for emb in self.embeddings]
        max_idx = np.argmax(similarities)
        max_sim = similarities[max_idx]
        
        if max_sim >= self.similarity_threshold:
            return (self.answers[max_idx], max_sim)
        
        return None
    
    def set(self, query: str, answer: str):
        """写入缓存"""
        emb = self._get_embedding(query)
        self.embeddings.append(emb)
        self.answers.append(answer)
    
    def get_stats(self) -> Dict:
        """获取缓存统计"""
        return {
            "cache_size": len(self.answers),
            "threshold": self.similarity_threshold
        }

优化前：$120,000/月
优化后：$120,000 × 50% × 75% × 90% × 85% ≈ $34,425/月
综合节省：约71%