AI 应用的成本、延迟与质量三角

# 简单的规则路由示例
def route_request(user_message: str, intent_score: float) -> str:
    if intent_score > 0.85 and len(user_message) < 100:
        return "flash"       # 简单问题走轻量模型
    elif intent_score > 0.5:
        return "standard"    # 中等复杂度走中端模型
    else:
        return "flagship"    # 复杂问题走旗舰模型

// 带置信度的路由中间件
interface RoutingResult {
  model: 'flash' | 'standard' | 'flagship';
  confidence: number;
  reason: string;
}
 
function classifyAndRoute(
  message: string,
  classifier: IntentClassifier
): RoutingResult {
  const result = classifier.predict(message);
 
  // 置信度不足时升级到更高级模型
  if (result.confidence < 0.6) {
    return { model: 'flagship', confidence: result.confidence, reason: 'low_confidence' };
  }
 
  const modelMap: Record<string, RoutingResult['model']> = {
    faq: 'flash',
    order_issue: 'standard',
    complaint: 'flagship',
  };
 
  return {
    model: modelMap[result.intent] ?? 'standard',
    confidence: result.confidence,
    reason: result.intent,
  };
}

# 精确匹配缓存实现
import hashlib
import json
from functools import lru_cache
 
def make_cache_key(messages: list[dict], model: str) -> str:
    content = json.dumps(messages, sort_keys=True) + model
    return hashlib.sha256(content.encode()).hexdigest()
 
async def cached_completion(
    messages: list[dict],
    model: str,
    cache_backend,
    llm_client
) -> str:
    key = make_cache_key(messages, model)
 
    # 尝试从缓存获取
    cached = await cache_backend.get(key)
    if cached:
        return cached
 
    # 缓存未命中，调用 LLM
    response = await llm_client.complete(messages=messages, model=model)
    await cache_backend.set(key, response, ttl=3600)
    return response

# 语义缓存实现（简化版）
from sentence_transformers import SentenceTransformer
import numpy as np
 
class SemanticCache:
    def __init__(self, threshold: float = 0.92):
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.threshold = threshold
        self.store: list[tuple[np.ndarray, str]] = []
 
    def get(self, query: str) -> str | None:
        query_vec = self.encoder.encode(query)
        for stored_vec, response in self.store:
            similarity = np.dot(query_vec, stored_vec) / (
                np.linalg.norm(query_vec) * np.linalg.norm(stored_vec)
            )
            if similarity >= self.threshold:
                return response
        return None
 
    def put(self, query: str, response: str):
        query_vec = self.encoder.encode(query)
        self.store.append((query_vec, response))

// Prompt 前缀缓存 —— 利用 Anthropic API 的 cache_control
async function chatWithPrefixCache(
  systemPrompt: string,
  userMessage: string,
  client: AnthropicClient
) {
  return client.messages.create({
    model: 'claude-sonnet-4-20250514',
    max_tokens: 4096,
    system: [
      {
        type: 'text',
        text: systemPrompt,
        cache_control: { type: 'ephemeral' },  // 标记为可缓存前缀
      },
    ],
    messages: [{ role: 'user', content: userMessage }],
  });
}

# 降级策略实现
import asyncio
from datetime import timedelta
 
async def call_with_fallback(
    messages: list[dict],
    primary_model: str,
    fallback_models: list[str],
    timeout: float = 10.0,
    llm_client = None,
) -> tuple[str, str]:
    """依次尝试多个模型，超时或失败时降级"""
    models = [primary_model] + fallback_models
 
    for model in models:
        try:
            response = await asyncio.wait_for(
                llm_client.complete(messages=messages, model=model),
                timeout=timeout,
            )
            return response, model
        except asyncio.TimeoutError:
            continue
        except Exception:
            continue
 
    # 所有模型都失败，返回模板结果
    return generate_template_response(messages), 'template'
 
 
# 成本异常时主动降级
async def cost_aware_completion(
    messages: list[dict],
    monthly_budget_remaining: float,
    llm_client = None,
) -> str:
    if monthly_budget_remaining < 100:
        # 预算紧张，强制使用最便宜的模型
        model = 'flash'
    elif monthly_budget_remaining < 500:
        model = 'standard'
    else:
        model = 'flagship'
 
    return await llm_client.complete(messages=messages, model=model)

// 流式输出 —— 降低感知延迟
async function* streamCompletion(
  messages: ChatMessage[],
  model: string,
  client: LLMClient
): AsyncGenerator<string> {
  const stream = await client.stream({ messages, model });
 
  for await (const chunk of stream) {
    // 每收到一个 token 就推送给前端
    yield chunk.content;
  }
}
 
// 前端配合使用：用户看到第一个字的时间大幅缩短
// 即使总生成时间不变，感知延迟从「等待 5 秒」变成「0.3 秒开始出字」

// 成本监控中间件
interface CostMetrics {
  model: string;
  inputTokens: number;
  outputTokens: number;
  cacheHit: boolean;
  latency: number;
  cost: number;
  degraded: boolean;
  userId: string;
  route: string;
}
 
const PRICE_TABLE: Record<string, { input: number; output: number }> = {
  flash:    { input: 0.075,  output: 0.30 },
  standard: { input: 0.50,   output: 1.50 },
  flagship: { input: 5.00,   output: 15.00 },
};
 
function calculateCost(
  model: string,
  inputTokens: number,
  outputTokens: number,
  cacheHit: boolean
): number {
  const price = PRICE_TABLE[model];
  if (!price) return 0;
 
  const inputCost = cacheHit
    ? (inputTokens / 1_000_000) * price.input * 0.1  // 缓存 token 价格约 1/10
    : (inputTokens / 1_000_000) * price.input;
  const outputCost = (outputTokens / 1_000_000) * price.output;
 
  return inputCost + outputCost;
}