Команда Python for Devs подготовила перевод статьи о том, как с помощью LlamaIndex и Pydantic можно превратить сканы чеков в структурированные данные. Минимум кода — и у вас готовый CSV для анализа.

---

Ручной ввод данных из чеков, счетов и контрактов отнимает часы времени и часто приводит к ошибкам. А что если можно автоматически извлекать структурированные данные из таких документов всего за несколько минут?

В этой статье вы узнаете, как превратить изображения чеков в структурированные данные с помощью LlamaIndex, а затем экспортировать результат в таблицу для анализа.

Полный исходный код и Jupyter-ноутбук для этого туториала доступны на GitHub. Клонируйте репозиторий и повторяйте шаги вместе с нами!

Чему вы научитесь

• Преобразовывать сканированные чеки в структурированные данные с помощью LlamaParse и Pydantic-моделей

• Проверять точность извлечения, сравнивая результаты с эталонной разметкой

• Исправлять ошибки парсинга с помощью предобработки низкокачественных изображений

• Экспортировать очищенные данные чеков в формат таблицы

Знакомство с LlamaIndex

LlamaIndex — это фреймворк, который соединяет LLM с вашими данными через три ключевые возможности:

• Загрузка данных: встроенные ридеры для PDF, изображений, веб-страниц и баз данных автоматически преобразуют содержимое в обрабатываемые узлы.

• Структурированное извлечение: преобразование неструктурированного текста в Pydantic-модели с автоматической валидацией на базе LLM.

• Поиск и индексация: векторные хранилища и семантический поиск, позволяющие выполнять запросы с учётом контекста по вашим документам.

Он избавляет от шаблонного кода для загрузки, парсинга и запросов к данным, позволяя сосредоточиться на разработке LLM-приложений.

Ниже приведено сравнение LlamaIndex с двумя другими популярными фреймворками для LLM-приложений:

Установка

Для начала установите необходимые пакеты:

• llama-index: базовый фреймворк LlamaIndex с функциями индексации и поиска

• llama-parse: сервис парсинга документов для PDF, изображений и сложных макетов

• llama-index-program-openai: интеграция с OpenAI для извлечения структурированных данных в формате Pydantic

• python-dotenv: загрузка переменных окружения из .env-файлов

• rapidfuzz: библиотека нечеткого сопоставления строк, например для сравнения названий компаний с мелкими отличиями

pip install llama-index llama-parse llama-index-program-openai python-dotenv rapidfuzz

Настройка окружения

Создайте файл .env для хранения API-ключей:

# .env
LLAMA_CLOUD_API_KEY="your-llama-parse-key"
OPENAI_API_KEY="your-openai-key"

Получить ключи можно здесь:

• LlamaParse API: cloud.llamaindex.ai

• OpenAI API: platform.openai.com/api-keys

Загрузите переменные окружения с помощью load_dotenv:

from dotenv import load_dotenv
import os

load_dotenv()

Настройте LLM по умолчанию через Settings:

from llama_index.core import Settings
from llama_index.llms.openai import OpenAI

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.context_window = 8000

Settings хранит глобальные настройки, так что каждый движок запросов и программа используют одну и ту же конфигурацию LLM. Установка temperature = 0 заставляет модель возвращать детерминированный и структурированный вывод.

Базовая обработка изображений с LlamaParse

В этом туториале мы будем использовать SROIE Dataset v2 с Kaggle. Этот датасет содержит реальные сканы чеков из соревнования ICDAR 2019.

Загрузить датасет можно с сайта Kaggle или через CLI:

# Установите Kaggle CLI (один раз)
uv pip install kaggle

# Настройте учётные данные Kaggle (один раз на окружение)
export KAGGLE_USERNAME=your_username
export KAGGLE_KEY=your_api_key

# Создайте папку и скачайте архив (~1 ГБ)
mkdir -p data
kaggle datasets download urbikn/sroie-datasetv2 -p data

# Распакуйте и посмотрите несколько файлов
unzip -q -o data/sroie-datasetv2.zip -d data

Мы будем использовать данные из директории data/SROIE2019/train/, в которой есть:

• img: оригинальные изображения чеков

• entities: эталонная разметка для валидации

Загрузим первые 10 чеков в список путей:

from pathlib import Path

receipt_dir = Path("data/SROIE2019/train/img")
num_receipts = 10
receipt_paths = sorted(receipt_dir.glob("*.jpg"))[:num_receipts]

Посмотрим на первый чек:

from IPython.display import Image

first_receipt_path = receipt_paths[0]
Image(filename=first_receipt_path)

Теперь используем LlamaParse, чтобы преобразовать первый чек в markdown:

from llama_parse import LlamaParse

# Парсинг чеков с помощью LlamaParse
parser = LlamaParse(
    api_key=os.environ["LLAMA_CLOUD_API_KEY"],
    result_type="markdown",  # формат вывода
    num_workers=4,  # параллельная обработка
    language="en",  # подсказка для OCR
    skip_diagonal_text=True,  # игнорировать наклонный текст
)
first_receipt = parser.load_data(first_receipt_path)[0]

Предпросмотр markdown для первого чека:

preview = "\n".join(first_receipt.text.splitlines()[:10])
print(preview)

Вывод:

tan woon yann
BOOK TA K (TAMAN DAYA) SDN BHD
789417-W
NO.5: 55,57 & 59, JALAN SAGU 18,
TAMAN DaYA,
81100 JOHOR BAHRU,
JOHOR.

LlamaParse успешно преобразует изображение чека в текст, но структуры здесь нет: названия компаний, даты и суммы перемешаны в сплошной текст. Такой формат неудобен для экспорта в таблицы или аналитические инструменты.

В следующем разделе мы будем использовать Pydantic-модели, чтобы автоматически извлекать структурированные поля вроде company, total и purchase_date.

Извлечение структурированных данных с помощью Pydantic

Pydantic — это Python-библиотека, которая использует аннотации типов для валидации данных и автоматического преобразования типов. Определив схему для чека один раз, вы сможете получать структурированные данные в едином формате, независимо от того, как именно выглядит исходный чек.

Начнём с описания двух Pydantic-моделей, которые отражают структуру чека:

from datetime import date
from typing import List, Optional
from pydantic import BaseModel, Field, ValidationInfo, model_validator


class ReceiptItem(BaseModel):
    """Представляет одну строку из чека."""

    description: str = Field(description="Название товара точно так, как указано в чеке")
    quantity: int = Field(default=1, ge=1, description="Количество товара (целое число)")
    unit_price: Optional[float] = Field(
        default=None, ge=0, description="Цена за единицу в валюте чека"
    )
    discount_amount: float = Field(
        default=0.0, ge=0, description="Скидка, применённая к этой позиции"
    )


class Receipt(BaseModel):
    """Структурированные поля, извлечённые из розничного чека."""

    company: str = Field(description="Название компании или магазина")
    purchase_date: Optional[date] = Field(
        default=None, description="Дата в формате YYYY-MM-DD"
    )
    address: Optional[str] = Field(default=None, description="Адрес компании")
    total: float = Field(description="Итоговая сумма к оплате")
    items: List[ReceiptItem] = Field(default_factory=list)

Теперь создадим OpenAIPydanticProgram, который укажет LLM извлекать данные в соответствии с нашей моделью Receipt:

from llama_index.program.openai import OpenAIPydanticProgram

prompt = """
You are extracting structured data from a receipt.
Use the provided text to populate the Receipt model.
Interpret every receipt date as day-first.
If a field is missing, return null.

{context_str}
"""

receipt_program = OpenAIPydanticProgram.from_defaults(
    output_cls=Receipt,
    llm=Settings.llm,
    prompt_template_str=prompt,
)

Проверим на первом документе, что всё работает, прежде чем запускать обработку всего набора:

# Обработка первого чека
structured_first_receipt = receipt_program(context_str=first_receipt.text)

# Выведем результат в JSON для удобства
print(structured_first_receipt.model_dump_json(indent=2))

Вывод:

{
  "company": "tan woon yann BOOK TA K (TAMAN DAYA) SDN BHD",
  "purchase_date": "2018-12-25",
  "address": "NO.5: 55,57 & 59, JALAN SAGU 18, TAMAN DaYA, 81100 JOHOR BAHRU, JOHOR.",
  "total": 9.0,
  "items": [
    {
      "description": "KF MODELLING CLAY KIDDY FISH",
      "quantity": 1,
      "unit_price": 9.0,
      "discount_amount": 0.0
    }
  ]
}

LlamaIndex заполняет схему Pydantic извлечёнными значениями:

• company: название компании из заголовка чека

• purchase_date: распознанная дата (2018-12-25)

• total: итоговая сумма (9.0)

• items: список позиций с названием, количеством и ценой

Теперь, когда извлечение работает, масштабируем процесс на все чеки. Функция ниже использует имя файла чека как уникальный идентификатор:

def extract_documents(paths: List[str], prompt: str, id_column: str = "receipt_id") -> List[dict]:
    """Извлечение структурированных данных из документов с помощью LlamaParse и LLM."""
    results: List[dict] = []

    # Инициализация парсера с параметрами OCR
    parser = LlamaParse(
        api_key=os.environ["LLAMA_CLOUD_API_KEY"],
        result_type="markdown",
        num_workers=4,
        language="en",
        skip_diagonal_text=True,
    )

    # Конвертация изображений в markdown
    documents = parser.load_data(paths)

    # Создание программы для структурированного извлечения
    program = OpenAIPydanticProgram.from_defaults(
        output_cls=Receipt,
        llm=Settings.llm,
        prompt_template_str=prompt,
    )

    # Извлечение данных из каждого документа
    for path, doc in zip(paths, documents):
        document_id = Path(path).stem
        parsed_document = program(context_str=doc.text)
        results.append(
            {
                id_column: document_id,
                "data": parsed_document,
            }
        )
    return results

# Извлечение данных из всех чеков
structured_receipts = extract_documents(receipt_paths, prompt)

Далее преобразуем извлечённые данные в DataFrame для удобного анализа:

import pandas as pd


def transform_receipt_columns(df: pd.DataFrame) -> pd.DataFrame:
    """Применить стандартные преобразования к колонкам DataFrame с чеками."""
    df = df.copy()
    df["company"] = df["company"].str.upper()
    df["total"] = pd.to_numeric(df["total"], errors="coerce")
    df["purchase_date"] = pd.to_datetime(
        df["purchase_date"], errors="coerce", dayfirst=True
    ).dt.date
    return df


def create_extracted_df(records: List[dict], id_column: str = "receipt_id") -> pd.DataFrame:
    df = pd.DataFrame(
        [
            {
                id_column: record[id_column],
                "company": record["data"].company,
                "total": record["data"].total,
                "purchase_date": record["data"].purchase_date,
            }
            for record in records
        ]
    )
    return transform_receipt_columns(df)


extracted_df = create_extracted_df(structured_receipts)
extracted_df

Большинство чеков распознаны корректно, но у чека X51005200938 есть проблемы:

• Название компании неполное («TH MNAN»)

• Итоговая сумма равна 0, хотя фактически другая

• Дата (2023-10-11) выглядит недостоверной

Сравнение извлечённых данных с эталоном

Чтобы проверить точность извлечения, загрузите эталонную разметку из data/SROIE2019/train/entities:

def normalize_date(value: str) -> str:
    """Привести строку даты к единому формату."""
    value = (value or "").strip()
    if not value:
        return value
    # Заменяем дефисы на слэши
    value = value.replace("-", "/")
    parts = value.split("/")
    # Преобразуем 2-значный год в 4-значный (например, 18 -> 2018)
    if len(parts[-1]) == 2:
        parts[-1] = f"20{parts[-1]}"
    return "/".join(parts)


def create_ground_truth_df(
    label_paths: List[str], id_column: str = "receipt_id"
) -> pd.DataFrame:
    """Создать DataFrame с эталонными данными из JSON-файлов разметки."""
    records = []
    # Загружаем каждый JSON-файл и извлекаем ключевые поля
    for path in label_paths:
        payload = pd.read_json(Path(path), typ="series").to_dict()
        records.append(
            {
                id_column: Path(path).stem,
                "company": payload.get("company"),
                "total": payload.get("total"),
                "purchase_date": normalize_date(payload.get("date")),
            }
        )

    df = pd.DataFrame(records)
    # Применяем те же преобразования, что и к извлечённым данным
    return transform_receipt_columns(df)


# Загружаем эталонную разметку
label_dir = Path("data/SROIE2019/train/entities")
label_paths = sorted(label_dir.glob("*.txt"))[:num_receipts]

ground_truth_df = create_ground_truth_df(label_paths)
ground_truth_df

Проверим точность извлечения, сравнив результаты с эталоном.

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

from rapidfuzz import fuzz


def fuzzy_match_score(text1: str, text2: str) -> int:
    """Вычислить метрику сходства для двух строк (fuzzy matching)."""
    return fuzz.token_set_ratio(str(text1), str(text2))

Проверим нечёткое сопоставление на примерах названий компаний:

# Почти идентичные строки — высокий балл
print(f"Score: {fuzzy_match_score('BOOK TA K SDN BHD', 'BOOK TA .K SDN BHD'):.2f}")

# Другая пунктуация — совпадение всё ещё неплохое
print(f"Score: {fuzzy_match_score('MR D.I.Y. JOHOR', 'MR DIY JOHOR'):.2f}")

# Полностью разные строки — низкий балл
print(f"Score: {fuzzy_match_score('ABC TRADING', 'XYZ COMPANY'):.2f}")

Вывод:

Score: 97.14
Score: 55.17
Score: 27.27

Теперь напишем функцию сравнения, которая объединяет извлечённые данные с эталоном и применяет нечёткое сопоставление для названия компании и точное — для числовых полей:

def compare_receipts(
    extracted_df: pd.DataFrame,
    ground_truth_df: pd.DataFrame,
    id_column: str,
    fuzzy_match_cols: List[str],
    exact_match_cols: List[str],
    fuzzy_threshold: int = 80,
) -> pd.DataFrame:
    """Сравнить извлечённые данные с эталоном по заданным столбцам."""
    comparison_df = extracted_df.merge(
        ground_truth_df,
        on=id_column,
        how="inner",
        suffixes=("_extracted", "_truth"),
    )

    # Нечёткое сопоставление
    for col in fuzzy_match_cols:
        extracted_col = f"{col}_extracted"
        truth_col = f"{col}_truth"
        comparison_df[f"{col}_score"] = comparison_df.apply(
            lambda row: fuzzy_match_score(row[extracted_col], row[truth_col]),
            axis=1,
        )
        comparison_df[f"{col}_match"] = comparison_df[f"{col}_score"] >= fuzzy_threshold

    # Точное совпадение
    for col in exact_match_cols:
        extracted_col = f"{col}_extracted"
        truth_col = f"{col}_truth"
        comparison_df[f"{col}_match"] = (
            comparison_df[extracted_col] == comparison_df[truth_col]
        )

    return comparison_df


comparison_df = compare_receipts(
    extracted_df,
    ground_truth_df,
    id_column="receipt_id",
    fuzzy_match_cols=["company"],
    exact_match_cols=["total", "purchase_date"],
)

Посмотрим строки, где не совпали название компании, сумма или дата покупки:

def get_mismatch_rows(comparison_df: pd.DataFrame) -> pd.DataFrame:
    """Получить строки с несовпадениями, исключив служебные колонки с признаками совпадения."""
    # Столбцы c признаками совпадения и данные
    match_columns = [col for col in comparison_df.columns if col.endswith("_match")]
    data_columns = sorted([col for col in comparison_df.columns if col.endswith("_extracted") or col.endswith("_truth")])

    # Строки, где не все совпадения True
    has_mismatch = comparison_df[match_columns].all(axis=1).eq(False)

    return comparison_df[has_mismatch][data_columns]


mismatch_df = get_mismatch_rows(comparison_df)


mismatch_df

Это подтверждает наши наблюдения. Все чеки совпадают с эталонной разметкой, кроме чека с ID X51005200938, где расходятся следующие поля:

• Название компании

• Итоговая сумма

• Дата покупки

Давайте разберём этот чек подробнее и попытаемся понять, в чём проблема.

import IPython.display as display

file_to_inspect = receipt_dir / "X51005200938.jpg"

display.Image(filename=file_to_inspect)

Этот чек выглядит меньше остальных в датасете, что может повлиять на читаемость OCR. В следующем разделе мы увеличим масштаб изображения, чтобы улучшить извлечение.

Обрабатываем изображения для более точного извлечения

Создайте функцию для увеличения масштаба изображения:

from PIL import Image


def scale_image(image_path: Path, output_dir: Path, scale_factor: int = 3) -> Path:
    """Увеличить изображение с использованием высококачественного ресемплинга.

    Args:
        image_path: путь к исходному изображению
        output_dir: директория для сохранения увеличенного изображения
        scale_factor: во сколько раз увеличить изображение (по умолчанию 3x)

    Returns:
        Путь к увеличенному изображению
    """
    # Загружаем изображение
    img = Image.open(image_path)

    # Увеличиваем изображение с качественным ресемплингом
    new_size = (img.width * scale_factor, img.height * scale_factor)
    img_resized = img.resize(new_size, Image.Resampling.LANCZOS)

    # Сохраняем в выходную директорию с тем же именем файла
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / image_path.name
    img_resized.save(output_path, quality=95)

    return output_path

Примените функцию к проблемному чеку:

problematic_receipt_path = receipt_dir / "X51005200938.jpg"
adjusted_receipt_dir = Path("data/SROIE2019/train/img_adjusted")

scaled_image_path = scale_image(problematic_receipt_path, adjusted_receipt_dir, scale_factor=3)

Извлечём структурированные данные из увеличенного изображения:

problematic_structured_receipts = extract_documents([scaled_image_path], prompt)
problematic_extracted_df = create_extracted_df(problematic_structured_receipts)

problematic_extracted_df

Отлично! Масштабирование исправило извлечение. Название компании и дата покупки распознаны верно. Итоговая сумма 112.46 против 112.45 приемлема, так как на распечатанном чеке 112.45 действительно выглядит как 112.46.

Экспорт очищенных данных в CSV или Excel

Примените масштабирование ко всем чекам. Скопируйте оставшиеся изображения в директорию с обработанными файлами, исключив уже увеличенный чек:

import shutil

clean_receipt_paths = [scaled_image_path]
# Copy all receipts except the already processed one
for receipt_path in receipt_paths:
    if receipt_path != problematic_receipt_path:  # Skip the already scaled image
        output_path = adjusted_receipt_dir / receipt_path.name
        shutil.copy2(receipt_path, output_path)
        clean_receipt_paths.append(output_path)
        print(f"Copied {receipt_path.name}")

Запустим пайплайн снова с обработанными изображениями:

clean_structured_receipts = extract_documents(clean_receipt_paths, prompt)
clean_extracted_df = create_extracted_df(clean_structured_receipts)
clean_extracted_df

Результат:

Отлично! Теперь все чеки совпадают с эталонной разметкой.

Теперь можно экспортировать датасет в таблицу буквально за пару строк кода:

import pandas as pd

# Export to CSV
output_path = Path("reports/receipts.csv")
output_path.parent.mkdir(parents=True, exist_ok=True)
clean_extracted_df.to_csv(output_path, index=False)
print(f"Exported {len(clean_extracted_df)} receipts to {output_path}")

Вывод:

Exported 10 receipts to reports/receipts.csv

Экспортированные данные теперь можно импортировать в табличные редакторы, аналитические инструменты или BI-платформы.

Ускоряем обработку с помощью асинхронного параллельного выполнения

LlamaIndex поддерживает асинхронную обработку для параллельной работы с несколькими чеками. Используя async/await и метод aget_nodes_from_documents(), вы можете обрабатывать чеки параллельно, а не последовательно, заметно сокращая общее время.

Вот как изменить функцию извлечения для асинхронной обработки. Параметр num_workers=10 означает, что парсер будет обрабатывать до 10 чеков одновременно:

import asyncio


async def extract_documents_async(
    paths: List[str], prompt: str, id_column: str = "receipt_id"
) -> List[dict]:
    """Асинхронное извлечение структурированных данных из документов с помощью LlamaParse."""
    results: List[dict] = []

    parser = LlamaParse(
        api_key=os.environ["LLAMA_CLOUD_API_KEY"],
        result_type="markdown",
        num_workers=10,  # Обрабатывать до 10 чеков параллельно
        language="en",
        skip_diagonal_text=True,
    )

    # Асинхронная загрузка документов для параллельной обработки
    documents = await parser.aload_data(paths)

    program = OpenAIPydanticProgram.from_defaults(
        output_cls=Receipt,
        llm=Settings.llm,
        prompt_template_str=prompt,
    )

    for path, doc in zip(paths, documents):
        document_id = Path(path).stem
        parsed_document = program(context_str=doc.text)
        results.append({id_column: document_id, "data": parsed_document})

    return results


# Запуск через asyncio
structured_receipts = await extract_documents_async(receipt_paths, prompt)

Подробности смотрите в документации по асинхронному режиму LlamaIndex.

Попробуйте сами

Идеи из этого туториала оформлены как переиспользуемый пайплайн в этом репозитории на GitHub. В коде есть как синхронные, так и асинхронные версии:

Синхронные пайплайны (простая, последовательная обработка):

• Универсальный пайплайн (document_extraction_pipeline.py): повторно используемая функция извлечения, работающая с любой Pydantic-схемой

• Пайплайн для чеков (extract_receipts_pipeline.py): законченный пример со схемой Receipt, масштабированием изображений и преобразованиями данных

Асинхронные пайплайны (параллельная обработка с ускорением в 3–10 раз):

• Асинхронный универсальный пайплайн (async_document_extraction_pipeline.py): параллельная обработка документов

• Асинхронный пайплайн для чеков (async_extract_receipts_pipeline.py): пакетная обработка чеков с отслеживанием прогресса

Запустите пример извлечения чеков:

# Synchronous version (simple, sequential)
uv run extract_receipts_pipeline.py

# Asynchronous version (parallel processing, 3-10x faster)
uv run async_extract_receipts_pipeline.py

Либо создайте свой собственный экстрактор, импортировав extract_structured_data() и передав свою Pydantic-схему, промпт для извлечения и при необходимости функции предобработки.

Русскоязычное сообщество про Python

Друзья! Эту статью подготовила команда Python for Devs — канала, где каждый день выходят самые свежие и полезные материалы о Python и его экосистеме. Подписывайтесь, чтобы ничего не пропустить!

Заключение и следующие шаги

В этом туториале вы увидели, как LlamaIndex автоматизирует извлечение данных из чеков при минимальном количестве кода. Вы преобразовали отсканированные изображения в структурированные данные, проверили результаты по эталонной разметке и экспортировали «чистый» CSV, готовый к анализу.

Идеи для улучшения пайплайна извлечения чеков:

• Более богатые схемы: добавьте вложенные Pydantic-модели для деталей вендора, способов оплаты и построчных позиций

• Правила валидации: помечайте выбросы (например, суммы свыше $500 или будущие даты) для ручной проверки

• Многостадийные workflows: соберите пользовательский workflow, объединяющие предобработку изображений, извлечение, валидацию и экспорт с обработкой ошибок