Продвинутый RAG: диспетчеризация вопросов с профилированием документа, стратегией чанков и аудитом на основе GPT-4.1

В предыдущей статье мы разобрали, как диспетчеризация спасает RAG от монолитного безумия: парсили вопрос, определяли сложность и выбирали модель с чанкингом. Но практика показала — даже этого мало. Вы можете идеально классифицировать запрос, но если документ — это сборная солянка из таблиц, кода и прозы, фиксированная стратегия чанкинга размажет смысл по токенам. А как проверить, что ответ не галлюцинация, не тратя при этом кучу денег?

Ответ — профилирование документа и аудит на базе GPT-4.1. В этой статье я покажу, как заставить RAG не просто искать, а думать о том, где искать и проверять, что нашел. Без воды, с кодом, который работает в проде.

Профилирование документа: когда вы перестаете гадать на кофейной гуще

Допустим, у вас в базе два документа: 200-страничный договор с вложенными таблицами и короткая инструкция по эксплуатации. Один и тот же семантический чанкинг разобьет договор на логические блоки, а инструкцию разорвет на бессмысленные куски. И наоборот — рекурсивный чанкинг отлично подойдет для инструкции, но убьет таблицы. Решение — перед обработкой документа снимать его профиль.

Вот как выглядит функция профилирования:

import re
import tiktoken

def profile_document(text: str) -> dict:
    """Извлекаем профиль документа для выбора стратегии"""
    enc = tiktoken.encoding_for_model("gpt-4.1")
    tokens = enc.encode(text)
    lines = text.split('\n')
    headings = [l for l in lines if l.startswith('#') or re.match(r'^[A-ZА-Я][^.!?]*$', l.strip())]
    tables = sum(1 for l in lines if '|' in l and l.count('|') >= 3)
    code_blocks = len(re.findall(r'```(?:\w+)?', text))
    sentences = re.split(r'[.!?]+', text)
    avg_sent_len = sum(len(s.split()) for s in sentences if s) / max(len(sentences), 1)
    
    # Определяем доминирующий тип
    type_ = 'text'
    if tables / max(len(lines), 1) > 0.05:
        type_ = 'mixed' if code_blocks else 'table'
    elif code_blocks > 3 and len(code_blocks) > len(headings):
        type_ = 'code'
    elif avg_sent_len > 40:
        type_ = 'legal'  # длинные предложения — юридические/научные
    
    return {
        'total_tokens': len(tokens),
        'total_lines': len(lines),
        'headings_count': len(headings),
        'tables_density': tables / max(len(lines), 1),
        'code_blocks': code_blocks,
        'avg_sentence_length': round(avg_sent_len, 1),
        'dominant_type': type_,
        'structure_score': min(1.0, len(headings) / (len(lines) / 20))
    }

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

Диспетчер на стероидах: парсинг вопроса + профиль документа

Теперь у нас есть два источника: профиль вопроса (из предыдущей статьи — диспетчеризация RAG) и профиль документа. Диспетчер должен их скомбинировать и выдать три параметра: стратегию чанкинга, модель-генератор и модель эмбеддинга.

Логика выбора:

• Если тип документа — 'table' (много таблиц): используем чанкинг по строкам или семантический с детекцией таблиц. Для сложных запросов (complexity > 3) поднимаем тир модели до expensive.
• Если тип — 'code': рекурсивный чанкинг с разделителями по функциям/классам. Модель — средняя, но с контекстом 128K (GPT-4.1-mini).
• Если тип — 'legal' (длинные предложения): фиксированный чанкинг по 1000 токенов с перекрытием 200, но с обязательным реранкингом через cross-encoder (см. Cross-Encoders и Reranking).
• Если структура слабая (structure_score < 0.3): семантический чанкинг через LLM (медленно, но качественно).

Код диспетчера:

from typing import Literal

def dispatch(
    query_profile: dict, 
    doc_profile: dict
) -> dict:
    """Выбирает стратегию на основе двух профилей"""
    complexity = query_profile.get('complexity', 1)
    qtype = query_profile.get('type', 'fact')
    dtype = doc_profile.get('dominant_type', 'text')
    structure = doc_profile.get('structure_score', 0.5)
    
    # Стратегия чанкинга
    if dtype == 'table':
        chunk_strategy = 'by_row'
    elif dtype == 'code':
        chunk_strategy = 'recursive_function'
    elif dtype == 'legal' or complexity >= 4:
        chunk_strategy = 'fixed_1000_overlap_200'
    elif structure < 0.3:
        chunk_strategy = 'semantic_llm'
    else:
        chunk_strategy = 'fixed_500_overlap_50'
    
    # Тир модели (наследуем из query_profile)
    if complexity >= 4 or qtype == 'analysis':
        model_tier = 'expensive'
    elif complexity >= 3 or qtype == 'comparison':
        model_tier = 'medium'
    else:
        model_tier = 'cheap'
    
    # Если документ очень длинный (>100K токенов), поднимаем до expensive для лучшего summarization
    if doc_profile['total_tokens'] > 100_000 and model_tier == 'cheap':
        model_tier = 'medium'
    
    return {
        'chunk_strategy': chunk_strategy,
        'model_tier': model_tier,
        'embedding_model': 'text-embedding-3-large' if complexity > 2 else 'text-embedding-3-small'
    }

Аудит на GPT-4.1: как я перестал бояться и полюбил логи

Даже с идеальной диспетчеризацией LLM может галлюцинировать. Аудит — это чек-поинт после генерации ответа. Он не должен быть дорогим, но должен отлавливать фактические ошибки. Я использую GPT-4.1 (не mini, не turbo — именно полную версию) для проверки, потому что он лучше всего держит контекст и умеет следовать структурированному выводу.

Как это работает: после получения ответа мы отдаем GPT-4.1 три вещи: исходный вопрос, найденные чанки (релевантные) и сгенерированный ответ. Модель возвращает JSON с оценками: factual_consistency, completeness, hallucination_detail.

import json

def audit_answer(query: str, chunks: list[str], answer: str) -> dict:
    """Проверяет ответ на галлюцинации с помощью GPT-4.1"""
    response = openai.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": """Ты — аудитор RAG. Проверь ответ на соответствие фактам из чанков.
            Верни JSON:
            {
                "factual_consistency": 0..1 (1 = полностью соответствует),
                "completeness": 0..1 (1 = покрывает все аспекты вопроса),
                "hallucinations": [строка с описанием каждой галлюцинации] или [],
                "recommendation": "accept" | "regenerate" | "fallback"
            }"""},
            {"role": "user", "content": f"Вопрос: {query}\n\nЧанки: {'---'.join(chunks[:5])}\n\nОтвет: {answer}"}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    return json.loads(response.choices[0].message.content)

Если аудит возвращает recommendation: "regenerate", мы повторяем генерацию с усиленным промптом (добавляем инструкцию «не противоречь чанкам»). Если "fallback" — возвращаем пользователю ответ «не удалось найти достоверную информацию» и логируем инцидент.

Собираем пайплайн: от запроса до ответа с верификацией

Теперь сведем все в единый класс. Он принимает запрос, находит подходящий документ (по векторному поиску или по тегу), получает его профиль, диспетчеризует, генерирует ответ и аудитирует.

class AdvancedRAGPipeline:
    def __init__(self, vector_store, document_profiles: dict):
        self.store = vector_store
        self.profiles = document_profiles  # doc_id -> profile
    
    def answer(self, query: str, doc_id: str = None):
        # 1. Парсим вопрос
        qp = parse_query(query)  # из предыдущей статьи
        
        # 2. Получаем профиль документа (по doc_id или из первого найденного)
        if doc_id and doc_id in self.profiles:
            dp = self.profiles[doc_id]
        else:
            # fallback — находим документ по query и берем его профиль
            results = self.store.similarity_search(query, k=1)
            if not results:
                return "Не найден релевантный документ"
            doc_id = results[0].metadata['doc_id']
            dp = self.profiles.get(doc_id, {})
        
        # 3. Диспетчеризация
        config = dispatch(qp, dp)
        
        # 4. Чанкинг (реализация зависит от выбранной стратегии)
        chunks = self._chunk_document(doc_id, config['chunk_strategy'])
        
        # 5. Поиск + генерация
        relevant = self.store.similarity_search(query, chunks=chunks, k=5)
        model_map = {'cheap': 'gpt-4o-mini', 'medium': 'gpt-4.1-mini', 'expensive': 'gpt-4.1'}
        answer = self._generate(query, relevant, model=model_map[config['model_tier']])
        
        # 6. Аудит (только для сложных вопросов)
        if qp['complexity'] >= 3:
            audit = audit_answer(query, relevant, answer)
            if audit['recommendation'] == 'regenerate':
                # повтор с жестким промптом
                answer = self._generate(query, relevant, 
                    prompt_override="Строго следуй фактам из чанков. Не добавляй отсебятину.",
                    model='gpt-4.1')
                audit = audit_answer(query, relevant, answer)
            if audit['recommendation'] == 'fallback':
                return "Извините, не удалось проверить ответ."
        
        return answer

Грабли и лайфхаки: что реально пошло не так

Расскажу о трех проблемах, с которыми столкнулся лично, внедряя этот пайплайн.

1 Профилирование сломалось на PDF с картинками

Если документ — отсканированный PDF, функция profile_document видит строки текста, но не видит таблицы. Решение — добавить этап OCR (через GPT-4.1 Vision или Tesseract) перед профилированием. Но это дорого. Я сделал проще: если документ явно помечен как «image-heavy», профиль заполняется вручную через метаданные.

2 Аудит бесконечно регенерирует

Если GPT-4.1 слишком строг, он может постоянно возвращать regenerate. Я добавил лимит: максимум две регенерации, потом форсированный ответ. И отдельно логгирую такие кейсы для анализа.

3 Перекос в сторону expensive модели

Диспетчер слишком часто выбирал expensive модель для документов с structure_score < 0.3. Пришлось ввести поправку: если complexity <= 2, даже при слабой структуре используем medium модель. Сэкономим бюджет.

Если вы хотите копнуть глубже в тему верификации и ранжирования, рекомендую прочитать 10 критических ошибок RAG в продакшене — там отличные кейсы по аудиту и реранкингу. А для правильных промптов аудитора — наш сборник Промпты для RAG.

Эти три компонента — профилирование документа, динамическая диспетчеризация и аудит через GPT-4.1 — превращают RAG из черного ящика в прозрачную систему. Она не просто генерирует ответ, а объясняет, почему выбрала именно такую стратегию, и сама себя проверяет. В следующей статье расскажу, как добавить feedback loop на основе логов аудита — чтобы система самообучалась не повторять ошибки.

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