Реализация маршрутизации запросов между LLM-провайдерами (LLM Router)
LLM Router — прослойка, которая направляет запросы к оптимальному провайдеру на основе характеристик запроса: сложности, типа задачи, требований к latency, стоимости. Простые классификации — GPT-4o-mini. Сложный code generation — Claude Opus. Realtime чат — Groq. Это снижает стоимость в 3–5× без потери качества.
Архитектура роутера
from anthropic import Anthropic
from openai import OpenAI
from groq import Groq
from dataclasses import dataclass
from typing import Callable, Optional
import re
@dataclass
class RoutingRule:
name: str
condition: Callable[[str, dict], bool]
provider: str
model: str
reason: str
class LLMRouter:
def __init__(self):
self.openai_client = OpenAI()
self.anthropic_client = Anthropic()
self.groq_client = Groq()
self.rules: list[RoutingRule] = [
# Realtime/простые запросы → Groq (быстро и дёшево)
RoutingRule(
name="simple_chat",
condition=lambda text, meta: len(text) < 200 and not meta.get("complex"),
provider="groq",
model="llama-3.1-8b-instant",
reason="Short simple query — using fast inference",
),
# Code generation → Claude (лучшее качество кода)
RoutingRule(
name="code_generation",
condition=lambda text, meta: self._is_code_task(text),
provider="anthropic",
model="claude-sonnet-4-5",
reason="Code task — Claude performs better",
),
# Reasoning/математика → o3-mini
RoutingRule(
name="reasoning",
condition=lambda text, meta: self._is_reasoning_task(text),
provider="openai",
model="o3-mini",
reason="Reasoning task — using o3-mini",
),
# Default — GPT-4o
RoutingRule(
name="default",
condition=lambda text, meta: True,
provider="openai",
model="gpt-4o",
reason="Default routing",
),
]
def _is_code_task(self, text: str) -> bool:
code_keywords = [
"напиши код", "реализуй", "функция", "класс", "алгоритм",
"python", "javascript", "sql", "рефакторинг", "отладь"
]
return any(kw in text.lower() for kw in code_keywords)
def _is_reasoning_task(self, text: str) -> bool:
reasoning_keywords = [
"докажи", "вычисли", "оптимизируй", "найди оптимальное",
"математически", "логически следует", "минимизируй"
]
return any(kw in text.lower() for kw in reasoning_keywords)
def route(self, text: str, meta: dict = None) -> RoutingRule:
"""Определяет провайдера для запроса"""
meta = meta or {}
for rule in self.rules:
if rule.condition(text, meta):
return rule
return self.rules[-1] # Default
def complete(
self,
messages: list[dict],
system: str = None,
**kwargs
) -> str:
"""Выполняет запрос через оптимального провайдера"""
user_message = messages[-1]["content"] if messages else ""
rule = self.route(user_message)
print(f"[Router] {rule.name} → {rule.provider}/{rule.model}: {rule.reason}")
if rule.provider == "anthropic":
return self._call_anthropic(messages, rule.model, system, **kwargs)
elif rule.provider == "openai":
return self._call_openai(messages, rule.model, system, **kwargs)
elif rule.provider == "groq":
return self._call_groq(messages, rule.model, system, **kwargs)
def _call_anthropic(self, messages, model, system, **kwargs) -> str:
kwargs.pop("temperature", None) if "o" in model else None
response = self.anthropic_client.messages.create(
model=model,
max_tokens=kwargs.get("max_tokens", 2048),
system=system or "",
messages=messages,
temperature=kwargs.get("temperature", 0.1),
)
return response.content[0].text
def _call_openai(self, messages, model, system, **kwargs) -> str:
all_messages = []
if system:
all_messages.append({"role": "system", "content": system})
all_messages.extend(messages)
params = {"model": model, "messages": all_messages}
if "o1" not in model and "o3" not in model:
params["temperature"] = kwargs.get("temperature", 0.1)
response = self.openai_client.chat.completions.create(**params)
return response.choices[0].message.content
def _call_groq(self, messages, model, system, **kwargs) -> str:
all_messages = []
if system:
all_messages.append({"role": "system", "content": system})
all_messages.extend(messages)
response = self.groq_client.chat.completions.create(
model=model,
messages=all_messages,
temperature=kwargs.get("temperature", 0),
)
return response.choices[0].message.content
Семантический роутер (ML-based)
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticRouter:
"""Роутинг на основе семантического сходства с примерами задач"""
def __init__(self):
self.model = SentenceTransformer("BAAI/bge-small-en-v1.5")
# Примеры задач для каждого провайдера
self.routes = {
"groq_fast": [
"what time is it", "hello", "who are you",
"привет", "простой вопрос", "краткий ответ"
],
"anthropic_code": [
"write a python function", "debug this code",
"refactor this class", "implement algorithm",
"код на python", "напиши функцию"
],
"openai_reasoning": [
"solve this math problem", "prove this theorem",
"optimize this algorithm", "logical deduction",
"математическая задача", "докажи утверждение"
],
}
# Предвычисляем эмбеддинги
self.route_embeddings = {}
for route, examples in self.routes.items():
self.route_embeddings[route] = self.model.encode(examples)
def route(self, query: str) -> str:
query_embedding = self.model.encode(query)
best_route = "openai_default"
best_score = 0.5 # Минимальный порог
for route, embeddings in self.route_embeddings.items():
scores = np.dot(embeddings, query_embedding) / (
np.linalg.norm(embeddings, axis=1) * np.linalg.norm(query_embedding)
)
max_score = scores.max()
if max_score > best_score:
best_score = max_score
best_route = route
return best_route
Мониторинг и аналитика роутинга
from collections import defaultdict
import time
class RouterWithMetrics:
def __init__(self, router: LLMRouter):
self.router = router
self.stats = defaultdict(lambda: {"count": 0, "total_time": 0, "total_cost": 0})
def complete(self, messages: list[dict], **kwargs) -> str:
user_message = messages[-1]["content"]
rule = self.router.route(user_message)
start = time.time()
result = self.router.complete(messages, **kwargs)
elapsed = time.time() - start
self.stats[rule.name]["count"] += 1
self.stats[rule.name]["total_time"] += elapsed
return result
def report(self) -> dict:
return {
route: {
"count": stats["count"],
"avg_time": stats["total_time"] / max(stats["count"], 1),
}
for route, stats in self.stats.items()
}
Практический кейс: SaaS платформа
Профиль запросов: 60% — простые вопросы (FAQ, статусы), 25% — аналитика и суммаризация, 15% — генерация кода.
До роутинга: все запросы → GPT-4o = $450/месяц.
После роутинга:
- Простые → Groq Llama 8B: $12/месяц
- Аналитика → GPT-4o-mini: $45/месяц
- Код → Claude Sonnet: $67/месяц
- Итого: $124/месяц (-72%)
Качество при A/B тестировании: сопоставимое (98% пользователей не заметили разницу).
Сроки
- Rule-based роутер: 2–3 дня
- Семантический роутер с ML: 1 неделя
- Мониторинг + A/B тестирование: 1 неделя