LLM 成本优化实战：10 种降低 API 费用的方法

import os
from typing import Dict, Any

class ModelRouter:
    """根据请求复杂度路由到不同模型"""
    
    def __init__(self):
        self.routes = {
            "simple": {
                "model": "gpt-4o-mini",
                "max_tokens": 512,
                "cost_per_1k": 0.00015,  # input $0.15/1M
            },
            "medium": {
                "model": "gpt-4o",
                "max_tokens": 1024,
                "cost_per_1k": 0.0025,
            },
            "complex": {
                "model": "o3",
                "max_tokens": 4096,
                "cost_per_1k": 0.015,
            },
        }
    
    def classify(self, query: str) -> str:
        """基于规则快速分类"""
        query_lower = query.lower()
        
        # 简单查询：问候、翻译短句、简单事实
        if any(kw in query_lower for kw in ["你好", "翻译", "what is", "hi ", "hello"]):
            return "simple"
        
        # 复杂查询：代码生成、多步推理、长文写作
        if any(kw in query_lower for kw in ["写代码", "分析", "设计", "架构", "debug"]):
            return "complex"
        
        # 默认中等
        return "medium"
    
    def route(self, query: str) -> Dict[str, Any]:
        level = self.classify(query)
        return self.routes[level]


# 使用示例
router = ModelRouter()
config = router.route("帮我写一个 Python 排序算法")
print(f"路由到: {config['model']}, 预估成本: ${config['cost_per_1k'] * 2:.6f}/1K tokens")

# 优化前
verbose_prompt = """
你是一个专业的客服助手。你的职责是回答用户关于产品的问题。
请始终保持礼貌和专业。如果你不知道答案，请说"我需要帮您查询一下"。
请不要编造信息。回答要简洁明了。避免使用过于技术化的术语。
如果用户的问题与产品无关，请礼貌地将话题引导回产品相关内容。
请在回答结束时询问用户是否还有其他问题。
"""

# 优化后（含义不变，Token 减少约 50%）
optimized_prompt = """专业客服助手。职责：回答产品问题。
规则：礼貌专业；不知则说"我需要帮您查询一下"；不编造；简洁避免术语；跑题则引导回产品；结尾询问是否还有问题。"""

# Token 对比
import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o")
print(f"优化前: {len(enc.encode(verbose_prompt))} tokens")
print(f"优化后: {len(enc.encode(optimized_prompt))} tokens")

import hashlib
import numpy as np
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class CacheEntry:
    answer: str
    embedding: np.ndarray
    created_at: datetime
    hit_count: int = 0

class SemanticCache:
    """基于向量相似度的语义缓存"""
    
    def __init__(self, threshold: float = 0.92, ttl_minutes: int = 60):
        self.cache: dict[str, CacheEntry] = {}
        self.threshold = threshold
        self.ttl = timedelta(minutes=ttl_minutes)
    
    def _embed(self, text: str) -> np.ndarray:
        """使用 embedding API 获取向量（这里用模拟）"""
        # 实际使用: openai.embeddings.create(model="text-embedding-3-small", input=text)
        return np.random.randn(1536)  # 模拟
    
    def _cosine_sim(self, a: np.ndarray, b: np.ndarray) -> float:
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    def get(self, query: str) -> str | None:
        query_emb = self._embed(query)
        now = datetime.now()
        
        best_sim = 0.0
        best_key = None
        
        for key, entry in self.cache.items():
            if now - entry.created_at > self.ttl:
                continue
            sim = self._cosine_sim(query_emb, entry.embedding)
            if sim > best_sim:
                best_sim = sim
                best_key = key
        
        if best_sim >= self.threshold and best_key:
            entry = self.cache[best_key]
            entry.hit_count += 1
            return entry.answer
        return None
    
    def set(self, query: str, answer: str):
        self.cache[query] = CacheEntry(
            answer=answer,
            embedding=self._embed(query),
            created_at=datetime.now(),
        )


# 使用示例
cache = SemanticCache(threshold=0.92, ttl_minutes=30)

# 第一次查询 — miss
result = cache.get("Python 怎么读取 JSON 文件？")
if result is None:
    print("Cache miss, calling API...")
    answer = "使用 json.load() 函数..."
    cache.set("Python 怎么读取 JSON 文件？", answer)

# 第二次相似查询 — hit
result = cache.get("Python 如何读取 JSON？")
print(f"Cache hit: {result is not None}")

import openai

client = openai.OpenAI()

def smart_completion(prompt: str, max_tokens: int = 1024) -> str:
    """带提前终止的智能补全"""
    collected = []
    
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
        stream=True,
        stream_options={"include_usage": True},
    )
    
    total_tokens = 0
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            collected.append(content)
            total_tokens += 1
            
            # 检测到结束语提前终止
            joined = "".join(collected[-10:])
            if any(stop in joined for stop in ["完毕。", "以上是", "## 结束"]):
                break
        
        if chunk.usage:
            total_tokens = chunk.usage.total_tokens
    
    print(f"实际使用 tokens: {total_tokens}")
    return "".join(collected)

import json
import openai

client = openai.OpenAI()

def create_batch_file(requests: list[dict]) -> str:
    """创建批处理文件"""
    batch_lines = []
    for req in requests:
        line = {
            "custom_id": req["id"],
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "gpt-4o",
                "messages": req["messages"],
                "max_tokens": 1000,
            },
        }
        batch_lines.append(json.dumps(line))
    
    # 上传批处理文件
    file_content = "\n".join(batch_lines)
    file_obj = client.files.create(
        file=(file_content.encode(), "batch.jsonl"),
        purpose="batch",
    )
    
    # 创建批处理任务
    batch = client.batches.create(
        input_file_id=file_obj.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",
    )
    
    return batch.id

# 批量处理 10000 条数据
requests = [
    {"id": f"req-{i}", "messages": [{"role": "user", "content": f"分析文本: {text}"}]}
    for i, text in enumerate(data_batch)
]

batch_id = create_batch_file(requests)
print(f"批处理任务已创建: {batch_id}")
print("24 小时内完成，费用享受 50% 折扣")

from collections import deque

class ContextManager:
    """智能上下文窗口管理"""
    
    def __init__(self, max_messages: int = 10, max_tokens: int = 4000):
        self.messages: deque = deque(maxlen=max_messages)
        self.max_tokens = max_tokens
        self.summary = ""
    
    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        
        # 超过阈值时触发摘要压缩
        if self._estimate_tokens() > self.max_tokens:
            self._compress()
    
    def _estimate_tokens(self) -> int:
        """粗略估算 Token 数（1 token ≈ 4 chars 中文 / 0.75 word 英文）"""
        total_chars = sum(len(m["content"]) for m in self.messages)
        return total_chars // 2  # 中英混合粗略值
    
    def _compress(self):
        """将旧消息压缩为摘要"""
        old_messages = list(self.messages)[:len(self.messages)//2]
        new_messages = list(self.messages)[len(self.messages)//2:]
        
        # 调用小模型生成摘要
        summary_prompt = "请将以下对话历史压缩为简洁摘要：\n"
        for m in old_messages:
            summary_prompt += f"{m['role']}: {m['content'][:200]}\n"
        
        # 实际调用: client.chat.completions.create(model="gpt-4o-mini", ...)
        self.summary = f"[对话摘要] 用户询问了技术问题，已提供解决方案。"
        self.messages = deque(new_messages, maxlen=self.max_messages)
    
    def get_context(self) -> list[dict]:
        context = []
        if self.summary:
            context.append({"role": "system", "content": self.summary})
        context.extend(self.messages)
        return context

# 使用 response_format 强制 JSON 输出
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "分析以下文本的情感，返回JSON格式",
    }],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "sentiment",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "sentiment": {"type": "string", "enum": ["pos", "neg", "neu"]},
                    "score": {"type": "number"},
                },
                "required": ["sentiment", "score"],
                "additionalProperties": False,
            },
        },
    },
    max_tokens=100,  # JSON 输出可以设置很小的 max_tokens
)

# 场景：意图分类
# 方案 A：LLM 生成（贵且慢）
def classify_with_llm(text: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"分类: {text}\n选项: 技术/商务/闲聊"}],
        max_tokens=10,
    )
    return resp.choices[0].message.content

# 方案 B：Embedding + KNN（便宜100倍）
from sklearn.neighbors import KNeighborsClassifier
import numpy as np

# 预计算训练集 Embedding
intents = ["技术", "商务", "闲聊"]
training_examples = {
    "技术": ["怎么写代码", "Python教程", "服务器配置", "API调用"],
    "商务": ["产品价格", "合作方案", "报价单", "合同条款"],
    "闲聊": ["今天天气不错", "你好呀", "最近怎么样", "吃饭了吗"],
}

# 训练 KNN 分类器
X, y = [], []
for intent, examples in training_examples.items():
    for ex in examples:
        emb = get_embedding(ex)  # 一次性预计算
        X.append(emb)
        y.append(intent)

clf = KNeighborsClassifier(n_neighbors=3)
clf.fit(X, y)

def classify_with_embedding(text: str) -> str:
    emb = get_embedding(text)  # 成本仅 $0.00002/1K
    return clf.predict([emb])[0]

class ProviderArbitrage:
    """多供应商价格套利"""
    
    providers = {
        "openai": {
            "gpt-4o": {"input": 2.5, "output": 10.0},
            "gpt-4o-mini": {"input": 0.15, "output": 0.6},
        },
        "azure_openai": {
            "gpt-4o": {"input": 2.5, "output": 10.0},
            "gpt-4o-mini": {"input": 0.15, "output": 0.6},
        },
        "deepseek": {
            "deepseek-v3": {"input": 0.27, "output": 1.1},
        },
        "anthropic": {
            "claude-3.5-sonnet": {"input": 3.0, "output": 15.0},
        },
    }
    
    def cheapest(self, capability: str = "general_chat") -> tuple[str, str]:
        """返回最便宜的可用模型"""
        candidates = [
            ("deepseek", "deepseek-v3", 0.27, 1.1),
            ("openai", "gpt-4o-mini", 0.15, 0.6),
            ("anthropic", "claude-3.5-sonnet", 3.0, 15.0),
        ]
        # 按输入价格排序
        candidates.sort(key=lambda x: x[2])
        return candidates[0][0], candidates[0][1]

import time
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class UsageRecord:
    timestamp: float
    model: str
    input_tokens: int
    output_tokens: int
    cost: float

class CostMonitor:
    """实时成本监控与告警"""
    
    def __init__(self, daily_budget: float = 100.0):
        self.daily_budget = daily_budget
        self.records: list[UsageRecord] = []
        self.alert_thresholds = [0.5, 0.8, 0.95]
    
    def record(self, model: str, input_tokens: int, output_tokens: int, cost: float):
        rec = UsageRecord(time.time(), model, input_tokens, output_tokens, cost)
        self.records.append(rec)
        self._check_budget()
    
    def _check_budget(self):
        today_cost = self._today_cost()
        ratio = today_cost / self.daily_budget
        
        for threshold in self.alert_thresholds:
            if ratio >= threshold and not self._already_alerted(threshold):
                self._send_alert(
                    f"⚠️ 日预算已用 {ratio:.0%} (${today_cost:.2f}/${self.daily_budget})"
                )
                break
    
    def _today_cost(self) -> float:
        today_start = time.time() - (time.time() % 86400)
        return sum(r.cost for r in self.records if r.timestamp >= today_start)
    
    def summary(self) -> dict:
        """生成成本摘要"""
        by_model = defaultdict(lambda: {"cost": 0, "calls": 0, "tokens": 0})
        for r in self.records:
            by_model[r.model]["cost"] += r.cost
            by_model[r.model]["calls"] += 1
            by_model[r.model]["tokens"] += r.input_tokens + r.output_tokens
        return dict(by_model)