API智能路由实战：五招砍掉80%AI账单

flowchart TD
    A[用户请求] --> B{任务分类器}
    B -->|分类/提取 70%| C[Tier1: Qwen3-8B<br/>$0.01/M]
    B -->|摘要/翻译 15%| D[Tier2: DeepSeek V4 Flash<br/>$0.25/M]
    B -->|分析/编码 10%| E[Tier3: Claude Sonnet 4.6<br/>$3.00/M]
    B -->|深度推理 5%| F[Tier4: Claude Opus 4.6<br/>$5.00/M]
    C --> G[质量检查]
    D --> G
    G -->|通过 ✅| H[返回结果]
    G -->|未通过 ❌| I[升级到下一级模型]
    I --> H
    E --> H
    F --> H

from enum import Enum
from openai import OpenAI

client = OpenAI(api_key="your-key", base_url="https://openrouter.ai/api/v1")

class ModelTier(Enum):
    BUDGET = "qwen/qwen3-8b"        # $0.01/M
    STANDARD = "deepseek/deepseek-v4-flash"  # $0.25/M
    PREMIUM = "anthropic/claude-sonnet-4.6"   # $3.00/M
    FRONTIER = "anthropic/claude-opus-4.6"    # $5.00/M

TASK_ROUTING = {
    "classify": ModelTier.BUDGET,    # 分类任务用最便宜的
    "extract": ModelTier.BUDGET,     # 信息提取也是
    "summarize": ModelTier.STANDARD, # 摘要用中等
    "translate": ModelTier.STANDARD, # 翻译也是
    "analyze": ModelTier.PREMIUM,    # 分析用好的
    "code_review": ModelTier.PREMIUM,
    "research": ModelTier.FRONTIER,  # 研究才用最贵的
    "complex_code": ModelTier.FRONTIER,
}

def route_request(task_type: str, prompt: str, max_tokens: int = 1000):
    tier = TASK_ROUTING.get(task_type, ModelTier.STANDARD)
    response = client.chat.completions.create(
        model=tier.value,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

import anthropic

client = anthropic.Anthropic()

SYSTEM_PROMPT = """你是一个专业的法律文档分析助手...
[这里放3000+ tokens的详细指令和示例]
"""

def analyze_with_cache(contract_text: str):
    response = client.messages.create(
        model="claude-sonnet-4.6",
        max_tokens=2000,
        system=[{
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}  # 关键：标记缓存点
        }],
        messages=[{
            "role": "user",
            "content": f"分析这份合同：\n\n{contract_text}"
        }],
    )
    # 查看缓存命中情况
    usage = response.usage
    print(f"缓存读取token: {usage.cache_read_input_tokens}")
    return response.content[0].text

import json
from openai import OpenAI

client = OpenAI()

def classify_with_escalation(text: str, categories: list) -> dict:
    """先用便宜模型分类，置信度不够就升级"""
    prompt = f"""将以下文本分类到这些类别之一：{categories}
    文本：{text}
    返回JSON：{{"category": "...", "confidence": 0.0-1.0}}"""

    # 第一轮：用最便宜的模型
    resp = client.chat.completions.create(
        model="gpt-4o-mini",       # $0.15/M
        max_tokens=100,
        response_format={"type": "json_object"},
        messages=[{"role": "user", "content": prompt}],
    )
    result = json.loads(resp.choices[0].message.content)

    # 置信度够高就直接返回
    if result.get("confidence", 0) >= 0.85:
        return result              # 80%+ 的请求在这里结束

    # 置信度不够，升级到强模型
    resp = client.chat.completions.create(
        model="gpt-5.4",           # $2.00/M，但只用于20%的请求
        max_tokens=100,
        response_format={"type": "json_object"},
        messages=[{"role": "user", "content": prompt}],
    )
    return json.loads(resp.choices[0].message.content)

MAX_TOKENS_MAP = {
    "classify": 10,      # "spam" 或 "not_spam"
    "short_answer": 100,  # 一两句话
    "paragraph": 300,     # 一段话
    "article": 2000,      # 一篇文章
    "analysis": 4000,     # 深度分析
}

response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[...],
    max_tokens=MAX_TOKENS_MAP[task_type],  # 永远手动设置！
)

import json
from openai import OpenAI

client = OpenAI()

def submit_batch(requests: list) -> str:
    """提交批量请求，50%折扣"""
    batch_lines = []
    for i, req in enumerate(requests):
        batch_lines.append(json.dumps({
            "custom_id": f"request-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": req.get("model", "gpt-4o-mini"),
                "messages": req["messages"],
                "max_tokens": req.get("max_tokens", 100),
            }
        }))

    # 上传JSONL文件
    with open("/tmp/batch.jsonl", "w") as f:
        f.write("\n".join(batch_lines))

    batch_file = client.files.create(
        file=open("/tmp/batch.jsonl", "rb"),
        purpose="batch"
    )

    # 创建批量任务
    batch = client.batches.create(
        input_file_id=batch_file.id,
        endpoint="/v1/chat/completions",
        completion_window="24h",  # 24小时内完成
    )
    return batch.id

flowchart LR
    subgraph 优化前
        A1[全部请求<br/>→ GPT-4o] --> A2[月费: $750/天<br/>$22,500/月]
    end
    subgraph 优化后
        B1[70% 简单分析] --> B2[缓存/Qwen3-8B<br/>$12/天]
        B3[20% 标准分析] --> B4[DeepSeek V4 Flash<br/>$30/天]
        B5[8% 复杂分析] --> B6[DeepSeek V4 Pro<br/>$52/天]
        B7[2% 专家审查] --> B8[DeepSeek-R1<br/>$42/天]
    end
    B2 --> C[总计: $136/天<br/>$4,080/月]
    B4 --> C
    B6 --> C
    B8 --> C
    A2 -.->|省82%| C

月API费 = Σ(各任务类型 × 该类型调用量 × 该类型最优模型单价)