Своя симуляция

Полный контракт среды #

Среда — это класс (или функция-фабрика) с тремя требованиями:

1. reset() — начало эпизода, возвращает начальное наблюдение.
2. step(action) — один шаг, возвращает 4- или 5-tuple.
3. n_actions — число действий, объявленное явно.

Минимальная рабочая среда:

class MyEnv:
    n_actions = 4   # ОБЯЗАТЕЛЬНО

    def reset(self):
        self.pos = [0, 0]
        return [float(self.pos[0]), float(self.pos[1])]

    def step(self, action: int):
        dx, dy = [(0,1),(0,-1),(-1,0),(1,0)][action]
        self.pos[0] += dx
        self.pos[1] += dy
        done = (self.pos == [5, 5])
        reward = 1.0 if done else -0.01
        return [float(self.pos[0]), float(self.pos[1])], reward, done, {}

def make_env():
    return MyEnv()

reset() #

Вызывается в начале каждого эпизода. Возвращает начальное наблюдение.

Кузня принимает два формата:

• Просто наблюдение: return obs
• Gymnasium-стиль: return (obs, info) — info игнорируется.

step(action) #

Один шаг симуляции. Поддерживаются два формата возврата:

return obs, reward, done, info

return obs, reward, terminated, truncated, info

Формат определяется автоматически по длине возвращаемого tuple.

n_actions — число действий #

Три способа объявить (в порядке приоритета):

# Вариант 1: action_space.n (приоритет 1)
from types import SimpleNamespace
class MyEnv:
    action_space = SimpleNamespace(n=4)

# Вариант 2: атрибут n_actions (рекомендуется)
class MyEnv:
    n_actions = 4

# Вариант 3: словарь spec
class MyEnv:
    spec = {"n_actions": 4, "obs_dim": 8}

obs_dim — размерность наблюдения #

Обнаруживается автоматически из формы первого наблюдения reset(). Явно задавать не нужно — Кузня выведет из данных.

Требования к наблюдению:

• Любой объект, который numpy преобразует в float32-массив: список чисел, numpy-массив, tuple.
• Форма должна быть одинаковой на всех шагах. Изменение формы между шагами — ошибка.

Награда (reward) #

• Число (int или float).
• NaN и inf запрещены — такой шаг немедленно прерывает ран с чёткой ошибкой.
• Плотная (dense) награда обучается значительно быстрее разреженной.

Изоляция и таймауты #

Когда simulation.timeout_seconds > 0, среда запускается в отдельном процессе через multiprocessing.spawn. Общение — через Pipe.

Главный процесс                    Subprocess (ваша среда)
    │                                       │
    ├── send(("reset",))  ──────────────►  driver.reset()
    │                      ◄──────────────  send(("ok", obs))
    │
    ├── send(("step", 3))  ──────────────►  driver.step(3)
    │                      ◄──────────────  send(("ok", obs, reward, done, info))

Если ответ не пришёл за timeout_seconds — подпроцесс принудительно завершается.

Когда выключать изоляцию

timeout_seconds = 0 запускает среду в процессе — без overhead subprocess:

"simulation": {
    "timeout_seconds": 0
}

Используйте in-process если:

• Ваша среда не имеет бесконечных циклов.
• Вы доверяете коду среды.
• Скорость критична (тысячи шагов в секунду).

Захват вывода

Все print() внутри среды перехватываются — они не попадают в stdout приложения. Доступны через handle.logs и отображаются в UI.

kuz sim — отладка #

Перед запуском обучения всегда проверяйте среду отдельно:

python3 kuz.py sim sims/my_env.py

Вывод при успехе:

  проверяю sims/my_env.py…
  ✓ симуляция рабочая
    obs_dim:   4
    n_actions: 4
    reward:    [-0.22, -0.02]
    · n_actions взят из атрибута n_actions
    · obs_dim выведен из формы первого наблюдения: (4,)

С кастомной точкой входа:

python3 kuz.py sim sims/my_env.py --entry MyEnvClass

Метод render() — реплей эпизода #

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

Возвращаемый формат — любой словарь, который сериализуется в JSON:

def render(self):
    # Gridworld-пример: сетка символов
    grid = [["." for _ in range(self.width)] for _ in range(self.height)]
    for wx, wy in self.walls:
        grid[wy][wx] = "#"
    gx, gy = self.goal
    grid[gy][gx] = "G"
    ax, ay = self.pos
    grid[ay][ax] = "A"
    return {
        "grid": grid,
        "agent": [int(ax), int(ay)],
        "goal": [int(gx), int(gy)],
        "steps": int(self.steps)
    }

Написать среду с нуля: пошаговый туториал #

Задача: агент учится идти вправо на числовой прямой.

1. 1

   Определите задачу

   • Состояние: позиция агента (1 число) и расстояние до цели (1 число) → obs_dim = 2.
   • Действия: 0 = влево, 1 = вправо → n_actions = 2.
   • Цель: достичь позиции 10. Горизонт: 50 шагов.
2. 2

   Напишите класс

   python — sims/line_walk.py копировать

   class LineWalk:
       n_actions = 2           # ОБЯЗАТЕЛЬНО
   
       def __init__(self):
           self.pos = 0
           self.goal = 10
           self.steps = 0
   
       def reset(self):
           self.pos = 0
           self.steps = 0
           # Нормализованное наблюдение: позиция и расстояние до цели
           return [self.pos / self.goal,
                   (self.goal - self.pos) / self.goal]
   
       def step(self, action):
           self.steps += 1
           prev_dist = abs(self.goal - self.pos)
           self.pos += 1 if action == 1 else -1
           new_dist = abs(self.goal - self.pos)
   
           # Плотная награда: +1 за приближение, -1 за удаление
           reward = 1.0 if new_dist < prev_dist else -1.0
   
           done = (self.pos == self.goal)
           if done:
               reward += 5.0  # бонус за цель
   
           return ([self.pos / self.goal,
                    (self.goal - self.pos) / self.goal],
                   reward, done, {"pos": self.pos})
   
   def make_env():
       return LineWalk()

3. 3

   Проверьте среду вручную

   python копировать

   if __name__ == "__main__":
       env = make_env()
       obs = env.reset()
       total = 0
       for _ in range(50):
           obs, r, done, info = env.step(1)  # всегда вправо
           total += r
           if done:
               break
       print(f"pos={info['pos']}, total_reward={total:.1f}")

4. 4

   Проверьте через kuz sim

   bash копировать

   python3 kuz.py sim sims/line_walk.py

5. 5

   Запустите обучение

   bash копировать

   # Создать конфиг из RL-пресета
   kuz init walk.json --preset gridworld
   
   # Отредактировать walk.json:
   # заменить "path": "sims/gridworld.py" на "path": "sims/line_walk.py"
   
   # Запустить обучение
   kuz train walk.json

Типичные ошибки #

Контракт obs_dim при изоляции #

В subprocess-режиме объект среды не передаётся через pickle — только числа. obs_dim и n_actions читаются внутри дочернего процесса через атрибуты. Если obs_dim нельзя определить до первого reset(), а класс не объявляет obs_dim/observation_space — всё равно работает: Кузня выведет его из формы первого наблюдения.

Встроенные примеры #