Своя GPT-like LLM по WH40K с нуля. Часть 1: токенизируем Империум

Меня зовут Владимир, и я давно хотел погрузиться во вселенную Warhammer40K. И в прошлом году я решил серьёзно подойти к этому вопросу – поискал в интернете с чего начать. Интернет подсказал, что единственный путь начать – почитать Ересь Хоруса. Дальнейший поиск показал, что на момент поиска цикл содержит 56 романов, 24 повести и ещё всякое-разное, написанное более чем 19 различными авторами. Столько читать у меня времени не было, поэтому было принято единственное правильное решение – обучу-ка я LLM на лоре Warhammer40K и пусть она меня просвещает! И чтобы было интереснее, код этой LLM я решил написать сам, на голом pytorch (ну почти).

Данная статья – первая в цикле статей по созданию и обучению GPT-like LLM с нуля. В них я постараюсь (в очередной раз) рассказать, как работают Decoder-only модели, как обучить небольшую LLM, сколько это заняло у меня времени. Статьи будут разбиты по логическим разделам и цикл будет примерно следующего содержания:

1. Подготовка и токенизация данных (вы находитесь здесь)

2. Трансформер

3. Сборка и обучение LLM

4. SFT этап (дообучение на Вопрос-Ответ)

5. Интеграция с Hugging Face

Содержание может меняться и дополняться ссылками по мере написания

Если кому-то просто интересно, что получилось, и неохота ждать до конца, то результат под спойлером

Ну а кому интересно, как написать и обучить модель с нуля – продолжим. И первое, что нам надо – данные.

Сырые данные

Данные для LLM – это текст. Если кто-то внезапно решил читать эту статью не понимая, как работают LLM, то грубо поясню – LLM тупо предсказывают вероятность следующего токена (читай буквы или слова) к уже имеющейся последовательности. И основная (и самая трудоёмкая) задача первого этапа обучения (pre-training) – это научить модель верно предсказывать эти токены, для чего и требуются огромные наборы различных текстов. Согласно закону масштабирования (известному также, как Закон масштабирования Шиншиллы) часто приводят ориентир около 20 токенов текста на один параметр модели.

Текст может быть любой, но главное помнить – на чём научишь LLM, так она и будет общаться. Условно, наберёшь датасет с форумов пацанских цитат – модель будет общаться, как Джейсон Стэйтем.

Данные берут из разных источников. Используют википедию, открытые репозитории, (Reddit и ему подобные). Для сбора данных есть, например Common Crawl, которая сканирует весь интернет и собирает данные в свой архив. Последний (на момент написания статьи) дамп от мая 2026 содержит более двух миллиардов страниц общим объёмом около 400 ТБ текста. В моём случае я хотел обучить модель не на нашей вселенной, а на вселенной Warhammer40K, так что это были различные тексты по Вахе, которые удалось найти.

Первым делом нашу кучку текстов надо привести к удобообрабатываемому формату – я выбрал JSON. Из каждого источника необходимо извлечь текст, запаковать в JSON и сохранить для дальнейшей обработке. Возможно это было лишним этапом, но мне было удобнее именно так. Для преобразования текстов решил сделать следующее – написать несколько экстракторов текста из файлов и сделать единый пайплайн обработки, который в зависимости от расширения файла вызывал бы соответствующий экстрактор. Начнем с pdf:

import fitz

def extract_text_from_pdf(file_path):
    try:
        doc = fitz.open(file_path)
    except Exception as e:
        raise RuntimeError(f'Failed to open PDF: {e}')

    text_parts = []
    for page in doc:
        try:
            text = page.get_text().strip()
            if text:
                text_parts.append(text)
        except Exception:
            continue
    doc.close()
    full_text = ' '.join(text_parts).strip()
    return full_text if full_text else None

Писал относительно давно, и код уже немного legacy (надо бы fitz заменить), но работает. Никакой OCR в код не добавлял специально – если файл падал с ошибкой или возвращал пустое содержимое, то вручную прогонял его через распознавание. А так – идём по страничкам, собираем текст, потом соединяем в одну строку. Дальше fb2:

from lxml import etree

def extract_text_from_fb2(file_path):
    parser = etree.XMLParser(recover=True, encoding='utf-8')
    with open(file_path, 'rb') as f:
        content_bytes = f.read()
    first_line = content_bytes.split(b'n', 1)[0].decode('latin-1', errors='ignore')
    if 'windows-1251' in first_line.lower() or 'cp1251' in first_line.lower():
        content = content_bytes.decode('cp1251')
    else:
        try:
            content = content_bytes.decode('utf-8')
        except UnicodeDecodeError:
            content = content_bytes.decode('cp1251')
    tree = etree.fromstring(content.encode(), parser)
    text = ' '.join(tree.xpath(
            '//fb2:p//text()',
            namespaces={'fb2': 'http://www.gribuser.ru/xml/fictionbook/2.0'}
    )).strip()

    return text if text else None

Первым делом создаём парсер lxml с recover=True — при битой разметке он пытается восстановить документ, а не падает сразу. Далее пришлось добавить определение кодировки – по умолчанию файл декодировался в utf-8, а многие файлы были в кодировке Windows-1251, из-за чего парсер падал. Дальше собираем все абзацы книги в одну строку – с разметкой и заголовками решил не загоняться. Ну и epub:

import ebooklib
from ebooklib import epub
from bs4 import BeautifulSoup

def extract_text_from_epub(file_path):
    try:
        book = epub.read_epub(file_path, {'ignore_ncx': True})
    except Exception as e:
        raise RuntimeError(f'Failed to read EPUB: {e}')

    text_parts = []

    for item in book.get_items():
        if item.get_type() == ebooklib.ITEM_DOCUMENT:
            soup = BeautifulSoup(item.get_content(), 'html.parser')
            # Удаляем скрипты и стили
            for script in soup(['script', 'style']):
                script.decompose()
            text = soup.get_text(separator=' ', strip=True)
            if text:
                text_parts.append(text)

    full_text = ' '.join(text_parts).strip()
    return full_text if full_text else None

Аналогично fb2 (но другими библиотеками) извлекаем чистый текст из книги и склеиваем в строку. Теперь пайплайн. Сначала зарегистрируем наши экстракторы по расширениям файлов:

EXTENSION_TO_EXTRACTOR = {
    '.fb2': extract_text_from_fb2,
    '.epub': extract_text_from_epub,
    '.pdf': extract_text_from_pdf
}

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

def process_book_files(input_dir: Path, output_dir: Path | None = None, allowed_ext: list[str] = None) -> None:
    error_logs = []
    counter = Counter()
    ext_counter = Counter()

    if output_dir is None:
        output_dir = Path(__file__).parent / 'JSON'
    output_dir.mkdir(parents=True, exist_ok=True)

    if allowed_ext is None:
        allowed_extensions = set(EXTENSION_TO_EXTRACTOR.keys())
    else:
        allowed_extensions = set()
        for ext in allowed_ext:
            ext = ext.lower()
            if not ext.startswith('.'):
                ext = f'.{ext}'
            if ext in EXTENSION_TO_EXTRACTOR:
                allowed_extensions.add(ext)
            else:
                print(f'Warning: неподдерживаемое расширение: "{ext}"')

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

def process_book_files(input_dir: Path, output_dir: Path | None = None, allowed_ext: list[str] = None) -> None:
    # Предыдущий код
    files_list = [
        f for f in input_dir.rglob('*')
        if f.is_file() and f.suffix.lower() in allowed_extensions
    ]

Средствами Path рекурсивно обходим заданный каталог и фильтруемся по расширениям, которые нужно обработать. Ну а дальше пошла предобработка:

def process_book_files(input_dir: Path, output_dir: Path | None = None, allowed_ext: list[str] = None) -> None:
    # Предыдущий код
    for file_path in tqdm(files_list, desc='Обработка'):
        ext = file_path.suffix.lower()
        extractor = EXTENSION_TO_EXTRACTOR.get(ext)

        try:
            text = extractor(file_path)
            if text is None:
                error_logs.append(f'Warning: Не найден текст в "{file_path}"')
                counter['warnings'] += 1
                continue
            output_file_path = output_dir / f'{file_path.stem}.json'
            data = {
                'parent_folder': file_path.parent.name,
                'file_name': file_path.name,
                'text': text
            }
            with open(output_file_path, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            counter['processed'] += 1
            ext_counter[ext] += 1
        except Exception as e:
            counter['errors'] += 1
            error_logs.append(f'Ошибка при обработке "{file_path}": {e}')

Проходим по полученному списку файлов, извлекаем текст, формируем и сохраняем JSON, пишем статистику. Наверное надо было добавить к JSON-файлу хотя-бы поле create_at, но было лень. Да и данных было не много – всегда можно с нуля всё ещё раз прогнать.

Сохраняем ошибки преобразования.

def process_book_files(input_dir: Path, output_dir: Path | None = None, allowed_ext: list[str] = None) -> None:
    # Предыдущий код
    if error_logs:
        LOG_DIR.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        log_file_path = LOG_DIR / f'log_{timestamp}.txt'

        with open(log_file_path, 'w', encoding='utf-8') as log_file:
            log_file.write('n'.join(error_logs) + 'n')

        if len(error_logs) < 10:
            print('n--- Log ---')
            for log in error_logs:
                print(log)
        else:
            print(f'Лог ошибок сохранён в: {log_file_path}')

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

def process_book_files(input_dir: Path, output_dir: Path | None = None, allowed_ext: list[str] = None) -> None:
    # Предыдущий код
    if counter:
        msg = '; '.join(f'{key.capitalize()}: {counter[key]}' for key in counter)
        print('-' * len(msg), msg, '-' * len(msg), sep='n')

    if ext_counter:
        ext_msg = 'Обработано файлов: ' + ', '.join(
            f'{ext}: {count}' for ext, count in sorted(ext_counter.items()))
        print(ext_msg)

В первой строке выводим общее количество удачных обработок, предупреждений (когда экстрактор вернул пустую строку) и случаев падения с ошибкой. Во второй строке – статистику по количеству файлов каждого расширения.

Всё. Данные готовы. Переходим к токенизатору.

Токенизатор

Итак, тексты у нас теперь в JSON. Но нейросети не работают с буквами, им нужны цифры, чтобы их умножать и считать вероятность. Токенизатор как раз переводит текст в числа. Звучит просто, но тут целая эволюция подходов — от “мешка слов” до байтового BPE, который я и использовал. Но для начала немного исторической теории.

Мешок слов

Самый древний подход — Bag of Words (“мешок слов”). Берём текст, режем на слова, каждому слову – свой номер. Порядок слов выбрасываем, получаем вектор “сколько раз встретилось”. Для классификации спама или тональности отзывов – норм. Для языковой модели – нет: она должна учиться предсказывать следующий токен с учётом порядка. Плюс редкие слова, опечатки, имена собственные – всё это раздувает словарь или улетает в <unk>.

<unk> – специальный токен у токенизаторов, обозначающий “Я ХЗ, что это за слово”

Логичными крайностями для мешка слов стали Word-level токенизация, где одно слово = один токен и Character-level, где за токен принимается один символ. Проблема первого подхода – просто огромный словарь и <unk>, второго – длинные неоптимальные последовательности. И тут надо понимать, что каждый лишний токен – это полный проход через все веса модели (если это конечно полносвязная сеть): для 1B параметров грубо ~2 GFLOPs на токен, для 500B–1T – уже терафлопсы. А на минуточку, у топовой бытовой RTX 5090 их (терафлопс) всего 104, хоть модель в 1T параметров в неё и не влезет.

Короче, мешок слов – не про LLM. Это быстро поняли и разработчики LLM, поэтому придумали несколько новых алгоритмов токенизации.

BPE

Идея BPE (Byte Pair Encoding) простая: начинаем с алфавита, смотрим, какая пара встречается в корпусе чаще всего (просто подсчётом количества), сливаем её в новый токен, повторяем, пока словарь не вырастет до нужного размера. В итоге получается словарь из букв, их пар, троек и так далее. Часто используемые слова, типа “он”, “она”, занимают по одному токену, А что-то редкое, типа “Империум”, будет состоять из отдельных токенов: “Им-пер-и-ум”.

У BPE токенизации есть разновидность – Byte-level BPE. За основу в ней уже берут не символы Unicode, а их байт-значения.

BPE — самый распространённый выбор для decoder-only LLM: простой, быстрый в обучении.

WordPiece

WordPiece чем-то похож на BPE, только пары сливаются не по частоте, а по выгоде для языковой модели: на каждом шаге берётся та пара, которая сильнее всего улучшает правдоподобие текста. То есть, в отличие от BPE берётся не просто частота появления пары, а отношение частоты появления пары к произведению частоты появлений каждого символа пары.

На практике результат похож на BPE – тот же субсловный словарь, те же куски слов, но критерий отбора другой. Главный потребитель – BERT и всё encoder-семейство (DistilBERT, RuBERT и т.д.). Фишка WordPiece: продолжение слова помечается префиксом ## (“Варп” = “Ва” + “##рп”), и модель видит, что это не отдельное слово, а хвост.

Unigram

У BPE и WordPiece словарь растёт: от букв к парам, от пар к кускам слов и целым словам. Unigram делает наоборот – стартует с огромного набора субстрок (по сути, кандидатов на токены из корпуса) и потом выкидывает лишнее. Логика: “что из этого набора реально помогает описать текст, а что можно убрать без большой потери?”. На практике Unigram даёт субсловный словарь, похожий на BPE, но с другой философией обучения. BPE спрашивает “что чаще всего склеить?”, Unigram — “какие куски вообще нужны в словаре?”. Для GPT-like моделей чаще берут BPE, Unigram чаще в encoder-decoder (T5) и части encoder-моделей. Описывать алгоритм я не буду, но если что, то почитать можно тут

Итого: эти три ветки это не линейная эволюция, а именно три разных алгоритма, со своими плюсами и минусами. Для своей модели я взял Byte-Level BPE в духе GPT-2.

Обучение Byte-Level BPE токенизатора

Обучить BPE токенизатор довольно просто:

1. Составляем первоначальный словарь из последовательности от 0 до 255. Это наши базовые байты-токены.

2. Преобразуем текст в список байтов в кодировке UTF-8

3. С помощью collections.Counter, считаем количество вхождений каждой пары байт в корпусе текста, проходя по всему тексту циклом вида Counter((byte_seq[i], byte_seq[i + 1]) for i in range(len(byte_seq) - 1))

4. Выбираем самую часто встречающуюся пару байт (например, это 0x42 и 0x72) и присваиваем этой паре id=256.

5. Заменяем в тексте все вхождения пары (0x42, 0x72) на наш новый id 0x100.

6. Повторяем с пункта 3, пока не наберём нужный размер словаря.

Дополнительно, можно добавить фильтрацию по частоте входжения (pair_counts.value() >= min_frequency). Начнём обучать токенизатор. И первым делом я загрузил все тексты в память (объём датасета позволял):

def extract_texts_from_json_dir(data_dir: str | Path) -> list[str]:
    data_path = Path(data_dir)
    texts = []
    if not data_path.is_dir():
        raise FileNotFoundError(f'Директория "{data_dir}" не существует')
    json_files = list(data_path.glob('*.json'))
    if not json_files:
        raise FileNotFoundError(
            f'В директории "{data_dir}" не найдено .json файлов')
    for json_file in json_files:
        try:
            with open(json_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                if isinstance(data, dict) and 'text' in data:
                    text = data['text']
                    if isinstance(text, str):
                        texts.append(text.strip())
                else:
                    print(f'Файл {json_file.name} не содержит поля "text"')
        except Exception as e:
            print(f'Не удалось прочитать файл {json_file.name}: {e}')
            continue
    return texts

Весь метод – защита от самого себя😁 Проверяю, что передана именно папка, что в ней есть JSON файлы, что они парсятся в словарь, в котором есть текстовое поле text. Ну и возвращаю список текстов. А дальше я столкнулся с одной проблемкой. Писать свой BPE можно – алгоритм выше это около 30 строк Python кода. Но я попробовал, и мне не понравилось, уж очень медленно это всё работало на нативном Python. Поэтому с токенизатором я немного считерил, и воспользовался библиотекой tokenizers от HuggingFace. Если верить GitHub-у, то библиотека на 72% состоит из Rust, и работает явно быстрее самописной:

def train_byte_level_bpe_tokenizer(
        texts: list[str], vocab_size: int = 50257, min_frequency: int = 2, save_dir: str | Path = 'tokenizer_config'
) -> PreTrainedTokenizerFast:
    if not texts:
        raise ValueError('Список текстов пуст.')

    tokenizer = Tokenizer(BPE())
    tokenizer.pre_tokenizer = ByteLevel(add_prefix_space=False)
    trainer = BpeTrainer(
        vocab_size=vocab_size, min_frequency=min_frequency, special_tokens=['<|endoftext|>', '<pad>'], show_progress=True,)
    tokenizer.train_from_iterator(texts, trainer=trainer)
    tokenizer.post_processor = processors.ByteLevel(trim_offsets=False)
    tokenizer.decoder = ByteLevelDecoder()

    save_path = Path(save_dir)
    save_path.mkdir(parents=True, exist_ok=True)

    hf_tokenizer = PreTrainedTokenizerFast(
        tokenizer_object=tokenizer,
        bos_token='<|endoftext|>',
        eos_token='<|endoftext|>',
        pad_token='<pad>',
    )
    hf_tokenizer.save_pretrained(save_dir)

    return hf_tokenizer

Конструктором Tokenizer(BPE()) мы создаём пайплайн обучения токенизатора и задаём алгоритм – наш целевой BPE. Параметр tokenizer.pre_tokenizer задаёт этап предобработки текста (перевод текста в байты). Далее создаём тренера, в котором приводим все параметры обучения. В моём случае это размер словаря, фильтрация по минимальному вхождению, резерв ID под специальные токены и визуализация прогресса обучения. НА вопрос “Почему vocab_size = 50257” честно отвечу – косяк! Это должно было быть 50к токенов + 256 начальных символов + EOS. Но я задал тренеру два токена – PAD и EOS, поэтому формально размер словаря вышел, как цена в МВидео – 49999.

После завершения обучения (вызов tokenizer.train_from_iterator) дополняем токенизатор обратным преобразованием из байт в текст и сохраняем в папку для дальнейшего использования.

Токенизатор готов, время обрабатывать данные.

Создание датасета

Любое машинное обучение строится на одном принципе – подаём что-то на вход модели, сравниваем значение выхода с целевым, корректируем веса модели. Учитывая, что LLM должна учиться предсказывать следующий токен, наш датасет должен на каждой позиции спрашивать: “какой токен идёт следующим?”. Думаю теории достаточно для написания датасета:

from torch.utils.data import Dataset
from transformers import PreTrainedTokenizerBase

class W40kDataset(Dataset):
    def __init__(
            self, tokenizer: PreTrainedTokenizerBase, json_path: str | Path, max_length: int = 1024, force_reprocess=False):
        self.tokenizer = tokenizer
        self.max_length = max_length
        self.sep_token_id = tokenizer.convert_tokens_to_ids('<|endoftext|>')

        self.processed_dir = Path(json_path) / 'processed'
        self.processed_dir.mkdir(exist_ok=True)
        self.data_file = self.processed_dir / 'tokenized_data.npy'

        if force_reprocess or not self.data_file.exists():
            self._preprocess_data(json_path)

        self.token_blocks = np.load(self.data_file, mmap_mode='r')

Так как наша LLM будет на торче, то и датасет наследуем от торчёвого torch.utils.data.Dataset. В конструктор передадим токенизатор (необходим для первичной обработки тескта), путь к исходным текстам и размер блока (он же размер контекстного окна нашей будущей модели). Сами данные мы будем хранить в numpy-массиве, в захардкоженной папке processed и “лениво” подгружать с диска по мере необходимости (для меня это не необходимость, а скорее дань традиции для датасетов на сотни ГБ). Сам же numpy-массив формируется методом _preprocess_data:

class W40kDataset(Dataset):
    # Предыдущий код
    def _preprocess_data(self, json_dir_path):
        all_token_ids = []
        json_files = list(Path(json_dir_path).glob('*.json'))

        for json_path in tqdm(json_files, desc='Обработка файлов'):
            with open(json_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
                text = data['text']
                tokens = self.tokenizer.encode(text, add_special_tokens=False)
                tokens.append(self.sep_token_id)
                all_token_ids.extend(tokens)

        total_tokens = len(all_token_ids)
        num_blocks = total_tokens // self.max_length
        print(f'Всего в корпусе данных {total_tokens:,} токенов')

        blocks = []
        for i in tqdm(range(num_blocks), desc='Создание блоков данных'):
            start = i * self.max_length
            end = start + self.max_length
            blocks.append(all_token_ids[start:end])

        np.save(self.data_file, np.array(blocks, dtype=np.int32))

В очередной раз читаем наши JSON, только в этот раз превращаем текст в токены. К каждой последовательности токенов, полученной из файла, мы добавляем наш EOS и складываем всё в общий массив. Далее этот массив нарезается на блоки max_length длины и сохраняется в файл. “Гениальность” решения в том, что загрузка – ленивая, а запись нет. В реальности на больших данных надо было и запись делать поблочно – по достижению max_length сохранять в файл и чистить буфер, но

Ну и стандартные методы класса Dataset:

class W40kDataset(Dataset):
    # Предыдущий код
    def __len__(self):
        return len(self.token_blocks)

    def __getitem__(self, idx):
        block = self.token_blocks[idx]
        return (
            torch.tensor(block[:-1], dtype=torch.long),
            torch.tensor(block[1:], dtype=torch.long))

Decoder-only модели устроены так, что предсказывают следующий токен для каждого токена во входной последовательности. Этим мы и пользуемся в датасете – наш таргет это сдвинутый на одну позицию вход. Таким образом мы с одного батча получим целых 1023 предсказания.

Создадим датасет

Итоги

На этой замечательной ноте первая часть заканчивается. Датасет для претрейна получен, и в следующей части мы перейдём к реализации ванильной Decoder-only LLM.

Код проекта тут