LLM 成本优化深度指南：省下 80% API 费用

┌──────────────────────────────────────────────┐
│              总 API 成本                      │
│  ┌──────────────────────────────────────┐    │
│  │  Token 优化 (减少输入)                │    │
│  │  - Prompt 压缩                       │    │
│  │  - 历史截断                          │    │
│  │  - Few-shot 精简                     │    │
│  └──────────────────────────────────────┘    │
│  ┌──────────────────────────────────────┐    │
│  │  模型路由 (降低单价)                  │    │
│  │  - 简单 → 小模型 ($0.15/1M)         │    │
│  │  - 复杂 → 大模型 ($10/1M)           │    │
│  └──────────────────────────────────────┘    │
│  ┌──────────────────────────────────────┐    │
│  │  架构优化 (减少调用)                  │    │
│  │  - 缓存 (命中率 60%)                 │    │
│  │  - 批量推理                          │    │
│  │  - 流式输出提前终止                   │    │
│  └──────────────────────────────────────┘    │
└──────────────────────────────────────────────┘

class PromptCompressor:
    """多层 Prompt 压缩"""

    def compress(self, system_prompt: str, history: list, query: str) -> str:
        # 1. 系统提示词精简
        system = self._trim_system(system_prompt)
        # 2. 历史对话压缩
        history = self._compress_history(history)
        # 3. Few-shot 示例选择（只保留最相关的）
        examples = self._select_relevant_examples(query, k=2)
        # 4. 组装
        return f"{system}\n\n{examples}\n\n历史：{history}\n\n问题：{query}"

    def _trim_system(self, prompt: str) -> str:
        """去除冗余描述，保留核心指令"""
        # 去除重复空格和换行
        import re
        prompt = re.sub(r'\n{3,}', '\n\n', prompt)
        prompt = re.sub(r' {2,}', ' ', prompt)
        # 去除注释性文字
        prompt = re.sub(r'#.*\n', '', prompt)
        return prompt.strip()

    def _compress_history(self, history: list, max_turns: int = 5) -> str:
        """压缩对话历史：保留最近 N 轮 + 摘要"""
        if len(history) <= max_turns:
            return "\n".join([f"用户: {h['user']}\n助手: {h['assistant']}"
                            for h in history])
        # 早期对话生成摘要
        old = history[:-max_turns]
        recent = history[-max_turns:]
        summary = self._summarize(old)
        recent_text = "\n".join([f"用户: {h['user']}\n助手: {h['assistant']}"
                                  for h in recent])
        return f"[之前对话摘要：{summary}]\n\n{recent_text}"

import tiktoken

class TokenMonitor:
    def __init__(self, model="gpt-4"):
        self.encoder = tiktoken.encoding_for_model(model)

    def count(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def estimate_cost(self, input_tokens: int, output_tokens: int,
                      model: str) -> float:
        PRICING = {
            "gpt-4": {"input": 0.03, "output": 0.06},
            "gpt-4o": {"input": 0.005, "output": 0.015},
            "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
            "claude-3.5-sonnet": {"input": 0.003, "output": 0.015},
        }
        p = PRICING.get(model, PRICING["gpt-4"])
        return (input_tokens * p["input"] + output_tokens * p["output"]) / 1000

from enum import Enum

class Complexity(Enum):
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"

MODEL_MAP = {
    Complexity.SIMPLE: "gpt-4o-mini",      # $0.15/$0.60 per 1M
    Complexity.MODERATE: "gpt-4o",          # $5/$15 per 1M
    Complexity.COMPLEX: "gpt-4",            # $30/$60 per 1M
}

class ModelRouter:
    def __init__(self, classifier_model="gpt-4o-mini"):
        self.classifier = classifier_model

    async def classify(self, query: str) -> Complexity:
        """用小模型做分类，成本低"""
        prompt = f"""判断以下查询的复杂度，只返回 simple/moderate/complex：
- simple: 简单问答、翻译、格式转换
- moderate: 代码生成、分析总结、多步推理
- complex: 复杂推理、创意写作、数学证明

查询：{query}"""
        resp = await self._call_model(self.classifier, prompt, max_tokens=5)
        level = resp.strip().lower()
        return Complexity(level) if level in ["simple", "moderate", "complex"] else Complexity.MODERATE

    async def route(self, query: str) -> str:
        complexity = await self.classify(query)
        return MODEL_MAP[complexity]

import asyncio

class BatchInference:
    """批量推理：合并多请求，利用 batch API 折扣"""

    def __init__(self, batch_size=100, max_wait=5.0):
        self.batch_size = batch_size
        self.max_wait = max_wait
        self.queue = asyncio.Queue()
        self.results = {}

    async def submit(self, query: str) -> str:
        task_id = id(query)
        await self.queue.put((task_id, query))
        # 等待结果
        while task_id not in self.results:
            await asyncio.sleep(0.01)
        return self.results.pop(task_id)

    async def process_batches(self, llm_client):
        while True:
            batch = []
            try:
                # 等待第一个请求
                first = await asyncio.wait_for(
                    self.queue.get(), timeout=self.max_wait
                )
                batch.append(first)
                # 非阻塞收集更多
                while len(batch) < self.batch_size:
                    try:
                        item = self.queue.get_nowait()
                        batch.append(item)
                    except asyncio.QueueEmpty:
                        break
                # 批量调用
                results = await self._batch_call(llm_client, batch)
                for task_id, result in results:
                    self.results[task_id] = result
            except asyncio.TimeoutError:
                continue

class TieredCache:
    """三级缓存 + 成本追踪"""

    def __init__(self):
        self.stats = {"hit_l1": 0, "hit_l2": 0, "miss": 0, "saved_cost": 0.0}

    async def get_or_call(self, query: str, llm_call, cache):
        # L1 精确缓存
        exact = cache.exact_get(query)
        if exact:
            self.stats["hit_l1"] += 1
            self.stats["saved_cost"] += self._estimate_cost(query)
            return exact

        # L2 语义缓存
        semantic = cache.semantic_get(query)
        if semantic:
            self.stats["hit_l2"] += 1
            self.stats["saved_cost"] += self._estimate_cost(query)
            return semantic

        # 未命中 → LLM 调用
        self.stats["miss"] += 1
        response = await llm_call(query)
        cache.set(query, response)
        return response

    def report(self) -> dict:
        total = sum(self.stats.values())
        hit_rate = (self.stats["hit_l1"] + self.stats["hit_l2"]) / max(total, 1)
        return {"hit_rate": hit_rate, "saved_cost": self.stats["saved_cost"]}

def tco_analysis(monthly_tokens_millions: float, api_price_per_1k: float,
                 server_monthly_cost: float, gpu_cost: float,
                 gpu_lifetime_months: int = 36):
    """TCO 对比分析"""
    api_monthly = monthly_tokens_millions * 1000 * api_price_per_1k
    self_deploy_monthly = server_monthly_cost + gpu_cost / gpu_lifetime_months

    break_even = server_monthly_cost / (api_price_per_1k * 1000)  # 百万 tokens
    cheaper = "self-deploy" if self_deploy_monthly < api_monthly else "api"

    return {
        "api_monthly_cost": api_monthly,
        "self_deploy_monthly_cost": self_deploy_monthly,
        "break_even_tokens_millions": break_even,
        "cheaper_option": cheaper,
        "monthly_savings": abs(api_monthly - self_deploy_monthly),
    }