Курсы

Appearance

11. Продвинутая кастомизация «под продакшн»: Аннотации и интеграция в Celery ​

Аннотирование и точечные акценты ​

Роль выносных указателей в дебаге моделей ​

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

Для этой задачи в Matplotlib реализован мощный и гибкий метод координатной сетки ax.annotate(text, xy, xytext, arrowprops):

  • text — строковое содержимое вашего комментария.
  • xy — кортеж координат (x, y) на графике, на которые должна строго указывать стрелка.
  • xytext — кортеж координат (x, y), где физически должен располагаться сам текст надписи.
  • arrowprops — словарь настроек внешнего вида выносной стрелки (её стиль, цвет и толщина).
python
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()
ax.plot(, [1.5, 0.2, 0.4])

# Adding a precision text annotation pointer to highlight the local minimum point
ax.annotate(
    text="Best Model Weights Saved", 
    xy=(1.0, 0.2), 
    xytext=(1.5, 0.8),
    arrowprops=dict(facecolor="black", shrink=0.05, width=1)
)
ax.grid(True)

Промышленная интеграция в бэкенд-пайплайны ​

Рендеринг в бинарные байтовые потоки без сохранения на диск ​

В промышленной enterprise-разработке (в Celery-воркерах, микросервисах Faststream/FastAPI) сохранять сгенерированные графики в виде промежуточных физических файлов на жесткий диск сервера — плохая практика. Дисковый ввод-вывод (I/O) работает медленно, а файлы засоряют хранилище Docker-контейнера.

Для оптимизации процесса скомпилированный холст из оперативной памяти отправляют напрямую в асинхронную сеть или TensorBoard, упаковывая его в виртуальный бинарный поток в ОЗУ с помощью стандартного модуля Python io.BytesIO. Метод fig.savefig() поддерживает передачу такого байтового буфера вместо строкового пути к файлу.

python
import io
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()
ax.plot(, [5, 10])

# 1. Instantiating an isolated binary buffer block in RAM memory
binary_buffer_stream = io.BytesIO()

# 2. Saving the canvas data model straight to the RAM bytes stream
fig.savefig(binary_buffer_stream, format="png")

# 3. Rewinding the stream pointer to the beginning for safe reading
binary_buffer_stream.seek(0)

# 4. Extracting raw bytes bytes string data blocks array
raw_bytes_data = binary_buffer_stream.read()
# Now raw_bytes_data can be pushed natively to S3 storage, Telegram API, or TensorBoard

Освобождение памяти холста методом fig.close() ​

Метод fig.savefig() лишь сериализует байты холста в файл, но сам объект fig продолжает жить в оперативной памяти процесса. В интерактивной сессии это незаметно, однако в долгоживущем воркере Celery каждый цикл «создать график → сохранить PNG → передать в OSS» накапливает в памяти тысячи холстов, пока процесс не упрется в лимит контейнера и не упадет с OOM.

Решение — явно закрывать холст сразу после экспорта:

python
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot

# Long-running Celery worker renders a loss snapshot on every training job
fig = matplotlib.pyplot.figure()
ax = fig.add_subplot(1, 1, 1)
ax.plot([1, 2, 3], [0.9, 0.5, 0.2])
fig.savefig("job_loss_snapshot.png")

# Releasing the canvas memory so the worker never accumulates stale figures
fig.close()

Упражнения базовые (5 шт) (только с опорой на объясненную в данном уроке теорию) ​

1. Добавление простейшей текстовой аннотации ​

Задача: Создайте график. С помощью метода ax.annotate() добавьте текстовую надпись "Anomaly Peak" в точку с координатами (2.0, 10.0), расположив сам текст в координатах (3.0, 12.0). Настройку стрелки arrowprops пока оставьте пустой.

Посмотреть решение
python
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()

# Adding plain absolute text annotation pointer coordinate anchors
ax.annotate(text="Anomaly Peak", xy=(2.0, 10.0), xytext=(3.0, 12.0))
ax.grid(True)

2. Аннотирование с кастомизацией стиля стрелки ​

Задача: Повторите аннотирование точки xy=(1.0, 0.5). Расположите текст в xytext=(2.0, 1.0), подписав его как "Minimum Loss". В параметр arrowprops передайте словарь настроек со стрелкой черного цвета: dict(facecolor="black", width=2).

Посмотреть решение
python
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()

# Configuring arrow properties configuration dictionary inside annotation sub-node
ax.annotate(
    text="Minimum Loss", 
    xy=(1.0, 0.5), 
    xytext=(2.0, 1.0),
    arrowprops=dict(facecolor="black", width=2)
)

3. Рендеринг холста в байтовый поток io.BytesIO ​

Задача: Настройте Matplotlib для работы в headless-режиме. Создайте холст и осей. Инициализируйте пустой байтовый поток io.BytesIO(), сохраните в него холст в формате "png" и выведите на экран размер полученного байтового буфера с помощью метода .getbuffer().nbytes.

Посмотреть решение
python
import io
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()

# Capturing frame bytes stream without disk footprint operations
ram_buffer = io.BytesIO()
fig.savefig(ram_buffer, format="png")

print("Total physical size inside RAM buffer stream:", ram_buffer.getbuffer().nbytes, "bytes")

4. Сброс указателя потока seek(0) после рендеринга ​

Задача: Напишите фрагмент кода бэкенд-пайплайна, который принимает объект потока bytes_io, сохраняет в него график через метод .savefig() и вызывает команду сброса каретки указателя чтения в начальное нулевое положение, подготавливая буфер к отправке по сети.

Посмотреть решение
python
import io
import matplotlib.pyplot

fig, ax = matplotlib.pyplot.subplots()
bytes_io = io.BytesIO()

fig.savefig(bytes_io, format="png")
# Essential rewind statement step to allow reading data back from the stream
bytes_io.seek(0)

5. Извлечение сырых байт методом .read() ​

Задача: На основе подготовленного и сброшенного в начало потока bytes_io из прошлого упражнения, вызовите метод полного считывания содержимого в переменную raw_binary_string. Выведите тип полученного объекта через встроенную функцию type().

Посмотреть решение
python
import io

bytes_io = io.BytesIO()
# ... code that populates and seeks the stream ...

# Fetching bytes string data layer from hardware RAM memory buffers
raw_binary_string = bytes_io.read()
print("Extracted variable data type classification:", type(raw_binary_string))  # <class 'bytes'>