LLM五层省钱架构：从路由到缓存的成本压缩实战

graph TD
    A[用户请求 100%] --> B{是否重复/相似?}
    B -->|约40%是| C[重复请求: 直接返回缓存]
    B -->|约60%否| D{任务复杂度?}
    D -->|约30%简单| E[路由到小模型]
    D -->|约20%中等| F[路由到中等模型]
    D -->|约10%复杂| G[路由到大模型]
    E --> H{是否批量任务?}
    F --> H
    G --> H
    H -->|是| I[Batch API 5折]
    H -->|否| J{Prompt前缀可缓存?}
    J -->|是| K[缓存Token 1折]
    J -->|否| L[标准调用 原价]

    style C fill:#4CAF50,color:#fff
    style E fill:#8BC34A,color:#fff
    style I fill:#FF9800,color:#fff
    style K fill:#2196F3,color:#fff
    style L fill:#f44336,color:#fff

pip install routellm

from routellm import Router

router = Router(
    routers=["mf"],  # 使用Matrix Factorization路由器
    strong_model="openai/gpt-4o",
    cheap_model="openai/gpt-4o-mini",
    threshold=0.5  # 阈值越低，越多请求走cheap模型
)

# 路由器自动判断复杂度
result = router.chat.completions.create(
    model="router-mf-0.5",
    messages=[{"role": "user", "content": "帮我翻译这段话"}]
)

import openai

def smart_route(user_prompt: str) -> str:
    """用便宜模型判断复杂度，再路由到合适的模型"""
    # Step 1: 用最便宜的模型判断
    judge = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "system", 
            "content": "判断这个任务的复杂度，只回答: simple/medium/complex"
        }, {
            "role": "user", 
            "content": user_prompt
        }],
        max_tokens=10
    )

    complexity = judge.choices[0].message.content.strip().lower()

    # Step 2: 路由到对应模型
    model_map = {
        "simple": "gpt-4o-mini",    # $0.15/$0.60
        "medium": "deepseek-chat",   # $0.27/$1.10
        "complex": "gpt-4o"          # $2.50/$10.00
    }

    return model_map.get(complexity, "gpt-4o-mini")

import anthropic

client = anthropic.Anthropic()

# 将system prompt标记为缓存
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": """你是一个专业的代码审查助手。
            你的职责是：
            1. 检查代码质量
            2. 发现潜在bug
            3. 提出优化建议
            ...（这里可能是几千token的指令）""",
            "cache_control": {"type": "ephemeral"}  # 关键：标记缓存
        }
    ],
    messages=[{
        "role": "user",
        "content": "帮我审查这段代码：def add(a, b): return a + b"
    }]
)

flowchart LR
    A[新请求] --> B[生成Embedding]
    B --> C{向量相似度 > 阈值?}
    C -->|是 0.95+| D[返回缓存结果]
    C -->|否| E[调用LLM]
    E --> F[存入缓存]
    F --> G[返回结果]

    style D fill:#4CAF50,color:#fff
    style E fill:#FF9800,color:#fff

from gptcache import Cache
from gptcache.adapter import openai as cache_openai
from gptcache.embedding import OpenAI as EmbedOpenAI
from gptcache.similarity_evaluation import Cosine

# 初始化缓存
cache = Cache()
cache.init(
    embedding_func=EmbedOpenAI().to_embeddings,
    similarity_evaluation=Cosine(),
    data_manager=manager  # 可选Redis/SQLite/ChromaDB
)

# 透明替换openai调用
response = cache_openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "什么是RAG?"}],
    cache_obj=cache,
    similarity_threshold=0.85  # 相似度阈值
)

import openai
import json

client = openai.OpenAI()

# Step 1: 准备批量请求
requests = []
for i, text in enumerate(texts_to_translate):
    requests.append({
        "custom_id": f"translate-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",  # Batch用mini更划算
            "messages": [{"role": "user", "content": f"翻译成中文: {text}"}]
        }
    })

# Step 2: 上传文件
batch_file = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch"
)

# Step 3: 创建批量任务
batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

# Step 4: 等待完成后获取结果（通常几小时内完成）
# status = client.batches.retrieve(batch.id)

graph TB
    A[请求进入] --> B{是否需要顶级质量?}
    B -->|否，草稿/分类/格式化| C[免费模型层]
    B -->|是| D{是否实时?}
    C --> C1[Llama 3.3 70B<br/>OpenRouter免费]
    C --> C2[DeepSeek Chat<br/>OpenRouter免费]
    C --> C3[Qwen 2.5 7B<br/>OpenRouter免费]
    D -->|否| E[Batch API<br/>5折]
    D -->|是| F{复杂度?}
    F -->|简单| G[Haiku/4o-mini<br/>$0.15-$0.80/M]
    F -->|复杂| H[Sonnet/4o<br/>$2.50-$3.00/M]

    style C fill:#4CAF50,color:#fff
    style E fill:#FF9800,color:#fff
    style G fill:#2196F3,color:#fff
    style H fill:#f44336,color:#fff

xychart-beta
    title "五层架构累计成本压缩效果"
    x-axis ["基线" "+路由" "+Prompt缓存" "+语义缓存" "+Batch" "+免费层"]
    y-axis "月度成本 USD" 0 --> 9000
    bar [8500, 4900, 4050, 2850, 2170, 1700]