Agent Lab Journal
Самое важное из мира AI — в канале MAX AgentLabОтдельные разборы, инструменты и практические схемы Читайте Agent Lab в Telegram Разборы, кейсы и новости о практических AI-агентах без лишней воды.

Как построить LLM-роутер и проверить экономию без потери качества

Как построить LLM-роутер и проверить экономию без потери качества
Temporary fallback cover; replace in editorial pass.

Уровень: продвинутый · Чтение: до 12 минут · Результат: работающий роутер с лимитами бюджета, эскалацией и журналом затрат, проверенный на фиксированном наборе запросов против базовой стратегии

Зачем это нужно и в чём подвох

Типичная картина: все запросы, от «переведи слово» до «найди ошибку в миграции», уходят одной самой сильной модели. Это надёжно, но дорого. Очевидное решение — LLM-роутер: дешёвая модель берёт простые запросы, сложные отправляются сильной.

Подвох в том, как роутер решает, что дешёвого ответа достаточно. Самый популярный способ — спросить у модели её уверенность и выйти раньше, если она высокая. Этот ранний выход ненадёжен: модель может быть уверенной и ошибаться, и тогда роутер закрепляет неверный ответ, а сэкономленные деньги превращаются в незаметную деградацию качества. Поэтому в этой статье решение об эскалации принимает внешняя проверка: валидация схемы, обязательные поля, прогон тестов. Экономию мы признаём только после парного сравнения с базовой стратегией на заранее зафиксированном наборе запросов.

Все цены, пороги и примеры запросов ниже — шаблоны. Подставьте тарифы своего провайдера и собственные данные. Цифр «у нас получилось X% экономии» здесь нет намеренно: их нужно получить на своём наборе.

Архитектура

  1. Классификатор уровня. Дешёвые детерминированные правила (длина, ключевые слова) выбирают стартовый уровень: cheap или strong.
  2. Внешняя проверка во время работы. Это проверка, которая не знает эталонного ответа: парсится ли JSON, есть ли нужные ключи, проходят ли тесты. Самооценка модели в решение не входит.
  3. Эскалация. Если проверка не пройдена, запрос уходит на следующий уровень, но только если позволяет бюджет.
  4. Бюджет. Ограничены и стоимость одного запроса, и общая сумма. Когда лимит исчерпан, роутер возвращает явный статус. Тихой подмены на более слабую модель нет.
  5. Журнал затрат. Каждый вызов модели записывается отдельной строкой JSONL: модель, токены, стоимость, результат проверки, причина эскалации.

Главное различие: runtime-проверка работает в продакшене и не знает правильного ответа, а оценка (grade) использует эталон и существует только в эксперименте. Если эталон попадёт в runtime-проверку, результаты эксперимента окажутся завышенными.

Шаг 1. Окружение

Все команды создают файлы только в новой папке проекта и ничего не отправляют наружу:

mkdir llm-router && cd llm-router
python3 -m venv .venv
. .venv/bin/activate
mkdir -p cache runs data

Скрипты используют только стандартную библиотеку Python. SDK провайдера подключите сами в функции provider_call.

Шаг 2. Конфигурация

Файл router_config.json. Имена моделей и цены — заглушки, замените их актуальными значениями из прайса вашего провайдера.

{
  "tiers": { "cheap": "MODEL_CHEAP", "strong": "MODEL_STRONG" },
  "models": {
    "MODEL_CHEAP":  { "usd_per_mtok_in": 0.0, "usd_per_mtok_out": 0.0 },
    "MODEL_STRONG": { "usd_per_mtok_in": 0.0, "usd_per_mtok_out": 0.0 }
  },
  "rules": {
    "long_prompt_chars": 4000,
    "strong_keywords": ["докажи", "миграц", "race condition", "архитектур"]
  },
  "max_output_tokens": 1024,
  "budget": { "total_usd": 5.0, "per_request_usd": 0.05 },
  "log_path": "runs/cost_log.jsonl"
}

Пока цены нулевые, бюджет не сработает, и это заметно сразу. Так и задумано: заглушку нельзя случайно принять за реальные данные.

Шаг 3. Роутер

Файл router.py:

"""LLM-роутер: выбор уровня, внешняя проверка, эскалация, бюджет, журнал затрат."""
import hashlib
import json
import time
import uuid
from pathlib import Path

CONFIG = json.loads(Path("router_config.json").read_text(encoding="utf-8"))
CACHE_DIR = Path("cache")


class BudgetExceeded(Exception):
    pass


class Budget:
    def __init__(self, total_usd, per_request_usd):
        self.total = total_usd
        self.per_request = per_request_usd
        self.spent = 0.0

    def reserve(self, worst_case):
        if self.spent + worst_case > self.total:
            raise BudgetExceeded("total budget")

    def charge(self, cost):
        self.spent += cost


def estimate_cost(model, tokens_in, tokens_out):
    p = CONFIG["models"][model]
    return tokens_in / 1e6 * p["usd_per_mtok_in"] + tokens_out / 1e6 * p["usd_per_mtok_out"]


def approx_tokens(text):
    # Грубая оценка только для резерва бюджета; фактические токены берём из ответа провайдера.
    return len(text) // 3 + 1


def provider_call(model, prompt, max_tokens):
    """Верните (text, tokens_in, tokens_out) из usage вашего провайдера."""
    raise NotImplementedError("подключите SDK провайдера")


def call_model(model, prompt, max_tokens):
    key = hashlib.sha256(f"{model}\n{max_tokens}\n{prompt}".encode()).hexdigest()
    path = CACHE_DIR / f"{key}.json"
    if path.exists():
        return tuple(json.loads(path.read_text(encoding="utf-8")))
    result = provider_call(model, prompt, max_tokens)
    path.write_text(json.dumps(result, ensure_ascii=False), encoding="utf-8")
    return result


def pick_tier(prompt):
    rules = CONFIG["rules"]
    if len(prompt) > rules["long_prompt_chars"]:
        return "strong"
    if any(k in prompt.lower() for k in rules["strong_keywords"]):
        return "strong"
    return "cheap"


def runtime_check(query, answer):
    """Проверка без эталона: только то, что доступно в продакшене."""
    check = query.get("runtime_check", {"type": "none"})
    if check["type"] == "json_keys":
        try:
            data = json.loads(answer)
        except json.JSONDecodeError:
            return False, "invalid_json"
        missing = [k for k in check["keys"] if k not in data]
        return (not missing), ("missing:" + ",".join(missing) if missing else "ok")
    if check["type"] == "non_empty":
        return bool(answer.strip()), "ok" if answer.strip() else "empty"
    # Нечего проверить: эскалировать нечем, честно помечаем.
    return True, "unchecked"


def log_event(record):
    with open(CONFIG["log_path"], "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")


def route(query, budget, force_tier=None, strategy="router"):
    request_id = str(uuid.uuid4())
    start = force_tier or pick_tier(query["prompt"])
    chain = ["cheap", "strong"] if start == "cheap" else ["strong"]
    spent, answer, final_model, status = 0.0, None, None, "ok"

    for step, tier in enumerate(chain):
        model = CONFIG["tiers"][tier]
        worst = estimate_cost(model, approx_tokens(query["prompt"]), CONFIG["max_output_tokens"])
        if spent + worst > budget.per_request:
            status = "per_request_cap"
            break
        try:
            budget.reserve(worst)
        except BudgetExceeded:
            status = "total_budget"
            break

        text, tin, tout = call_model(model, query["prompt"], CONFIG["max_output_tokens"])
        cost = estimate_cost(model, tin, tout)
        budget.charge(cost)
        spent += cost
        ok, reason = runtime_check(query, text)
        answer, final_model = text, model

        log_event({
            "ts": time.time(), "request_id": request_id, "query_id": query["id"],
            "strategy": strategy, "step": step, "tier": tier, "model": model,
            "tokens_in": tin, "tokens_out": tout, "cost_usd": round(cost, 8),
            "check_ok": ok, "check_reason": reason,
            "escalate": (not ok) and step < len(chain) - 1,
        })
        if ok:
            break
        status = "check_failed"

    return {"answer": answer, "model": final_model, "cost_usd": spent,
            "status": status, "escalated": final_model == CONFIG["tiers"]["strong"] and start == "cheap"}

Что важно в этом коде:

Шаг 4. Фиксированный набор запросов

Файл data/test.jsonl, по одной записи на строку. Ниже две записи-примера, а не реальный датасет:

{"id": "q001", "prompt": "Верни JSON с ключами city и country для: Казань", "runtime_check": {"type": "json_keys", "keys": ["city", "country"]}, "grade": {"type": "json_equals", "value": {"city": "Казань", "country": "Россия"}}}
{"id": "q002", "prompt": "Найди race condition в этом коде: ...", "runtime_check": {"type": "non_empty"}, "grade": {"type": "contains_all", "values": ["lock"]}}

Правила формирования набора:

Шаг 5. Прогон обеих стратегий

Файл evaluate.py. Здесь грейдер видит эталон, а роутер нет:

import argparse
import json
from pathlib import Path

from router import CONFIG, Budget, route


def grade(query, answer):
    if answer is None:
        return False
    g = query["grade"]
    if g["type"] == "json_equals":
        try:
            return json.loads(answer) == g["value"]
        except json.JSONDecodeError:
            return False
    if g["type"] == "contains_all":
        return all(s.lower() in answer.lower() for s in g["values"])
    raise ValueError(f"unknown grade type: {g['type']}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", required=True)
    ap.add_argument("--strategy", choices=["baseline", "router"], required=True)
    ap.add_argument("--out", required=True)
    args = ap.parse_args()

    b = CONFIG["budget"]
    budget = Budget(b["total_usd"], b["per_request_usd"])
    force = "strong" if args.strategy == "baseline" else None

    with open(args.out, "w", encoding="utf-8") as out:
        for line in Path(args.data).read_text(encoding="utf-8").splitlines():
            q = json.loads(line)
            r = route(q, budget, force_tier=force, strategy=args.strategy)
            out.write(json.dumps({
                "id": q["id"], "passed": grade(q, r["answer"]),
                "cost_usd": r["cost_usd"], "model": r["model"],
                "status": r["status"], "escalated": r["escalated"],
            }, ensure_ascii=False) + "\n")


if __name__ == "__main__":
    main()

Запуск. Сначала идёт baseline, чтобы ответы сильной модели попали в кэш:

python3 evaluate.py --data data/test.jsonl --strategy baseline --out runs/baseline.jsonl
python3 evaluate.py --data data/test.jsonl --strategy router   --out runs/router.jsonl

Учтите одну тонкость. Из-за общего кэша второй прогон не платит за повторные вызовы сильной модели, но в cost_usd стоимость всё равно записывается по токенам. Сравнение остаётся честным, а реальный счёт на эксперимент меньше.

Шаг 6. Парное сравнение и решение

Порог допустимых регрессий задаётся до того, как вы увидите результаты. Файл compare.py:

import json
import sys


def load(path):
    with open(path, encoding="utf-8") as f:
        return {r["id"]: r for r in map(json.loads, f)}


base, rout = load(sys.argv[1]), load(sys.argv[2])
max_regressions = int(sys.argv[3])
assert base.keys() == rout.keys(), "наборы запросов различаются"

both = sum(base[i]["passed"] and rout[i]["passed"] for i in base)
regressed = [i for i in base if base[i]["passed"] and not rout[i]["passed"]]
improved = [i for i in base if not base[i]["passed"] and rout[i]["passed"]]
cost_b = sum(r["cost_usd"] for r in base.values())
cost_r = sum(r["cost_usd"] for r in rout.values())
escalated = sum(r["escalated"] for r in rout.values())
capped = [i for i, r in rout.items() if r["status"] in ("per_request_cap", "total_budget")]

print(f"запросов: {len(base)}, оба прошли: {both}")
print(f"регрессии (baseline да, router нет): {len(regressed)} {regressed}")
print(f"улучшения (baseline нет, router да): {len(improved)} {improved}")
print(f"стоимость baseline: {cost_b:.4f} USD, router: {cost_r:.4f} USD")
if cost_b > 0:
    print(f"экономия: {100 * (1 - cost_r / cost_b):.1f}%")
print(f"эскалаций: {escalated}, упёрлись в бюджет: {capped}")
ok = len(regressed) <= max_regressions and not capped
print("РЕШЕНИЕ:", "принять" if ok else "не принимать")
python3 compare.py runs/baseline.jsonl runs/router.jsonl 0

Смотрите на расходящиеся пары, а не только на итоговый процент прохождения. Одинаковый общий процент может скрывать пять регрессий и пять случайных улучшений. Каждую регрессию из списка откройте вручную: обычно по ней видно, какой проверки не хватило в runtime_check.

Проверка результата

Роутер готов, когда выполнены все пункты:

Перед повторным полным прогоном удалите старый журнал (rm runs/cost_log.jsonl), иначе суммы задвоятся.

Типовые ошибки

Ограничения

Что дальше

Похожие практические разборы собраны в руководствах, определения терминов — в глоссарии. Следующий логичный шаг — подключить журнал затрат к ежедневному отчёту и раз в неделю повторять парное сравнение на свежей выборке трафика.