Продакшен RAG pipeline для PDF: реляционный парсинг, поиск по содержанию и типизированные ответы – полный гайд

Проблема: почему ваш RAG с PDF — это лотерея

Представьте: вы скормили RAG-системе 500-страничный PDF с техдокументацией. Задаёте вопрос: «Какие ограничения по температуре для корпуса IP68?» — а в ответ получаете какую-то общую фразу про «окружающую среду». Знакомо? Корень зла в том, что стандартный RAG режет PDF на куски без учета структуры. Заголовки, подразделы, таблицы — всё это превращается в кашу. А если у вас PDF с содержанием (TOC) — вы его просто игнорируете. Это не просто неэффективно — это опасно для продакшена.

В этом гайде мы построим pipeline, который: (1) парсит PDF реляционно — сохраняет иерархию заголовков и их связь с параграфами, (2) использует Table of Contents (TOC) как главный поисковый индекс, (3) возвращает типизированные ответы — не простыню текста, а структурированные JSON-объекты. Весь код написан на Python (актуальные версии на июль 2026). Никакого LangChain — только хардкор.

Решение: реляционная модель против плоской строки

Когда вы парсите PDF стандартным pdfplumber и собираете весь текст в одну строку — вы теряете контекст. А контекст — это родительский заголовок. Вопрос «Какие ограничения по температуре?» без привязки к разделу «Температурный режим эксплуатации» — мусор. Мы строим реляционный парсер: каждый chunk хранит ссылку на свой заголовок (1-го, 2-го, 3-го уровня), а TOC даёт нам карту всего документа.

Как это выглядит на практике? Мы извлекаем TOC как отдельную структуру (список разделов с номерами страниц), затем при разбивке текста каждый фрагмент «метим» его разделом из TOC. Так поиск идёт не по всему документу, а по релевантному разделу, найденному через TOC. Плюс мы забираем и соседние разделы для полноты.

Пошаговый план: код и логика

1 Реляционный парсинг PDF

Берём PyMuPDF (он же fitz) — он умеет выдёргивать структуру страниц и закладки (outline). Вот как получить TOC:

import fitz

doc = fitz.open("document.pdf")
toc = doc.get_toc()  # список: [level, title, page]
# Пример: [[1, "Введение", 1], [2, "Цели", 2], [1, "Методы", 3]]

Теперь извлекаем текст постранично, привязывая к текущему активному разделу TOC. Для каждой страницы проверяем, не начался ли новый раздел (по номеру страницы). Создаём chunks — список словарей с полями: {section_title, section_level, page_num, text}.

import re

chunks = []
current_section = ("Document", 0, 0)  # title, level, page_start

for page_num in range(len(doc)):
    page = doc[page_num]
    # проверяем, есть ли в toc раздел, начинающийся на этой странице
    for level, title, toc_page in toc:
        if toc_page == page_num + 1:
            chunks.append({
                "section": current_section[0],
                "level": current_section[1],
                "page": page_num,
                "text": page.get_text()
            })
            current_section = (title, level, page_num)
            break
    else:
        # продолжаем текущий раздел
        if chunks:
            chunks[-1]["text"] += page.get_text()
        else:
            chunks.append({
                "section": "",
                "level": 0,
                "page": page_num,
                "text": page.get_text()
            })

2 Индексация TOC и эмбеддинги

TOC сам по себе — это семантический скелет документа. Мы эмбеддим каждый заголовок и его текст раздела отдельно. Для поиска: когда приходит вопрос, мы сначала ищем в TOC — какой раздел наиболее релевантен (косинусная близость к заголовкам). Затем уже внутри этого раздела ищем по тексту. Это резко улучшает recall на точных вопросах.

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("intfloat/multilingual-e5-large-instruct")  # актуально на 07/2026
# Эмбеддинги заголовков TOC
toc_embeddings = model.encode([s[1] for s in toc])
# Эмбеддинги текстов разделов
section_embeddings = model.encode([chunk["text"] for chunk in chunks])

3 Поиск по содержанию (TOC retrieval)

При получении вопроса делаем две стадии:

1. Вычисляем эмбеддинг вопроса и ищем top-3 заголовка TOC.
2. Для каждого найденного заголовка достаём текст соответствующего раздела (плюс один соседний слева/справа для контекста).
3. Объединяем и передаём LLM.

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

query_emb = model.encode(["Какие ограничения по температуре для IP68?"])
toc_scores = cosine_similarity(query_emb, toc_embeddings)[0]
top_toc_idx = np.argsort(toc_scores)[-3:][::-1]  # индексы top-3

retrieved_chunks = []
for idx in top_toc_idx:
    section_title = toc[idx][1]
    # ищем соответствующий chunk по заголовку
    for chunk in chunks:
        if chunk["section"] == section_title:
            retrieved_chunks.append(chunk["text"])
            break

Такой подход снижает шум: если вопрос про температуру, модель не будет отвлекаться на раздел «Материалы корпуса». Но есть нюанс: заголовки должны быть осмысленными. Если в PDF заголовки типа «2.1» — эмбеддинг бессилен. Тут помогает контекст раздела: мы также эмбеддим первые предложения раздела.

4 Типизированные ответы с Pydantic

Вместо того чтобы просить LLM написать ответ свободным текстом, мы задаём схему ответа. Например, для вопроса про IP68 мы хотим получить JSON: {max_temp: int, min_temp: int, humidity_max: int}. Используем Pydantic для валидации и Few-shot промпт с примером.

from pydantic import BaseModel
from openai import OpenAI  # или локальная LLM через llama.cpp

class Specification(BaseModel):
    max_temperature: float
    min_temperature: float
    humidity_max: float

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")  # для локального Qwen3

prompt = f"""
Извлеки из документа технические характеристики. Ответ дай строго в формате JSON.
Контекст: {retrieved_chunks[0]}
Вопрос: Какие ограничения по температуре для корпуса IP68?
Пример ответа: {{ "max_temperature": 70.0, "min_temperature": -40.0, "humidity_max": 95.0 }}
"""

response = client.chat.completions.create(
    model="qwen3:8b",  # актуальная модель на июль 2026
    messages=[{"role": "user", "content": prompt}],
    response_format={ "type": "json_object" }
)

parsed = Specification.model_validate_json(response.choices[0].message.content)
print(parsed.model_dump())

Нюансы и частые ошибки

Вот что вылезает, когда вы пилите такой pipeline в первый раз:

• Пропущенные заголовки. Не все PDF имеют TOC. Если из get_toc() приходит пустой список — используйте эвристику по размеру шрифта. PyMuPDF: page.get_text("dict") даёт информацию о fontsize.
• Слияние разделов. Когда заголовок есть, а текст разбит на несколько страниц — не забудьте агрегировать. В коде выше мы добавляем текст к последнему chunk'у, но если раздел длинный — лучше жёстко разбивать на под-chunks с якорем на тот же заголовок.
• Типизация не срабатывает, если модель тупит. Рекомендую добавить ретрай с повторным промптом, а если JSON не валидируется — возвращать ответ в свободной форме с пометкой. Pydantic облегчает жизнь, но не гарантирует идеальное следование схеме.
• Забываете про мета-данные. Всегда храните номер страницы и имя файла в output. Это спасёт при отладке.

Кстати, если вы думаете, что TOC retrieval решит все проблемы, почитайте статью про 10 критических ошибок RAG в продакшене — там отлично описано, что даже с идеальным поиском можно провалить качество из-за плохого промпта.

Собираем всё вместе

Мы прошли от разрозненных страниц до структурированного ответа. Это не серебряная пуля, но base-level для любого продакшен RAG, где важна точность и интеграция. Код выше — это каркас. Для локального запуска советую Ollama с Qwen3:8B (стоит около 4-6 ГБ RAM), для эмбеддингов — intfloat/multilingual-e5-large-instruct. Если нужно обрабатывать таблицы — добавьте camelot или pymupdf-table, но это уже совсем другая история.

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