模型路由实战 按任务难度分级调用LLM成本直降40%

一、先算账：你的钱都花在哪了

很多Agent项目账单爆炸，不是因为调用次数多，而是大材小用：90%的请求只是"总结一句话""提取个日期"，却统统发给旗舰模型。先看一组典型价格对比：

假设日请求10万次、平均每次2000 token：如果全部用旗舰模型，日成本约$360；按难度分级后，约80%流量落到轻量/中端模型，日成本可压到$100以内，直降60~70%。这就是模型路由的价值。

二、任务难度三级体系

先把任务分成三档，路由才有依据：

三、路由分类器实现

路由分类器用特征规则 + 轻量模型双通道，成本几乎为零：

import re

class TaskRouter:
    """按任务难度路由到不同模型"""
    def __init__(self, light_model, mid_model, flagship_model):
        self.light = light_model      # 轻量模型
        self.mid = mid_model          # 中端模型
        self.flag = flagship_model    # 旗舰模型
        self.model_map = {"L1": self.light, "L2": self.mid, "L3": self.flag}

    # ---- 规则特征：零成本初判 ----
    def _rule_level(self, task: str, tools: list) -> str:
        # 需要调用工具 -> 至少 L2
        if tools:
            return "L2"
        # 长文本生成 / 代码 -> L3
        if re.search(r"重构|架构|设计|调试|优化|写一段.{6,}代码", task):
            return "L3"
        # 短输出、单步 -> L1
        if re.search(r"提取|分类|判断|翻译|总结.{0,20}$", task) and len(task) < 60:
            return "L1"
        return "L2"  # 默认中等，保守

    def route(self, task: str, tools: list = None) -> str:
        rule_level = self._rule_level(task, tools or [])
        # 规则判断为 L1/L3 时直接采纳（零成本）
        if rule_level in ("L1", "L3"):
            return rule_level
        # 规则拿不准（L2）时，用轻量模型做最终裁决
        prompt = f"""判断任务难度，只输出 L1 / L2 / L3：
L1=简单(分类/抽取/短输出)，L2=中等(工具调用/结构化输出)，L3=困难(多步推理/代码/长文)

任务：{task}
难度："""
        level = self.light.chat(prompt, max_tokens=3, temperature=0).strip().upper()
        return level if level in ("L1", "L2", "L3") else "L2"

    def call(self, task: str, tools: list = None, **kwargs):
        level = self.route(task, tools)
        model = self.model_map[level]
        return model.chat(task, **kwargs), level

要点：规则能拍板的场景绝不调用模型；只有规则模糊的任务才花一次轻量模型调用（每次不到0.01美分）。

四、分级调用与降级回退

路由之后，稳定性比省钱更重要。生产环境必须带超时降级与失败升级：

import time

class ResilientRouter(TaskRouter):
    def __init__(self, *args, timeout=10, **kwargs):
        super().__init__(*args, **kwargs)
        self.timeout = timeout
        self.stats = {"L1": 0, "L2": 0, "L3": 0, "upgraded": 0, "downgraded": 0}

    def call(self, task: str, tools: list = None, **kwargs):
        level = self.route(task, tools)
        self.stats[level] += 1
        try:
            return self._invoke_with_timeout(level, task, **kwargs), level
        except TimeoutError:
            # 低档超时 -> 升级一档重试（只升不降）
            if level != "L3":
                self.stats["upgraded"] += 1
                upper = {"L1": "L2", "L2": "L3"}[level]
                return self._invoke_with_timeout(upper, task, **kwargs), upper
            raise
        except Exception:
            # 模型报错 -> 降级到中端兜底，保证可用性
            self.stats["downgraded"] += 1
            return self._invoke_with_timeout("L2", task, **kwargs), "L2"

    def _invoke_with_timeout(self, level: str, task: str, **kwargs):
        model = self.model_map[level]
        start = time.time()
        result = model.chat(task, timeout=self.timeout, **kwargs)
        return result

降级策略铁律：轻量模型超时→升中端重试；中端报错→旗舰兜底可以，但旗舰报错→直接返回友好错误，绝不无限重试烧钱。

五、成本核算与监控

没有成本看板的路由等于没做。每次调用都要记账：

import sqlite3
import json

class CostTracker:
    PRICES = {  # 每百万token美元
        "light": {"in": 0.15, "out": 0.6},
        "mid": {"in": 3.0, "out": 15.0},
        "flag": {"in": 15.0, "out": 75.0},
    }

    def __init__(self, db_path="costs.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""CREATE TABLE IF NOT EXISTS calls(
            ts TEXT, task_hash TEXT, level TEXT, model TEXT,
            in_tokens INT, out_tokens INT, cost REAL, latency REAL)""")

    def record(self, level, model, in_tokens, out_tokens, latency):
        p = self.PRICES[level]
        cost = (in_tokens / 1e6) * p["in"] + (out_tokens / 1e6) * p["out"]
        self.conn.execute(
            "INSERT INTO calls VALUES(?,?,?,?,?,?,?,?)",
            (time.strftime("%Y-%m-%d %H:%M:%S"), "", level, model,
             in_tokens, out_tokens, round(cost, 6), round(latency, 3)))
        self.conn.commit()
        return cost

    def daily_report(self):
        row = self.conn.execute(
            "SELECT level, COUNT(*), SUM(cost) FROM calls "
            "WHERE ts >= date('now','-1 day') GROUP BY level").fetchall()
        total = sum(r[2] for r in row)
        return {"by_level": row, "total_cost": round(total, 4)}

建议设置日预算阈值告警：当日成本超过预估的120%时自动降级全部流量到中端模型，第二天再复盘调整路由规则。

六、落地效果与避坑指南

**三个避坑提醒**：第一，L3任务别省钱，复杂推理用轻量模型会陷入"重试地狱"，反而更贵；第二，路由规则要随业务迭代，每月用日志分析一次误路由比例；第三，先接成本监控再上路由，否则省了多少钱你根本说不清。模型路由的本质不是"用便宜的模型"，而是"让每个任务用恰好够用的模型"。