模型网关与多模型路由

from openai import OpenAI

client = OpenAI(
    base_url="http://llm-gateway.internal:4000/v1",
    api_key="sk-team-a-virtual-key",   # 网关签发的虚拟 key，不是真实供应商 key
)

resp = client.chat.completions.create(
    model="chat-default",              # 逻辑别名，由网关解析为真实模型
    messages=[{"role": "user", "content": "用一句话解释什么是 RAG"}],
)
print(resp.choices[0].message.content)

export LITELLM_MASTER_KEY="sk-master-please-rotate"
export DATABASE_URL="postgresql://litellm:pass@localhost:5432/litellm"

pip install "litellm[proxy]==1.74.0"

litellm --config ./config.yaml --port 4000 --num_workers 4

model_list:
  # ---- 旗舰模型：质量优先场景 ----
  - model_name: chat-premium            # 对外暴露的逻辑名
    litellm_params:
      model: openai/gpt-4.1             # 真实模型
      api_key: os.environ/OPENAI_API_KEY
      rpm: 5000
      timeout: 60
    model_info:
      input_cost_per_token: 0.000002    # $2 / 1M tokens
      output_cost_per_token: 0.000008

  # ---- 主力模型：性价比之选 ----
  - model_name: chat-default
    litellm_params:
      model: openai/gpt-4.1-mini
      api_key: os.environ/OPENAI_API_KEY
      rpm: 10000
      timeout: 30
    model_info:
      input_cost_per_token: 0.0000004   # $0.40 / 1M tokens
      output_cost_per_token: 0.0000016

  # ---- 自托管模型：走本地推理引擎 ----
  - model_name: chat-selfhosted
    litellm_params:
      model: openai/qwen3-32b
      api_base: http://vllm-inference.svc:8000/v1
      api_key: os.environ/VLLM_KEY
      rpm: 2000

  # ---- 备用供应商：故障转移目标 ----
  - model_name: chat-backup
    litellm_params:
      model: anthropic/claude-sonnet-4-6
      api_key: os.environ/ANTHROPIC_API_KEY
      rpm: 3000

router_settings:
  routing_strategy: cost-based-routing   # 成本优先，见第三节
  num_retries: 2
  retry_after: 1
  allowed_fails: 3                       # 连续失败 3 次进入冷却
  cooldown_time: 30                      # 冷却 30 秒
  timeout: 30
  fallbacks:
    - chat-premium: ["chat-backup", "chat-default"]   # 旗舰挂了降级
    - chat-default: ["chat-backup"]
  context_window_fallbacks:
    - chat-default: ["chat-premium"]     # 上下文超限时切到大窗口模型
  redis_host: os.environ/REDIS_HOST
  redis_port: 6379

general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL
  # 预算与配额
  max_budget: 5000                       # 全局月度预算（美元）
  budget_duration: 30d

litellm_settings:
  drop_params: true                      # 自动丢弃上游不支持的参数
  cache: true
  cache_params:
    type: redis
    ttl: 3600
  request_timeout: 30
  set_verbose: false

import os
from litellm import Router

router = Router(
    model_list=[
        {
            "model_name": "chat-default",
            "litellm_params": {
                "model": "openai/gpt-4.1-mini",
                "api_key": os.environ["OPENAI_API_KEY"],
                "rpm": 10000,
            },
        },
        {
            "model_name": "chat-backup",
            "litellm_params": {
                "model": "anthropic/claude-sonnet-4-6",
                "api_key": os.environ["ANTHROPIC_API_KEY"],
                "rpm": 3000,
            },
        },
    ],
    routing_strategy="cost-based-routing",
    num_retries=2,
    allowed_fails=3,
    cooldown_time=30,
    fallbacks=[{"chat-default": ["chat-backup"]}],
    redis_host=os.environ.get("REDIS_HOST", "localhost"),
)

def ask(prompt: str) -> str:
    resp = router.completion(
        model="chat-default",
        messages=[{"role": "user", "content": prompt}],
        timeout=30,
    )
    # 通过响应头/元数据判断实际命中了哪个部署
    served = resp._hidden_params.get("model_id", "unknown")
    print(f"[route] served by {served}")
    return resp.choices[0].message.content

if __name__ == "__main__":
    print(ask("把下面这段话压缩成一句话：……"))

import numpy as np
from openai import OpenAI

client = OpenAI(base_url="http://llm-gateway.internal:4000/v1",
                api_key="sk-team-a-virtual-key")

PROTOTYPES = {
    "chat-default": ["把这句话翻译成英文", "帮我改写这段文字", "提取关键词"],
    "chat-premium": ["证明这个定理", "分析这段代码的并发缺陷", "设计一个分布式方案"],
}

def embed(texts: list[str]) -> np.ndarray:
    r = client.embeddings.create(model="embed-default", input=texts)
    return np.array([d.embedding for d in r.data], dtype=np.float32)

def route(query: str) -> str:
    proto_vecs = {tier: embed(samples) for tier, samples in PROTOTYPES.items()}
    q = embed([query])[0]
    q = q / (np.linalg.norm(q) + 1e-9)
    best, best_score = "chat-default", -1.0
    for tier, vecs in proto_vecs.items():
        sims = vecs @ q / (np.linalg.norm(vecs, axis=1) + 1e-9)
        score = float(sims.max())
        if score > best_score:
            best, best_score = tier, score
    return best

if __name__ == "__main__":
    for q in ["帮我把这段话翻译成日语", "证明 P 是否等于 NP 的思路"]:
        print(q, "->", route(q))

import litellm

def count_cost(model: str, messages: list[dict], max_output: int = 512) -> float:
    in_tokens = litellm.token_counter(model=model, messages=messages)
    # 按最坏情况预留输出 token，避免预算被击穿
    in_cost = in_tokens * litellm.model_cost[model]["input_cost_per_token"]
    out_cost = max_output * litellm.model_cost[model]["output_cost_per_token"]
    return in_cost + out_cost

msgs = [{"role": "user", "content": "用 200 字总结这篇文章"}]
print(f"预估单次成本: ${count_cost('gpt-4.1-mini', msgs):.6f}")

curl -s -X POST http://llm-gateway.internal:4000/key/generate \
  -H "Authorization: Bearer $LITELLM_MASTER_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "team_id": "team-a",
    "max_budget": 200,
    "budget_duration": "30d",
    "rpm_limit": 600,
    "models": ["chat-default", "embed-default"]
  }'

import json
import logging
import time
from litellm.integrations.custom_logger import CustomLogger

logger = logging.getLogger("llm_gateway_audit")
logger.setLevel(logging.INFO)

class AuditLogger(CustomLogger):
    def log_success_event(self, kwargs, response_obj, start_time, end_time):
        meta = kwargs.get("litellm_params", {}).get("metadata", {})
        hidden = getattr(response_obj, "_hidden_params", {}) or {}
        record = {
            "ts": time.time(),
            "team": meta.get("user_api_key_team_id", "unknown"),
            "logical_model": kwargs.get("model"),
            "served_deployment": hidden.get("model_id"),
            "retry_count": hidden.get("retries", 0),
            "cost_usd": kwargs.get("response_cost", 0.0),
            "total_ms": (end_time - start_time).total_seconds() * 1000,
            "status": "success",
        }
        logger.info(json.dumps(record, ensure_ascii=False))

    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
        record = {
            "ts": time.time(),
            "logical_model": kwargs.get("model"),
            "error": str(kwargs.get("exception")),
            "total_ms": (end_time - start_time).total_seconds() * 1000,
            "status": "failure",
        }
        logger.warning(json.dumps(record, ensure_ascii=False))

audit = AuditLogger()