Диспетчеризация RAG: как парсить вопросы, выбирать стратегию чанкинга и тир модели

RAG-монолит мертв

В 2026 году иметь один RAG-пайплайн для всех запросов — роскошь. Фактологический вопрос «Какая высота Эвереста?» и аналитический «Почему растет инфляция в Аргентине?» требуют разной глубины извлечения, разного чанкинга и разных моделей. Один и тот же чанкинг (например, фиксированный по 500 токенов) для второго вопроса разорвет логические блоки. А держать GPT-4.1 на всех запросах — сжигать бюджет.

Решение — диспетчеризация (routing). Система анализирует сам запрос, профиль документа и на лету выбирает стратегию чанкинга и тир модели. В этой статье — готовый код и подход, который я использую в продакшене.

Шаг 1: Парсим вопрос — извлекаем признаки

Первый этап — классификатор намерений (intent classification). Он решает: простой вопрос или сложный, фактологический или рассуждение. Я использую маленькую модель (например, GPT-4o-mini или Claude Haiku) — она дешевая и быстрая.

Пример функции парсинга:

import openai

def parse_query(query: str) -> dict:
    """Извлекаем характеристики вопроса"""
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """Оцени запрос по шкалам:
- complexity: 1..5 (1 – простой факт, 5 – глубокий анализ)
- type: fact | analysis | summary | comparison
- needs_external_context: true/false
Ответь только JSON."""},
            {"role": "user", "content": query}
        ],
        temperature=0
    )
    return eval(response.choices[0].message.content)

Ключевой момент: классификатор не должен дорого стоить. На 10M запросов разница между Haiku и GPT-4.1 — тысячи долларов.

Шаг 2: Профиль документа — второй вектор решений

Даже зная сложность вопроса, нельзя выбрать чанкинг вслепую. Нужно знать, с каким документом работаем. Документы бывают:

• короткие (PDF на 2 страницы) — вообще можно не чанковать;
• длинные с четкой структурой (главы, секции) — семантический чанкинг по заголовкам;
• таблицы и код — специальный парсер.

Я создаю DocumentProfile — метаданные, которые вычисляются при индексации один раз:

from dataclasses import dataclass

@dataclass
class DocumentProfile:
    doc_id: str
    total_tokens: int
    has_headers: bool
    has_tables: bool
    avg_sentence_length: float
    language: str

Профиль сохраняется в векторной базе вместе с эмбеддингами. При запросе мы выбираем top-k документов и берем их профили.

Шаг 3: Выбор стратегии чанкинга

Теперь у нас есть parse_query и DocumentProfile. Пишем роутер чанкинга:

def choose_chunking_strategy(query_info: dict, profile: DocumentProfile) -> str:
    """Возвращаем имя стратегии: fixed, semantic, recursive"""
    if query_info['complexity'] <= 2 and profile.total_tokens < 1000:
        return 'fixed'  # фиксированный чанк по 512 токенов
    elif query_info['type'] == 'analysis' and profile.has_headers:
        return 'semantic'  # режем по заголовкам
    else:
        return 'recursive'  # рекурсивный с overlap

Зачем семантический чанкинг? Он сохраняет смысловые блоки. Для сложного вопроса «Сравните подходы к RAG в работах Lewis et al. и Shao et al.» — фиксированный чанк разорвет аргументацию, и модель выдаст кашу.

Шаг 4: Тир модели — экономим там, где можно

Использовать GPT-4.1 для ответа «столица Франции» — мазохизм. В 2026 году модели уже четко поделены на тиры:

Роутер модели:

TIER_MODELS = {
    'cheap': 'gpt-4o-mini',
    'medium': 'gpt-4o',
    'expensive': 'gpt-4.1'
}

def select_model(complexity: int) -> str:
    if complexity <= 2:
        return TIER_MODELS['cheap']
    elif complexity == 3:
        return TIER_MODELS['medium']
    else:
        return TIER_MODELS['expensive']

Шаг 5: Собираем диспетчер RAG

Финальный класс объединяет все части. Обратите внимание: мы НЕ вызываем LLM для каждого вопроса на классификацию — сначала пробуем эвристики (длина запроса, наличие «почему/сравни»). LLM — fallback.

class RAGRouter:
    def __init__(self, vector_store, chunkers: dict):
        self.vector_store = vector_store
        self.chunkers = chunkers  # {'fixed': ..., 'semantic': ..., 'recursive': ...}

    def route(self, query: str, top_k: int = 5):
        # 1. Быстрая эвристика
        if len(query) < 50 and query.endswith('?'):
            query_info = {'complexity': 1, 'type': 'fact'}
        else:
            query_info = parse_query(query)  # LLM call

        # 2. Получаем профили релевантных документов
        docs = self.vector_store.similarity_search(query, k=top_k)
        profiles = [doc.metadata['profile'] for doc in docs]

        # 3. Голосование профилей (берем самого частого или среднее)
        avg_profile = aggregate_profiles(profiles)

        # 4. Выбираем чанкинг и модель
        chunking = choose_chunking_strategy(query_info, avg_profile)
        model = select_model(query_info['complexity'])

        return {
            'chunking_strategy': chunking,
            'model': model,
            'query_info': query_info,
            'retrieved_docs': docs
        }

Конечно, это упрощение. В реальном продакшене добавляют кэш, батч-обработку профилей и A/B-тестирование стратегий. Но каркас именно такой.

5 типичных ошибок диспетчеризации

Набил шишек — делюсь. Вот что ломает диспетчер в проде:

• Неверная классификация из-за короткого запроса. «А Эверест?» — модель не понимает контекст. Решение: передавать историю диалога в парсер.
• Задержка на LLM-классификацию убивает UX. Если каждый запрос просит GPT-4o-mini — добавляем 300-500 мс. Используйте эвристики или маленькую модель с кэшем.
• Игнорирование профиля документа. Если документ — таблица, а чанкинг — семантический по заголовкам, чанкинг сломается. Подробнее — в разборе причин плохого поиска.
• Перекос стоимости. Думаете, cheap модель справится с анализом — но она галлюцинирует. Приходится перезапускать с expensive — теряем деньги и время. Лучше выбрать medium сразу.
• Отсутствие мониторинга. Без логов никогда не узнаете, что 30% сложных вопросов уходят в cheap-модель. Ставьте метрики: accuracy классификации, cost per query, latency.

Еще больше граблей — в статьях 10 критических ошибок RAG в продакшене и демо vs прод.

Мониторинг и адаптивная настройка

Диспетчеризация — не статичный скрипт. Вы собираете логи: какой вопрос, какой тир выбрали, какой ответ, какую оценку поставил пользователь (лайк/дизлайк). Раз в неделю переобучаете классификатор на новых данных.

Простой способ — сохранять (query, predicted_class, user_feedback) и дообучать маленькую модель лорой. Или использовать правила: если precision < 90% по классу «анализ» — повышаем порог сложности.

Не забудьте также следить за метриками самого RAG — об этом я писал в эксперименте про правильные данные, но неверный ответ.

Совет, который сэкономит вам месяцы: начните с бинарного разделения — факт vs анализ. Только когда наберете статистику, расширяйте до 5 уровней сложности и трех стратегий чанкинга. Итеративность — ваш главный инструмент.

В 2026 году, когда контекстные окна моделей перевалили за 10M токенов, кажется, что чанкинг не нужен. Но практика показывает: даже бесконечный контекст не гарантирует релевантности. Диспетчеризация — единственный способ масштабировать RAG без потери качества и бюджета. Ставьте роутер — и ваш RAG перестанет быть молотком, забивающим гвозди микроскопом.

Подписаться на канал