模型路由与选型：大小模型分层调度

# routing_benefit.py — 分层调度的收益估算
def routing_benefit(requests, prices, router) -> dict:
    """对比"全用旗舰模型"与"路由分层"的总成本。"""
    all_expensive = sum(
        estimate_cost(r, "flagship", prices) for r in requests)
    routed = sum(
        estimate_cost(r, router(r), prices) for r in requests)
    return {
        "before": all_expensive,
        "after": routed,
        "saving_pct": (1 - routed / all_expensive) * 100,
    }

# classify.py — 小模型复杂度分类器
def classify_complexity(query: str, small_model) -> str:
    """用便宜小模型判断复杂度，自带置信度。"""
    resp = small_model.classify(
        f"判断该任务复杂度（simple/medium/complex）：{query}")
    return resp["label"], resp.get("confidence", 0.5)

def route_by_confidence(query, small_model) -> str:
    """置信度高直接定级；置信度低保守升级。"""
    label, conf = classify_complexity(query, small_model)
    if conf >= 0.8:
        return MODEL_MAP[label]
    return MODEL_MAP["medium"]   # 不确定时取中间值

def static_route(query: str, feature: str, user_tier: str) -> str:
    """规则路由：按功能、用户等级、请求来源分流。"""
    if feature == "entity_extraction":
        return "gpt-4o-mini"        # 简单功能固定小模型
    if feature == "code_generation":
        return "gpt-4o"             # 复杂功能固定旗舰
    if user_tier == "free":
        return "gpt-4o-mini"        # 免费用户用小模型
    return "gpt-4o"

# model_gateway.py — 统一模型网关
class ModelGateway:
    def __init__(self):
        self.routes = {
            "small":   SmallModelClient(),
            "medium":  MediumModelClient(),
            "flagship": FlagshipModelClient(),
        }
        self.router = self._build_router()

    def complete(self, messages, ctx: dict = None) -> dict:
        ctx = ctx or {}
        model = self.router.route(messages, ctx)   # 动态选型
        try:
            resp = self.routes[model].complete(messages)
            return {"model": model, "response": resp, "ok": True}
        except ModelError:
            return self._fallback(messages, model)   # 降级

def build_routing_context(messages, ctx: dict) -> dict:
    """组装路由器需要的轻量特征，避免把全部上下文喂给路由器。"""
    last = messages[-1]
    return {
        "feature": ctx.get("feature"),
        "user_tier": ctx.get("user_tier"),
        "input_tokens": estimate_tokens(messages),
        "has_tools": bool(ctx.get("tools")),
        "query_text": last.get("content", "")[:500],   # 截断
        "requires_json": ctx.get("response_format") == "json",
    }

ROUTING_TABLE = {
    "simple": {
        "default": "gpt-4o-mini",
        "requires_json": "gpt-4o-mini",
        "requires_tools": "gpt-4o",          # 工具调用更强
    },
    "medium": {
        "default": "gpt-4o-mini",
        "requires_tools": "gpt-4o",
    },
    "complex": {
        "default": "gpt-4o",
    },
}

def apply_routing_table(label: str, ctx: dict) -> str:
    """按复杂度标签 + 上下文特征查表。"""
    rules = ROUTING_TABLE[label]
    if ctx.get("requires_tools"):
        return rules.get("requires_tools", rules["default"])
    if ctx.get("requires_json"):
        return rules.get("requires_json", rules["default"])
    return rules["default"]

# cascade.py — 级联调用
async def cascade(query: str, judge_fn) -> tuple[str, str]:
    """先小模型，质量不足升大模型。"""
    answer = await small_model.answer(query)
    if judge_fn(query, answer)["total"] >= 4:
        return answer, "small"
    answer = await flagship_model.answer(query)
    return answer, "flagship"

def cascade_expected_cost(rate_small_ok, cost_small, cost_judge, cost_large) -> float:
    """级联的期望成本 = 小模型通过的部分 + 升级的部分。"""
    cost_when_small = cost_small + cost_judge
    cost_when_large = cost_small + cost_judge + cost_large
    return rate_small_ok * cost_when_small + (1 - rate_small_ok) * cost_when_large

def plain_cost(cost_large):
    """不级联：一律旗舰。"""
    return cost_large

# fallback.py — 降级链
FALLBACK_CHAIN = {
    "flagship": ["medium", "small"],    # 旗舰失败 → 中 → 小
    "medium": ["small", "flagship"],    # 中失败 → 小 → 旗舰
    "small": ["medium", "flagship"],
}

async def complete_with_fallback(messages, preferred, chain=None):
    """按降级链逐级尝试。"""
    chain = chain or FALLBACK_CHAIN.get(preferred, [])
    for model in [preferred] + chain:
        try:
            resp = await clients[model].complete(messages)
            return {"model": model, "response": resp}
        except (TimeoutError, RateLimitError):
            log_fallback(preferred, model)   # 记录降级
            continue
    raise AllModelsUnavailable()

def track_degradation(entry: dict):
    """记录降级事件：触发原因、降级路径、影响。"""
    push_metric("model_gateway.fallback", 1,
                {"from": entry["preferred"], "to": entry["used"],
                 "reason": entry["reason"]})
    if entry["preferred"] != entry["used"]:
        push_metric("model_gateway.degradation_rate",
                    entry["used"] != entry["preferred"])

# selection_matrix.py — 模型性价比评分
def model_score(model, weights: dict) -> float:
    """按业务权重给模型打分：质量×w1 + 成本×w2 + 延迟×w3。"""
    return (
        weights["quality"] * model["quality"] / 10
        - weights["cost"] * model["cost_per_k"]
        - weights["latency"] * model["p95_ms"] / 1000
    )

def select_best_model(candidates, weights):
    """按加权得分选最优模型。"""
    return max(candidates, key=lambda m: model_score(m, weights))

def route_accuracy(golden, router, labels) -> float:
    """路由定级与人工标注的准确率。"""
    correct = sum(router(g["query"]) == labels[g["id"]]
                  for g in golden.cases)
    return correct / len(golden.cases)

def tune_threshold(cases, router, judge_fn, search_space=(0.5, 0.95)):
    """网格搜索最优置信度阈值：权衡成本与质量。"""
    best = None
    for threshold in [round(x, 2) for x in frange(*search_space)]:
        cost, quality = simulate(router, threshold, cases, judge_fn)
        score = quality - 0.5 * cost   # 业务自定义权衡
        if best is None or score > best["score"]:
            best = {"threshold": threshold, "score": score,
                    "cost": cost, "quality": quality}
    return best

def canary_route(query, canary_pct=0.05):
    """按百分比放流量到新路由策略，对比新旧质量。"""
    if hash(query) % 100 < canary_pct * 100:
        return new_router.route(query), "canary"
    return old_router.route(query), "control"

客服 Agent 请求构成（每日 10 万条）：
45% 简单（FAQ、订单状态、意图识别）
35% 中等（操作指引、多轮对话）
20% 复杂（纠纷处理、政策解释、工具调用）
目标：成本降 50% 以上，质量不降

# support_router.py — 客服分层调度
class SupportRouter:
    def __init__(self, small, medium, flagship):
        self.small = small
        self.medium = medium
        self.flagship = flagship

    def route(self, query: str, ctx: dict) -> str:
        feature = ctx.get("feature")
        if feature in {"intent", "slot_fill", "order_status"}:
            return "small"                       # 简单功能
        if feature in {"guide", "multi_turn"}:
            return "medium"
        if feature in {"dispute", "tool_call"}:
            return "flagship"                    # 复杂走旗舰
        # 默认：用分类器动态分级 + 级联兜底
        label = classify_complexity(query, self.small)[0]
        return {"simple": "small", "medium": "medium",
                "complex": "flagship"}[label]

- 路由准确率：抽样 200 条人工标注对比 ≥ 90%
- 质量：路由前后评测集分数下降 ≤ 3%
- 成本：周账单对比，节省率 ≥ 40%
- 降级率：< 1%，且降级后质量采样正常
- 延迟：P95 不超预算（路由本身开销 < 50ms）