LiteLLM 路由策略与 Fallback 机制源码剖析 #

router = litellm.Router(
    # ====== 1. 模型注册 ======
    model_list=[
        {
            "model_name": "gpt-4",
            "litellm_params": {
                "model": "azure/gpt-4",
                "api_key": os.environ["AZURE_API_KEY"],
                "api_base": "https://xxx.openai.azure.com/",
            },
            "tpm": 100000,      # 每分钟 token 限额
            "rpm": 1000,         # 每分钟请求限额
            "weight": 1,         # 权重（simple-shuffle 用）
        },
        {
            "model_name": "gpt-4",  # 同名 = 同一 model_group
            "litellm_params": {
                "model": "openai/gpt-4",
                "api_key": os.environ["OPENAI_API_KEY"],
            },
        },
    ],

    # ====== 2. 路由策略 ======
    routing_strategy="latency-based-routing",  # 5 种之一
    routing_groups=[...],                       # 按模型名分组独立路由（可选）
    routing_strategy_args={"ttl": 3600},        # 策略参数

    # ====== 3. 重试 ======
    num_retries=3,                              # 同 deployment 重试次数
    retry_after=0,                              # 失败后最小等待秒数
    retry_policy=litellm.RetryPolicy(           # 按异常类型定制
        RateLimitErrorRetries=5,
        TimeoutErrorRetries=3,
        ContentPolicyViolationErrorRetries=0,
    ),

    # ====== 4. Fallback ======
    fallbacks=[
        {"gpt-4": "gpt-3.5-turbo"},             # 特定模型 fallback
        {"*": "gpt-4o-mini"},                    # 全局兜底
    ],
    max_fallbacks=5,                            # 最大 fallback 深度
    context_window_fallbacks=[                  # 上下文超长自动降级
        {"gpt-3.5-turbo": "gpt-4-turbo"}
    ],
    content_policy_fallbacks=[                  # 内容审查拦截自动降级
        {"azure-gpt-4": "openai-gpt-4"}
    ],

    # ====== 5. Cooldown ======
    allowed_fails=3,                            # 失败 N 次后进入冷却
    cooldown_time=60,                           # 冷却时长（秒）

    # ====== 6. 高可用 ======
    redis_url="redis://localhost:6379",         # 多实例共享状态
    enable_pre_call_checks=True,                # 调用前健康检查
    enable_weighted_failover=True,              # 同组内权重故障转移
)

# router.py:949-1089 核心逻辑
def routing_strategy_init(self, routing_strategy, routing_strategy_args):
    """初始化路由策略引擎"""
    strategy_map = {
        "simple-shuffle":       self._init_simple_shuffle,
        "least-busy":           self._init_least_busy,
        "usage-based-routing":  self._init_lowest_tpm_rpm,
        "latency-based-routing": self._init_lowest_latency,
        "cost-based-routing":   self._init_lowest_cost,
    }

    init_fn = strategy_map.get(routing_strategy)
    if not init_fn:
        raise ValueError(f"Unknown routing strategy: {routing_strategy}")

    # 初始化策略实例，注入 DualCache（In-Memory + Redis）
    init_fn(routing_strategy_args)

class BaseRoutingStrategy(ABC):
    def __init__(self, dual_cache: DualCache, ...):
        self.dual_cache = dual_cache  # In-Memory + Redis 双缓存

    # 子类必须实现：
    # async def async_get_available_deployments(...) -> List[Deployment]

# router_strategy/simple_shuffle.py
def simple_shuffle(deployments: List[Deployment]) -> List[Deployment]:
    """
    随机打乱 deployment 列表，支持权重
    """
    import random

    # 分离有权重和无权重的 deployment
    weighted = [d for d in deployments if d.get("weight", 1) != 1]
    unweighted = [d for d in deployments if d.get("weight", 1) == 1]

    if weighted:
        # 按权重概率选择
        weights = [d["weight"] for d in weighted]
        total = sum(weights)
        # 权重归一化为概率
        probs = [w / total for w in weights]
        # 按概率打乱
        random.shuffle(weighted, lambda: random.choices(range(len(weighted)), weights=probs)[0])
    else:
        random.shuffle(unweighted)

    return weighted + unweighted

router = Router(
    model_list=[
        {"model_name": "gpt-4", "litellm_params": {...}, "weight": 3},  # 3/5 概率
        {"model_name": "gpt-4", "litellm_params": {...}, "weight": 1},  # 1/5 概率
        {"model_name": "gpt-4", "litellm_params": {...}, "weight": 1},  # 1/5 概率
    ],
    routing_strategy="simple-shuffle",  # 可省略，这是默认值
)

# router_strategy/least_busy.py 核心逻辑
class LeastBusyLoggingHandler(CustomLogger):
    """跟踪每个 deployment 的并发请求数"""

    def log_success_event(self, kwargs, response_obj, start_time, end_time):
        """请求成功: 递减并发计数"""
        deployment_id = kwargs["litellm_params"]["model_info"]["id"]
        key = f"leastbusy:{deployment_id}"
        self.dual_cache.in_memory_cache.async_increment(key, value=-1)

    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
        """请求失败: 同样递减并发计数"""
        ...

    def get_least_busy_deployments(self, deployments):
        """返回按并发数升序排序的 deployment 列表"""
        for d in deployments:
            dep_id = d["model_info"]["id"]
            concurrency = self.dual_cache.in_memory_cache.get(f"leastbusy:{dep_id}") or 0
            d["_concurrency"] = concurrency
        return sorted(deployments, key=lambda d: d["_concurrency"])

# router_strategy/lowest_tpm_rpm.py 核心逻辑
class LowestTPMLoggingHandler(CustomLogger):
    """跟踪每个 deployment 的 TPM/RPM 用量"""

    def log_success_event(self, kwargs, response_obj, start_time, end_time):
        model_group = kwargs["litellm_params"]["model_group"]
        dep_id = kwargs["litellm_params"]["model_info"]["id"]

        # 时间粒度: 日期-小时-分钟
        precise_minute = datetime.now().strftime("%Y-%m-%d-%H-%M")

        # 获取 token 数
        total_tokens = response_obj.usage.total_tokens if response_obj.usage else 0

        # 更新缓存: {model_group}_map: {dep_id: {"tpm": N, "rpm": M, "minute": "..."}}
        latency_key = f"{model_group}_map"
        current_data = self.dual_cache.in_memory_cache.get(latency_key) or {}

        if dep_id not in current_data:
            current_data[dep_id] = {}

        if current_data[dep_id].get("minute") != precise_minute:
            # 新分钟，重置计数
            current_data[dep_id] = {
                "minute": precise_minute,
                "tpm": total_tokens,
                "rpm": 1,
            }
        else:
            current_data[dep_id]["tpm"] += total_tokens
            current_data[dep_id]["rpm"] += 1

        self.dual_cache.in_memory_cache.set(latency_key, current_data)

# router_strategy/lowest_latency.py 核心逻辑
class LowestLatencyLoggingHandler(CustomLogger):
    """跟踪每个 deployment 的响应延迟"""

    def log_success_event(self, kwargs, response_obj, start_time, end_time):
        model_group = kwargs["litellm_params"]["model_group"]
        dep_id = kwargs["litellm_params"]["model_info"]["id"]

        # 计算响应时间（毫秒）
        response_ms = end_time - start_time

        # 流式请求：使用 Time-To-First-Token (TTFT) 而非总耗时
        if kwargs.get("stream"):
            ttft = kwargs.get("completion_start_time", end_time) - start_time
            time_to_first_token = ttft

        # 存储到缓存: {model_group}_map: {dep_id: {"latency": [t1, t2, ...]}}
        latency_key = f"{model_group}_map"
        current_data = self.dual_cache.in_memory_cache.get(latency_key) or {}

        if dep_id not in current_data:
            current_data[dep_id] = {"latency": []}

        current_data[dep_id]["latency"].append(response_ms)

        # 限制历史列表大小（默认 10），防止内存膨胀
        max_size = self.routing_args.max_latency_list_size  # 默认 10
        if len(current_data[dep_id]["latency"]) > max_size:
            current_data[dep_id]["latency"] = current_data[dep_id]["latency"][-max_size:]

        self.dual_cache.in_memory_cache.set(latency_key, current_data)

    def get_lowest_latency_deployments(self, deployments):
        """返回按平均延迟升序排序的 deployment 列表"""
        model_group = ...  # 获取当前 model_group
        latency_key = f"{model_group}_map"
        data = self.dual_cache.in_memory_cache.get(latency_key) or {}

        for d in deployments:
            dep_id = d["model_info"]["id"]
            latencies = data.get(dep_id, {}).get("latency", [])
            d["_avg_latency"] = sum(latencies) / len(latencies) if latencies else float('inf')

        return sorted(deployments, key=lambda d: d["_avg_latency"])

router = Router(
    model_list=[
        {"model_name": "gpt-4", "litellm_params": {"model": "azure/gpt-4", ...}},
        {"model_name": "gpt-4", "litellm_params": {"model": "openai/gpt-4", ...}},
        {"model_name": "gpt-4", "litellm_params": {"model": "vertex_ai/gpt-4", ...}},
    ],
    routing_strategy="latency-based-routing",
    routing_strategy_args={
        "ttl": 3600,                # 延迟数据 TTL（1 小时）
        "lowest_latency_buffer": 0,  # 延迟容忍缓冲（0 = 严格选最低）
    },
)

# router_strategy/lowest_cost.py 核心逻辑
class LowestCostLoggingHandler(CustomLogger):
    """按成本排序 deployment"""

    def get_lowest_cost_deployments(self, deployments):
        """返回按成本升序排序的 deployment 列表"""
        from litellm import cost_per_token

        for d in deployments:
            model = d["litellm_params"]["model"]
            # 从 model_prices_and_context_window_backup.json 查价格
            input_cost, output_cost = cost_per_token(model=model, prompt_tokens=1, completion_tokens=1)
            d["_cost_per_token"] = (input_cost + output_cost) / 2

        return sorted(deployments, key=lambda d: d["_cost_per_token"])

from litellm.types.router import RoutingGroup

router = Router(
    model_list=[
        # GPT 系列
        {"model_name": "gpt-4", "litellm_params": {"model": "openai/gpt-4", ...}},
        {"model_name": "gpt-4", "litellm_params": {"model": "azure/gpt-4", ...}},
        {"model_name": "gpt-3.5-turbo", "litellm_params": {"model": "openai/gpt-3.5-turbo", ...}},

        # Claude 系列
        {"model_name": "claude-3-opus", "litellm_params": {"model": "anthropic/claude-3-opus", ...}},
        {"model_name": "claude-3-sonnet", "litellm_params": {"model": "anthropic/claude-3-sonnet", ...}},
    ],
    # 全局默认策略
    routing_strategy="simple-shuffle",

    # GPT 系列独立使用 latency-based 策略
    routing_groups=[
        RoutingGroup(
            group_name="gpt-models",
            model_names=["gpt-4", "gpt-3.5-turbo"],
            routing_strategy="latency-based-routing",
            routing_strategy_args={"ttl": 1800},
        ),
    ],
)

# router.py:6901 核心签名
async def async_function_with_retries(self, *args, **kwargs):
    """
    对同一个 deployment 进行重试。
    重试次数由 num_retries 和 retry_policy 共同决定。
    """
    original_function = kwargs.pop("original_function")
    num_retries = kwargs.pop("num_retries")

    # 重试策略解析
    model_group = kwargs.get("model")
    model_group_retry_policy = kwargs.pop("model_group_retry_policy", self.model_group_retry_policy)

    _metadata["attempted_retries"] = 0
    _metadata["max_retries"] = num_retries

    try:
        # 第一次调用
        response = await self.make_call(original_function, *args, **kwargs)
        response = add_retry_headers_to_response(
            response=response, attempted_retries=0, max_retries=None
        )
        return response
    except Exception as e:
        current_attempt = 0
        original_exception = e

        # 获取该异常类型允许的重试次数
        num_retries_from_policy = _get_num_retries_from_retry_policy(
            exception=e,
            retry_policy=self.retry_policy,
            model_group_retry_policy=model_group_retry_policy,
            model_group=model_group,
        )
        # 取两者中的最大值
        num_retries = max(num_retries, num_retries_from_policy)

        while current_attempt < num_retries:
            try:
                current_attempt += 1
                _metadata["attempted_retries"] = current_attempt

                # 等待 retry_after 秒
                if self.retry_after > 0:
                    await asyncio.sleep(self.retry_after)

                response = await self.make_call(original_function, *args, **kwargs)
                response = add_retry_headers_to_response(
                    response=response, attempted_retries=current_attempt, max_retries=num_retries
                )
                return response
            except Exception as retry_e:
                last_exception = retry_e

        # 所有重试耗尽
        raise last_exception

def get_num_retries_from_retry_policy(exception, retry_policy, model_group_retry_policy, model_group):
    """根据异常类型确定重试次数"""
    # 优先级: model_group_retry_policy > retry_policy > 默认值

    # 检查模型组定制策略
    if model_group in model_group_retry_policy:
        policy = model_group_retry_policy[model_group]
    else:
        policy = retry_policy or RetryPolicy()

    # 按异常类型匹配
    if isinstance(exception, litellm.RateLimitError):
        return policy.RateLimitErrorRetries or 0
    elif isinstance(exception, litellm.Timeout):
        return policy.TimeoutErrorRetries or 0
    elif isinstance(exception, litellm.ContentPolicyViolationError):
        return policy.ContentPolicyViolationErrorRetries or 0
    elif isinstance(exception, litellm.BadRequestError):
        return policy.BadRequestErrorRetries or 0
    ...

class RetryPolicy(BaseModel):
    TimeoutErrorRetries: Optional[int] = None
    RateLimitErrorRetries: Optional[int] = None
    BadRequestErrorRetries: Optional[int] = None
    ContentPolicyViolationErrorRetries: Optional[int] = None

# router_utils/fallback_event_handlers.py 核心逻辑
async def run_async_fallback(
    litellm_router,
    *args,
    fallback_model_group: List[str],
    original_model_group: str,
    original_exception: Exception,
    fallback_depth: int,
    max_fallbacks: int,
    **kwargs,
):
    """
    递归 fallback：尝试 fallback_model_group 列表中的每个模型组。

    参数:
        fallback_model_group: fallback 模型组列表，如 ["gpt-4", "gpt-3.5-turbo"]
        fallback_depth: 当前 fallback 深度（递归层级）
        max_fallbacks: 最大 fallback 深度
    """
    # ====== 终止条件：达到最大 fallback 深度 ======
    if fallback_depth >= max_fallbacks:
        raise original_exception

    error_from_fallbacks = original_exception

    for mg in fallback_model_group:
        # 跳过原始模型组（避免死循环）
        if mg == original_model_group:
            continue

        try:
            # 记录 fallback 日志
            kwargs = litellm_router.log_retry(kwargs=kwargs, e=original_exception)
            verbose_router_logger.info(f"Falling back to model_group = {mg}")

            # 更新 model 参数
            if isinstance(mg, str):
                kwargs["model"] = mg
            elif isinstance(mg, dict):
                kwargs.update(mg)

            # 递增 fallback 深度
            fallback_depth += 1
            kwargs["fallback_depth"] = fallback_depth
            kwargs["max_fallbacks"] = max_fallbacks

            # 递归调用 async_function_with_fallbacks（支持嵌套 fallback）
            response = await litellm_router.async_function_with_fallbacks(
                *args, **kwargs
            )

            # 成功：添加 fallback 响应头
            response = add_fallback_headers_to_response(
                response=response,
                attempted_fallbacks=fallback_depth,
            )
            return response

        except Exception as fallback_e:
            error_from_fallbacks = fallback_e
            verbose_router_logger.warning(
                f"Fallback to {mg} also failed: {fallback_e}"
            )

    # 所有 fallback 都失败
    raise error_from_fallbacks

# 格式 1: 字符串映射
fallbacks = [{"gpt-4": "gpt-3.5-turbo"}]
# → gpt-4 失败时 fallback 到 gpt-3.5-turbo

# 格式 2: 列表（多个 fallback 选项，按顺序尝试）
fallbacks = [{"gpt-4": ["gpt-3.5-turbo", "claude-sonnet"]}]
# → gpt-4 失败时先试 gpt-3.5-turbo，再试 claude-sonnet

# 格式 3: 通配符（全局兜底）
fallbacks = [{"*": "gpt-4o-mini"}]
# → 任何模型失败时都 fallback 到 gpt-4o-mini

# default_fallbacks 自动追加为通配符
# 如果设置了 default_fallbacks=["gpt-4o-mini"]
# 则实际 fallbacks 变为: [..., {"*": ["gpt-4o-mini"]}]

# router.py 中的处理逻辑
context_window_fallbacks = [
    {"gpt-3.5-turbo": "gpt-4-turbo"},    # 16K → 128K
    {"claude-sonnet": "claude-opus"},     # 200K → 200K（但更强）
]

content_policy_fallbacks = [
    {"azure-gpt-4": "openai-gpt-4"},     # Azure 审核过严 → OpenAI 直连
    {"azure-gpt-3.5": "openai-gpt-3.5-turbo"},
]

router = Router(
    model_list=[
        {"model_name": "gpt-4", "litellm_params": {"model": "azure/gpt-4", ...}, "weight": 3},
        {"model_name": "gpt-4", "litellm_params": {"model": "openai/gpt-4", ...}, "weight": 1},
    ],
    routing_strategy="simple-shuffle",
    enable_weighted_failover=True,  # 新增
)

def add_fallback_headers_to_response(response, attempted_fallbacks):
    """添加 fallback 次数到响应头"""
    response._hidden_params["attempted_fallbacks"] = attempted_fallbacks
    return response

def add_retry_headers_to_response(response, attempted_retries, max_retries):
    """添加重试次数到响应头"""
    response._hidden_params["attempted_retries"] = attempted_retries
    response._hidden_params["max_retries"] = max_retries
    return response

# router_utils/cooldown_handlers.py
DEFAULT_COOLDOWN_TIME_SECONDS = 60

async def _async_get_cooldown_deployments(cache, model_group, current_time):
    """获取当前处于 cooldown 状态的 deployment 列表"""
    key = f"lite:cooldowndeployments:{model_group}"
    cooldown_list = await cache.async_get_cache(key) or []

    # 过滤掉已过期的 cooldown
    active_cooldowns = []
    for item in cooldown_list:
        if isinstance(item, dict):
            cooldown_time = item.get("cooldown_time", DEFAULT_COOLDOWN_TIME_SECONDS)
            cooldown_start = item.get("cooldown_start_time", 0)
            if current_time - cooldown_start < cooldown_time:
                active_cooldowns.append(item["deployment_id"])
    return active_cooldowns

async def _set_cooldown_deployments(cache, model_group, deployment_id, cooldown_time):
    """将 deployment 加入 cooldown 列表"""
    key = f"lite:cooldowndeployments:{model_group}"
    cooldown_list = await cache.async_get_cache(key) or []

    cooldown_list.append({
        "deployment_id": deployment_id,
        "cooldown_time": cooldown_time,
        "cooldown_start_time": time.time(),
    })

    await cache.async_set_cache(
        key=key,
        value=cooldown_list,
        ttl=cooldown_time * 2,  # TTL 设为 cooldown_time 的 2 倍，保证过期后自动清理
    )

# 如果没显式设置 allowed_fails，根据 TPM 自动计算
if allowed_fails is None:
    # 默认: TPM 的 1%（最少 3 次）
    allowed_fails = max(3, int(model_tpm * 0.01))

# router.py: _select_deployment_async() 中的 pre-call 检查
if self.enable_pre_call_checks:
    # 1. 部署亲和性检查
    deployment_affinity_check = DeploymentAffinityCheck(...)
    healthy_deployments = deployment_affinity_check.filter(deployments)

    # 2. 模型限流检查
    rate_limit_check = ModelRateLimitingCheck(...)
    healthy_deployments = rate_limit_check.filter(healthy_deployments)

    # 3. Prompt Caching 可用性检查
    caching_check = PromptCachingDeploymentCheck(...)
    healthy_deployments = caching_check.filter(healthy_deployments)

class DeploymentHealthCache:
    """
    维护 deployment 健康状态:
    - healthy: 正常调度
    - unhealthy: 失败次数接近 allowed_fails 阈值
    - cooldown: 已进入冷却，不参与调度
    """

    def record_failure(self, deployment_id, model_group):
        """记录一次失败"""
        key = f"health:{model_group}:{deployment_id}"
        failures = self.cache.get(key, 0) + 1
        self.cache.set(key, failures, ttl=self.cooldown_time * 2)

        if failures >= self.allowed_fails:
            self.set_cooldown(deployment_id, model_group)

    def is_healthy(self, deployment_id, model_group):
        """检查 deployment 是否健康"""
        key = f"health:{model_group}:{deployment_id}"
        failures = self.cache.get(key, 0)
        return failures < self.allowed_fails

# ====== Deployment 定义 ======
# 每个 deployment 代表一个模型实例
{
    "model_name": "gpt-4",                    # 路由别名（同一 model_name = 同一 model_group）
    "litellm_params": {
        "model": "azure/gpt-4",               # 实际 Provider 模型名
        "api_key": "***",
        "api_base": "https://xxx.openai.azure.com/",
        "api_version": "2024-02-01",
    },
    "model_info": {
        "id": "unique-deployment-id",          # 唯一标识（用于 Cooldown 跟踪）
        "region": "us-east",
    },
    "tpm": 100000,                             # 每分钟 token 限额（usage-based 用）
    "rpm": 1000,                               # 每分钟请求限额
    "weight": 1,                               # 权重（simple-shuffle 用）
    "tags": ["production", "high-priority"],   # 标签（tag-based 用）
}

# ====== Fallback 格式 ======
fallbacks = [
    {"gpt-4": "gpt-3.5-turbo"},                # 字符串映射: 单个 fallback
    {"gpt-4": ["gpt-3.5-turbo", "claude"]},    # 列表: 多个 fallback，按顺序尝试
    {"*": "gpt-4o-mini"},                       # 通配符: 全局兜底
]

# ====== RetryPolicy ======
RetryPolicy(
    TimeoutErrorRetries=3,                      # 超时重试 3 次
    RateLimitErrorRetries=5,                    # 限流重试 5 次
    ContentPolicyViolationErrorRetries=0,       # 内容违规不重试
    BadRequestErrorRetries=0,                   # 请求错误不重试
)

# ❌ 错误：fallback 链过长，延迟累积
fallbacks = [
    {"gpt-4": "gpt-3.5-turbo"},
    {"gpt-3.5-turbo": "claude-sonnet"},
    {"claude-sonnet": "claude-haiku"},
    {"claude-haiku": "gpt-4o-mini"},
    {"gpt-4o-mini": "gemini-flash"},
]
# 如果 gpt-4 失败，可能需要 5 次 fallback 才能得到结果，延迟 = 5 × 平均响应时间

# ✅ 正确：扁平化 fallback 链
fallbacks = [
    {"gpt-4": ["gpt-3.5-turbo", "claude-sonnet"]},  # 先试 gpt-3.5，再试 claude
]
max_fallbacks = 2  # 限制最大深度

# 不要混用，它们针对不同的异常类型
context_window_fallbacks = [
    {"gpt-3.5-turbo": "gpt-4-turbo"}   # 只在 ContextWindowExceededError 时触发
]
content_policy_fallbacks = [
    {"azure-gpt-4": "openai-gpt-4"}     # 只在 ContentPolicyViolationError 时触发
]
# 两者可以同时设置，不会冲突

# 通过响应头监控
response = await router.acompletion(...)
attempted_fallbacks = response._hidden_params.get("attempted_fallbacks", 0)

if attempted_fallbacks > 0:
    logging.warning(f"Request required {attempted_fallbacks} fallbacks")
    # 如果某模型频繁触发 fallback，考虑:
    # 1. 调整 fallback 链顺序
    # 2. 检查底层 deployment 健康状态
    # 3. 增加 Cooldown 敏感度

# 如果同时设置了 fallbacks 和 default_fallbacks
router = Router(
    fallbacks=[{"gpt-4": "gpt-3.5-turbo"}],
    default_fallbacks=["gpt-4o-mini"],  # 自动追加为 {"*": ["gpt-4o-mini"]}
)
# 实际 fallback 链: gpt-4 → gpt-3.5-turbo → gpt-4o-mini
# 如果 gpt-3.5-turbo 也失败，会再 fallback 到 gpt-4o-mini
# 建议: 只用 fallbacks 或只用 default_fallbacks，不要混用

# allowed_fails 默认根据 TPM 自动计算，但如果没设置 TPM:
# allowed_fails = max(3, int(model_tpm * 0.01))
# 如果 tpm=0，allowed_fails=3，意味着 3 次失败就进入 60 秒冷却

# 建议: 显式设置 allowed_fails
router = Router(
    allowed_fails=5,        # 至少允许 5 次失败
    cooldown_time=30,       # 冷却时间不要太长
)

# enable_weighted_failover 仅 async 路径支持
# 使用 sync 路径时:
router.completion(model="gpt-4", ...)  # ❌ weighted failover 不生效

# 应该使用:
await router.acompletion(model="gpt-4", ...)  # ✅ weighted failover 生效

# 如果部署了多个 LiteLLM Router 实例，但没配置 Redis:
router = Router(model_list=..., redis_url=None)
# 每个实例独立维护 Cooldown 状态
# 可能导致: 实例 A 已将某 deployment 冷却，但实例 B 仍在调度到它

# 建议: 多实例环境必须配置 Redis
router = Router(
    model_list=...,
    redis_url="redis://redis-cluster:6379",
)

# 当触发 ContextWindowExceededError 时:
# - Retry: ❌ 不执行（因为重试无意义）
# - 常规 fallback: ❌ 不走 fallbacks 列表
# - Context Window Fallback: ✅ 走 context_window_fallbacks 列表

# 这意味着: 如果只设置了 fallbacks 而没有设置 context_window_fallbacks，
# 上下文超长的请求不会 fallback，直接报错。

# 建议: 如果模型可能遇到上下文超长，必须设置 context_window_fallbacks
router = Router(
    context_window_fallbacks=[
        {"gpt-3.5-turbo": "gpt-4-turbo"},
    ],
)