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

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

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

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

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

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

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

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

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

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

Спойлер с результатом

Получилось – опыт ^[3] в обучении модели с нуля.

Не получилось – справочника по вселенной.

Модель вышла около 164М параметров, при этом датасет был всего 72М токенов и 2.5К пар QA. Законы масштабирования ^[4] не обманешь, получилось не очень – текст генерит, иногда даже связанный, instruct версия даже на вопросы иногда правильно отвечает. Но в целом – это не модель, которая живёт в мире Империума и Варпа. Но я не собираюсь сдаваться и прорабатываю варианты 😀

Код обучения тут ^[5]

Базовая модель тут ^[6]

Instruct версия тут ^[7]

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

Сырые данные

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

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

Данные берут из разных источников. Используют википедию, открытые репозитории, (Reddit и ему подобные). Для сбора данных есть, например Common Crawl ^[8], которая сканирует весь интернет и собирает данные в свой архив. Последний (на момент написания статьи) дамп от мая 2026 ^[9] содержит более двух миллиардов страниц общим объёмом около 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 в код не добавлял специально – если файл падал с ошибкой ^[10] или возвращал пустое содержимое, то вручную прогонял его через распознавание. А так – идём по страничкам, собираем текст, потом соединяем в одну строку. Дальше 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. Но нейросети не работают с буквами, им нужны цифры, чтобы их умножать и считать вероятность. Токенизатор как раз переводит текст в числа. Звучит просто, но тут целая эволюция ^[11] подходов — от “мешка слов” до байтового 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) простая: начинаем с алфавита, смотрим, какая пара встречается в корпусе чаще всего (просто подсчётом количества), сливаем её в новый токен, повторяем ^[12], пока словарь не вырастет до нужного размера. В итоге получается словарь из букв, их пар, троек и так далее. Часто используемые слова, типа “он”, “она”, занимают по одному токену, А что-то редкое, типа “Империум”, будет состоять из отдельных токенов: “Им-пер-и-ум”.

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

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

WordPiece

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

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

Unigram

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

Итого: эти три ветки это не линейная эволюция, а именно три разных алгоритма, со своими плюсами и минусами. Для своей модели я взял 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). Начнём обучать токенизатор. И первым делом я загрузил все тексты в память ^[15] (объём датасета позволял):

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 ^[16] от 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.

Код проекта тут ^[5]