Микрофреймворк Neuro Evolution: как создать десктопное GUI для обучения AI-агентов на Python с Gymnasium и wxPython

Надоело смотреть на строчки логов? Я тоже.

Ты запускаешь обучение агента в терминале. Episodes летят, reward скачет. Через час ты получаешь Excel-таблицу с цифрами. И всё. Агент — чёрный ящик. Ты не видишь, как он учится ходить, где застревает, почему падает в локальный минимум. Ты просто веришь в градиентный спуск.

Но есть другой путь. Свой собственный микрофреймворк, который рисует процесс обучения прямо на десктопе. Без облаков, без веб-сокетов, без дашбордов на React. Только Python, wxPython и gymnasium. Именно про это я рассказывал в статье про GUI-агентов — только там фокус был на том, как агент видит экран. А тут мы дадим агенту глаза снаружи.

1 В чём соль Neuro Evolution

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

Ключевая идея — не смешивать обучение и отрисовку в одном потоке. Если ты попытаешься вызывать env.step() внутри обработчика wx.Timer, твоя программа превратится в слайд-шоу с частотой кадра 0.5 fps. Обучение должно лететь на полной скорости в отдельном процессе, а GUI только подглядывает.

2 Выбор инструментов: почему именно они?

Кстати, про gymnasium. Если ты всё ещё используешь старый gym с 2021 года — срочно обновись. В новом Gymnasium переписали рендеринг, теперь rgb_array работает стабильно, а не через раз. Это критично, потому что мы будем передавать пиксели в GUI.

3 Архитектура Neuro Evolution (как не надо делать)

Самый популярный путь в ад — написать обучение в том же классе, что и окно. Вот антипример:

class MainFrame(wx.Frame):
    def __init__(self):
        # ...
        self.agent = DQNAgent()
        self.timer = wx.Timer(self)
        self.Bind(wx.EVT_TIMER, self.train_loop)
    
    def train_loop(self, event):
        action = self.agent.act(state)
        next_state, reward, done, _ = self.env.step(action)
        self.agent.remember(...)
        self.agent.replay()
        self.update_graph(reward)
        self.Refresh()

Правильная архитектура — разделение на три слоя:

• Ядро обучения — отдельный процесс (Worker). Он крутит цикл epizodov, делает шаги, обновляет нейросеть. Никакого wx, только gymnasium + torch/tensorflow.
• Канал связи — multiprocessing.Queue для данных (reward, loss, позиция) и multiprocessing.Array для скриншота среды (если нужно рендерить).
• GUI — wxPython, который раз в 50 мс забирает данные из очереди и перерисовывает графики/канвас.

4 Пишем код. Без воды.

Проект будет состоять из трёх файлов:

• worker.py — обучение агента в отдельном процессе.
• gui.py — главное окно с графиком и кнопками.
• main.py — точка входа.

Начнём с worker.py. Тут всё стандартно: среда Gymnasium, агент (пусть будет простой DQN из статьи От простого бота к самообучающемуся агенту), цикл обучения. Главный нюанс — пихать данные в очередь.

# worker.py
import gymnasium as gym
import numpy as np
from multiprocessing import Queue, Array
import time

class DQNAgent:
    # ... (ты знаешь, как писать DQN)

def train_worker(queue: Queue, screen_array: Array):
    env = gym.make('CartPole-v1', render_mode='rgb_array')
    agent = DQNAgent(state_size=4, action_size=2)
    episode = 0
    while True:
        state, _ = env.reset()
        total_reward = 0
        done = False
        while not done:
            action = agent.act(state)
            next_state, reward, done, truncated, info = env.step(action)
            agent.remember(state, action, reward, next_state, done)
            agent.replay()
            state = next_state
            total_reward += reward
            # каждые 10 шагов шлём данные в GUI
            if agent.step_count % 10 == 0:
                img = env.render()
                screen_array[:] = img.flatten()[:len(screen_array)]
                queue.put({
                    'episode': episode,
                    'step': agent.step_count,
                    'reward': total_reward,
                    'loss': agent.last_loss
                })
        episode += 1
        queue.put({'episode_done': episode, 'total_reward': total_reward})

Теперь gui.py. Тут главный фрейм, таймер, который читает из очереди, и Canvas для графика.

# gui.py
import wx
from multiprocessing import Queue, Array
import numpy as np

class TrainFrame(wx.Frame):
    def __init__(self, queue: Queue, screen_array: Array):
        super().__init__(None, title='Neuro Evolution - Training GUI', size=(900,600))
        self.queue = queue
        self.screen_array = screen_array
        self.rewards = []
        self.losses = []
        
        panel = wx.Panel(self)
        self.graph_panel = wx.Panel(panel, size=(600,400))
        self.graph_panel.Bind(wx.EVT_PAINT, self.on_paint_graph)
        
        self.fps_text = wx.StaticText(panel, label='FPS: --')
        btn_start = wx.Button(panel, label='Start')
        btn_start.Bind(wx.EVT_BUTTON, self.on_start)
        
        sizer = wx.BoxSizer(wx.VERTICAL)
        sizer.Add(self.graph_panel, 1, wx.EXPAND)
        sizer.Add(self.fps_text, 0, wx.ALL, 5)
        sizer.Add(btn_start, 0, wx.ALL, 5)
        panel.SetSizer(sizer)
        
        self.timer = wx.Timer(self)
        self.Bind(wx.EVT_TIMER, self.on_timer)
        self.timer.Start(100)  # 10 fps
        
    def on_timer(self, event):
        while not self.queue.empty():
            data = self.queue.get_nowait()
            if 'reward' in data:
                self.rewards.append(data['reward'])
            if 'loss' in data:
                self.losses.append(data['loss'])
            if 'episode_done' in data:
                self.fps_text.SetLabel(f'Episode {data["episode_done"]} done, reward {data["total_reward"]}')
        self.graph_panel.Refresh()
        
    def on_paint_graph(self, event):
        dc = wx.PaintDC(self.graph_panel)
        dc.SetBackground(wx.Brush('white'))
        dc.Clear()
        if len(self.rewards) < 2:
            return
        # рисуем simple line chart
        w, h = self.graph_panel.GetSize()
        dc.SetPen(wx.Pen('blue', 2))
        points = []
        for i, r in enumerate(self.rewards[-200:]):
            x = i / 200 * w
            y = h - (r - min(self.rewards[-200:])) / (max(self.rewards[-200:]) - min(self.rewards[-200:]) + 1e-6) * h
            points.append((x, y))
        dc.DrawLines(points)
        
    def on_start(self, event):
        # запускает worker-процесс (см. main.py)
        pass

И main.py — связующий клей.

# main.py
import wx
from multiprocessing import Queue, Array
import numpy as np
from worker import train_worker
from gui import TrainFrame
import multiprocessing as mp

if __name__ == '__main__':
    queue = Queue()
    # Screen size CartPole: 600x400 = 240000 bytes
    screen_array = Array('B', 600*400*3, lock=False)
    
    worker_process = mp.Process(target=train_worker, args=(queue, screen_array))
    worker_process.start()
    
    app = wx.App()
    frame = TrainFrame(queue, screen_array)
    frame.Show()
    app.MainLoop()
    
    worker_process.terminate()

Вот и всё. Запускаешь — и видишь, как reward ползёт вверх, а агент учится балансировать палку. Без джупитеров, без дашбордов.

5 Типичные ошибки и как их обойти

Ошибка 1: GUI висит. Ты забыл про wx.Yield() или долго обрабатываешь очередь. Решение: не делай сложной логики в таймере. Только чтение из очереди и перерисовка.

Ошибка 2: Очередь переполняется. Воркер шлёт данные каждый шаг, а GUI не успевает. Очередь растёт, память утекает. Решение: поставь максимальный размер очереди (Queue(maxsize=100)) и используй put_nowait с exc, сбрасывая старые данные.

Ошибка 3: Скриншоты не совпадают по размеру. При смене размера окна Gymnasium рендерит другое разрешение. Фикс: зафиксировать render_mode='rgb_array' и заранее задать размер окна через env.metadata['render_fps'] или свой wrapper.

Ошибка 4: Процесс-воркер не завершается. После закрытия окна он продолжает жрать CPU. Решение: подписываемся на wx.EVT_CLOSE и ставим флаг mp.Event, который воркер проверяет после каждого эпизода. Если не хочешь заморачиваться — worker_process.terminate().

6 Куда это применить? И что дальше?

Описаный микрофреймворк — скелет. На его основе можно делать обучение с подкреплением для любых сред: Lunar Lander, Atari, MuJoCo. Добавь сохранение модели каждые N эпизодов — получишь полноценный тренажёр для RL. А если обернуть это в Docker и дать REST API — выйдет продукт, как в статье про дообучение на EC2, только наоборот: модель обучается локально, а GUI показывает прогресс.

Параллельное обучение нескольких агентов — следующий шаг. Запусти 4-5 воркеров с разными гиперпараметрами, собирай статистику в общую базу. Это уже напоминает federated learning из материала про Flower Framework, только вместо глобальной агрегации у тебя — гонка агентов на глазах у пользователя.

Ещё один вектор — интегрировать визуализацию нейронной сети, как в статье про continual learning на микронейросетях. Рисуем веса в реальном времени, видим, как меняются связи. Это превращает обучение из рутины в искусство.

7 Немного субъективного. Зачем я это написал?

Потому что мне надоело смотреть на цифры. Когда агент учится ходить — это красиво. А когда он ходит только на бумаге — это грустно. Настоящее обучение должно быть зрелищным. Я верю, что в 2026 году, когда облачные дашборды стали дорогими, а данные хочется держать локально, десктопные GUI переживут ренессанс. Не верите? Посмотрите на статью GUI-агенты: как они видят — там та же идея, но с другой стороны.

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