FSM в боте MAX: как хранить состояние пользователя

Любой нелинейный сценарий в боте — это конечный автомат. Пользователь идёт по шагам: ввод имени, ввод телефона, подтверждение. Между шагами бот должен помнить, на каком этапе он находится и какие данные уже собрал. Эту память называют FSM, и от её устойчивости зависит, будет ли бот работать в проде или ронять диалоги при каждом перезапуске. В этой статье разбираем, как правильно собрать FSM в боте MAX от российского VK Tech: где хранить, как масштабировать, как тестировать и каких граблей избегать.

Что такое FSM в контексте бота

Конечный автомат (Finite State Machine, FSM) — это математическая модель, которая в каждый момент времени находится ровно в одном из конечного набора состояний. Переходы между состояниями определяются входящими событиями. В боте MAX состояние = «на каком шаге сценария находится конкретный пользователь», переход — это сообщение, callback кнопки или внешнее событие, которое двигает пользователя дальше по графу.

Формально FSM описывается пятёркой:

• S — множество состояний (idle, awaiting_name, awaiting_phone, done)
• s0 — начальное состояние (обычно idle)
• Σ — алфавит входных событий (текст сообщения, нажатие кнопки, таймер)
• δ — функция перехода (state, event) -> new_state
• F — множество финальных состояний

Для бота важно ещё одно — контекст (накопленные данные). Чистый FSM их не предусматривает, поэтому на практике говорят о расширенном автомате (Extended FSM), где к состоянию добавляется словарь собранных значений.

Зачем FSM в боте MAX

Простые FAQ-боты обходятся без FSM: запрос — ответ, без контекста. Но как только появляется хоть какой-то многошаговый сценарий, нужен FSM. Типичные кейсы:

• Регистрация в 5 шагов — имя, телефон, email, согласие на обработку ПДн, подтверждение через код.
• Оформление заказа — выбор товара, количество, адрес доставки, способ оплаты, подтверждение.
• Запись на услугу — мастер, услуга, дата, время, контакты, согласование.
• Опросы и квизы — последовательность вопросов с зависимостями (если ответил «да», задать дополнительный).
• Подача заявки в HR-бот — резюме, вакансия, мотивация, тестовое задание.
• Мульти-этапная техподдержка — категория проблемы, продукт, описание, скриншот, контакт.

Без FSM код такого бота превращается в гигантский if/elif по тексту последнего сообщения с проверками «а что было до этого». FSM выносит эту логику в явный граф, который можно нарисовать, обсудить с продуктом и протестировать.

Состояния, переходы, события — теория

Состояние должно быть атомарным и осмысленным. Плохо: step_3. Хорошо: awaiting_phone_confirmation. Имя состояния — документация. Через полгода вы сами не вспомните, что значит step_3.

Переход — это всегда тройка (from_state, event, to_state) плюс опциональные guard (условие) и action (побочный эффект). Например:

• (awaiting_phone, text_message, awaiting_confirmation) с guard is_valid_phone(text) и action save_phone_to_context().
• (awaiting_phone, text_message, awaiting_phone) с guard not is_valid_phone(text) и action send_error_message().

События в боте MAX:

• Текстовое сообщение (message.text).
• Нажатие inline-кнопки (callback_query.data).
• Загрузка файла или фото (message.attachments).
• Команды (/start, /cancel, /help).
• Внешний триггер (например, webhook от платёжной системы).
• Таймер (TTL истёк, нужно вычистить состояние).

Граф переходов FSM регистрации в ASCII:

        +--------+
        |  idle  |
        +---+----+
            | /start
            v
   +--------+--------+
   | awaiting_name   |
   +--------+--------+
            | text (valid name)
            v
   +--------+--------+
   | awaiting_phone  |<--+
   +--------+--------+   | text (invalid)
            | text       |
            | (valid)    |
            v            |
   +--------+--------+   |
   | awaiting_email  |   |
   +--------+--------+   |
            | text       |
            v            |
   +--------+--------+   |
   | awaiting_pdn    |---+ "нет"
   +--------+--------+
            | "да"
            v
   +--------+--------+
   |   confirmed     |
   +-----------------+

Реализация FSM в боте MAX: подходы

В экосистеме MAX (VK Tech) пока нет такого зоопарка фреймворков, как в Telegram, поэтому FSM чаще пишут руками поверх HTTP-клиента к Bot API. Три рабочих подхода:

1. Сторадж + диспетчер. Перед каждым апдейтом достаём состояние из стораджа по user_id, передаём в обработчик, обработчик возвращает новое состояние и данные. Самый прямолинейный.
2. Контекстный объект (FSMContext-стиль из aiogram). Хендлер получает уже заинжекченный контекст с методами get_state(), set_state(), update_data(). Удобнее в коде, но нужно писать обвязку.
3. Очередь событий (event sourcing для FSM). Сохраняем не состояние, а историю событий. State = свёртка (fold) всех событий пользователя. Полезно для аналитики и аудита, но избыточно для обычных ботов.

Большинство продакшен-ботов в MAX используют подход 2.

Минимальная реализация FSM на Python

Базовый класс стораджа и диспетчер. Без бот-фреймворка, чтобы было видно механику.

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
import json
import time

@dataclass
class FSMContext:
    state: str | None = None
    data: dict[str, Any] = field(default_factory=dict)
    updated_at: float = field(default_factory=time.time)

class BaseStorage(ABC):
    @abstractmethod
    async def get(self, user_id: int) -> FSMContext: ...

    @abstractmethod
    async def set(self, user_id: int, ctx: FSMContext) -> None: ...

    @abstractmethod
    async def clear(self, user_id: int) -> None: ...

class MemoryStorage(BaseStorage):
    """Только для разработки. Не использовать в проде."""
    def __init__(self):
        self._data: dict[int, FSMContext] = {}

    async def get(self, user_id: int) -> FSMContext:
        return self._data.get(user_id, FSMContext())

    async def set(self, user_id: int, ctx: FSMContext) -> None:
        ctx.updated_at = time.time()
        self._data[user_id] = ctx

    async def clear(self, user_id: int) -> None:
        self._data.pop(user_id, None)

MemoryStorage живёт в памяти процесса и теряет данные при рестарте. Подходит только для локальной разработки и тестов.

Хранилища состояния: сравнение

Какой сторадж выбрать — зависит от нагрузки, требований к надёжности и того, что уже есть в инфраструктуре.

Для типичного MAX-бота под нагрузкой до тысячи активных диалогов в минуту хватает Redis с одной репликой и AOF=everysec.

RedisStorage: продакшен-вариант

Реализация стораджа на Redis с TTL. Ключи именуются с префиксом и версией, чтобы можно было катить миграции без потери легитимных сессий.

import json
import redis.asyncio as redis

class RedisStorage(BaseStorage):
    def __init__(
        self,
        url: str = "redis://localhost:6379/0",
        prefix: str = "fsm:v1",
        ttl_seconds: int = 1800,  # 30 минут
    ):
        self._redis = redis.from_url(url, decode_responses=True)
        self._prefix = prefix
        self._ttl = ttl_seconds

    def _key(self, user_id: int) -> str:
        return f"{self._prefix}:user:{user_id}"

    async def get(self, user_id: int) -> FSMContext:
        raw = await self._redis.get(self._key(user_id))
        if raw is None:
            return FSMContext()
        payload = json.loads(raw)
        return FSMContext(
            state=payload.get("state"),
            data=payload.get("data", {}),
            updated_at=payload.get("updated_at", time.time()),
        )

    async def set(self, user_id: int, ctx: FSMContext) -> None:
        ctx.updated_at = time.time()
        payload = json.dumps({
            "state": ctx.state,
            "data": ctx.data,
            "updated_at": ctx.updated_at,
        })
        await self._redis.set(self._key(user_id), payload, ex=self._ttl)

    async def clear(self, user_id: int) -> None:
        await self._redis.delete(self._key(user_id))

Ключи получаются вида fsm:v1:user:1234567. При смене схемы FSM поднимаем версию в prefix (fsm:v2), старые сессии естественным образом протухают по TTL. Никаких «миграций» БД не нужно.

PostgreSQL custom storage

Если Redis в инфре нет, а Postgres есть — можно жить и на нём. Производительнее, чем кажется, особенно если завести отдельную таблицу с JSONB и индексом на user_id + updated_at.

CREATE TABLE fsm_state (
    user_id     BIGINT PRIMARY KEY,
    state       TEXT,
    data        JSONB NOT NULL DEFAULT '{}'::jsonb,
    version     TEXT NOT NULL DEFAULT 'v1',
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_fsm_updated_at ON fsm_state (updated_at);

Сам сторадж на asyncpg:

import asyncpg

class PostgresStorage(BaseStorage):
    def __init__(self, pool: asyncpg.Pool, version: str = "v1"):
        self._pool = pool
        self._version = version

    async def get(self, user_id: int) -> FSMContext:
        async with self._pool.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT state, data, updated_at FROM fsm_state "
                "WHERE user_id = $1 AND version = $2",
                user_id, self._version,
            )
        if row is None:
            return FSMContext()
        return FSMContext(
            state=row["state"],
            data=dict(row["data"]),
            updated_at=row["updated_at"].timestamp(),
        )

    async def set(self, user_id: int, ctx: FSMContext) -> None:
        async with self._pool.acquire() as conn:
            await conn.execute(
                "INSERT INTO fsm_state (user_id, state, data, version) "
                "VALUES ($1, $2, $3::jsonb, $4) "
                "ON CONFLICT (user_id) DO UPDATE "
                "SET state = $2, data = $3::jsonb, version = $4, "
                "    updated_at = NOW()",
                user_id, ctx.state, json.dumps(ctx.data), self._version,
            )

    async def clear(self, user_id: int) -> None:
        async with self._pool.acquire() as conn:
            await conn.execute(
                "DELETE FROM fsm_state WHERE user_id = $1", user_id,
            )

Очистка зависших сессий — отдельная джоба раз в 10 минут:

DELETE FROM fsm_state
WHERE updated_at < NOW() - INTERVAL '30 minutes';

Несколько реплик бота: почему нужен внешний storage

Как только вы поднимаете больше одного инстанса бота за балансировщиком (или просто за nginx со стратегией round-robin к двум контейнерам), in-process состояние перестаёт работать. Пользователь отправил имя — попал на реплику A, она запомнила. Отправил телефон — попал на реплику B, она ничего не знает и спрашивает имя заново.

Решения два, и оба сводятся к внешнему стораджу:

1. Sticky sessions — балансировщик закрепляет пользователя за конкретной репликой по user_id. Работает, но при падении реплики все её диалоги умирают. И масштабироваться сложнее.
2. Внешнее хранилище (Redis/Postgres) — любая реплика может обработать любой апдейт, состояние шарится. Это правильный путь.

Для MAX-бота на webhook'ах обычно сразу закладывают внешний сторадж, даже если пока одна реплика — миграция позже больнее, чем сделать сразу.

TTL состояний и очистка зависших сессий

Пользователь начал заполнять заявку, отвлёкся на час, вернулся. Что должен сделать бот? Ответ зависит от типа сценария.

• Long TTL (24 часа - неделя) — длинные анкеты, оформление крупных заказов, КИК. Удобно пользователю: вернулся через сутки и продолжил.
• Medium TTL (1-4 часа) — обычная регистрация, запись на услугу. Баланс между удобством и свежестью.
• Short TTL (15-30 минут) — быстрые сценарии: купить билет, заказать пиццу. После перерыва логичнее начать заново, чем дописывать вчерашний заказ.

Можно делать ступенчатую логику: на 15-й минуте отправить напоминание «Вы хотели оформить заказ. Продолжим?» с inline-кнопками «да / отмена». Если ответа нет ещё 15 минут — clear().

В Redis TTL встроен (SET ... EX 1800). В Postgres — фоновый воркер с DELETE WHERE updated_at < NOW() - INTERVAL '30 minutes'.

Откат назад, отмена и глобальные команды

В реальном FSM пользователю нужно давать выйти. Минимум:

• /cancel — сбрасывает текущее состояние, возвращает в idle.
• /start — то же самое плюс показывает приветствие.
• Кнопка «Назад» в каждом шаге — переводит в предыдущее состояние без потери уже введённых данных.

Глобальные команды реализуются как middleware до FSM-диспетчера: если апдейт = /cancel, чистим стейт и не передаём дальше. Так пользователь всегда может выйти из любого зацикленного сценария.

«Назад» сложнее. Если состояний 5, на каждое нужен явный переход в предыдущее. Хранить историю переходов в data["_history"] — рабочее решение:

async def go_back(user_id: int, storage: BaseStorage) -> None:
    ctx = await storage.get(user_id)
    history = ctx.data.get("_history", [])
    if not history:
        return
    prev_state = history.pop()
    ctx.state = prev_state
    ctx.data["_history"] = history
    await storage.set(user_id, ctx)

При каждом forward-переходе пушим текущее состояние в _history, при back — попаем.

Вложенные сценарии и как их избегать

Соблазн: «давайте в середине регистрации запустим под-сценарий выбора города». Реализуется через стек состояний (data["_stack"]), но превращает FSM в визуальный ад. Через полгода никто не разберётся, как пользователь оказался в awaiting_city_in_registration_in_form.

Лучшие практики:

• Линейные сценарии без вложенности. Если сценарий длинный — разбейте на этапы и сохраняйте промежуточные результаты.
• Композиция вместо вложения: завершите внешний сценарий, сохраните partial-результат, запустите второй.
• Явный parent_flow в data — если без вложенности никак, фиксируйте, в какой сценарий вернуться, явным полем.

Тестирование FSM

FSM — отличный кандидат на юнит-тесты. Логика переходов чистая, без сетевых вызовов, без MAX API. Тестируем функцию transition(state, event, data) -> (new_state, new_data).

import pytest

@pytest.mark.asyncio
async def test_registration_happy_path():
    storage = MemoryStorage()
    user_id = 42

    # /start
    await handle_command(user_id, "/start", storage)
    ctx = await storage.get(user_id)
    assert ctx.state == "awaiting_name"

    # имя
    await handle_text(user_id, "Иван Петров", storage)
    ctx = await storage.get(user_id)
    assert ctx.state == "awaiting_phone"
    assert ctx.data["name"] == "Иван Петров"

    # невалидный телефон
    await handle_text(user_id, "не телефон", storage)
    ctx = await storage.get(user_id)
    assert ctx.state == "awaiting_phone"  # остались на месте
    assert "phone" not in ctx.data

    # валидный телефон
    await handle_text(user_id, "+79991234567", storage)
    ctx = await storage.get(user_id)
    assert ctx.state == "awaiting_email"
    assert ctx.data["phone"] == "+79991234567"

Интеграционный слой (с реальным Redis в docker-compose) тестирует, что сторадж не теряет данные и корректно работает TTL. Юнит-тесты на FSM-логику — на каждом PR, интеграционные — на ночной CI.

Большой пример: FSM регистрации с 5 шагами

Соберём всё вместе. Регистрация: имя → телефон → email → согласие на ПДн → подтверждение.

from enum import Enum

class RegState(str, Enum):
    IDLE = "idle"
    NAME = "awaiting_name"
    PHONE = "awaiting_phone"
    EMAIL = "awaiting_email"
    PDN = "awaiting_pdn"
    CONFIRMED = "confirmed"

import re
PHONE_RE = re.compile(r"^\+?[78]\d{10}$")
EMAIL_RE = re.compile(r"^[\w.+-]+@[\w-]+\.[\w.-]+$")

async def handle_update(
    user_id: int,
    text: str,
    storage: BaseStorage,
    bot,
) -> None:
    ctx = await storage.get(user_id)

    # глобальная команда
    if text == "/cancel":
        await storage.clear(user_id)
        await bot.send(user_id, "Регистрация отменена.")
        return

    if text == "/start" or ctx.state in (None, RegState.IDLE):
        ctx.state = RegState.NAME
        ctx.data = {}
        await storage.set(user_id, ctx)
        await bot.send(user_id, "Привет! Как вас зовут?")
        return

    if ctx.state == RegState.NAME:
        if len(text) < 2 or len(text) > 80:
            await bot.send(user_id, "Имя 2-80 символов. Повторите.")
            return
        ctx.data["name"] = text
        ctx.state = RegState.PHONE
        await storage.set(user_id, ctx)
        await bot.send(user_id, "Введите телефон в формате +7XXXXXXXXXX")
        return

    if ctx.state == RegState.PHONE:
        if not PHONE_RE.match(text):
            await bot.send(user_id, "Невалидный телефон. Повторите.")
            return
        ctx.data["phone"] = text
        ctx.state = RegState.EMAIL
        await storage.set(user_id, ctx)
        await bot.send(user_id, "Введите email")
        return

    if ctx.state == RegState.EMAIL:
        if not EMAIL_RE.match(text):
            await bot.send(user_id, "Невалидный email. Повторите.")
            return
        ctx.data["email"] = text
        ctx.state = RegState.PDN
        await storage.set(user_id, ctx)
        await bot.send(
            user_id,
            "Согласны на обработку персональных данных? (да/нет)",
        )
        return

    if ctx.state == RegState.PDN:
        if text.strip().lower() not in ("да", "yes"):
            await storage.clear(user_id)
            await bot.send(user_id, "Без согласия не можем продолжить.")
            return
        ctx.data["pdn_accepted_at"] = time.time()
        ctx.state = RegState.CONFIRMED
        await storage.set(user_id, ctx)
        await bot.send(
            user_id,
            f"Готово! Имя: {ctx.data['name']}, телефон: {ctx.data['phone']}",
        )
        # передаём в CRM
        await crm.create_lead(ctx.data)
        await storage.clear(user_id)
        return

Структура читается сверху вниз как граф. Каждое состояние — независимый блок. Добавить шаг = добавить enum + блок.

Альтернативы FSM

FSM не всегда нужен. Если у бота десяток независимых команд без многошаговых сценариев — хватит роутера (диспетчера команд) и stateless-обработчиков. Признаки, что FSM избыточен:

• Каждый запрос самодостаточен (/weather Москва, /курс USD).
• Нет накопления контекста между сообщениями.
• Кнопки навигации, а не пошаговые формы.

В этом случае добавление FSM только усложнит код. Берите его, когда появляется хотя бы один сценарий из 3+ зависимых шагов.

Антипаттерны

Что ломает FSM в проде:

• Глобальное состояние на всех пользователей — классика на хакатонах. Каждому юзеру — свой user_id-ключ, никаких current_step модульного уровня.
• Shared mutable state между корутинами без блокировок — гонки при двух одновременных апдейтах от одного юзера. Используйте per-user lock в Redis (SET NX EX 5) или сериализацию через очередь.
• Хранение в chat_data фреймворка без персистентности — рестарт = потеря.
• Смешивание FSM и бизнес-логики в одном классе — превращается в неподдерживаемую кашу. Разделяйте: FSM знает про переходы, бизнес-сервисы — про CRM/платежи/БД.
• Отсутствие валидации на каждом шаге — пользователь введёт «мама» вместо телефона, и заявка уйдёт битой.
• Игнорирование TTL — Redis постепенно забивается мёртвыми сессиями, RAM течёт.
• Версионирование через миграцию данных — проще поднять prefix и дать старым сессиям умереть.

Итого

FSM — фундамент любого нетривиального бота в MAX. Корректное хранение состояния (в подавляющем большинстве случаев Redis), TTL и обработка таймаутов, версионирование через префикс ключа, разделение FSM-логики и бизнес-сервисов, юнит-тесты в отрыве от транспорта, глобальные команды поверх диспетчера, отказ от вложенных сценариев — стандартные требования к проду. Когда эта часть собрана аккуратно, добавлять новые сценарии становится дешевле и безопаснее, а перезапуски и масштабирование на N реплик перестают ломать диалоги.

Частые вопросы

Что такое FSM в боте MAX простыми словами?+

Конечный автомат — это набор состояний и переходов между ними. В боте MAX состояние = «на каком шаге сценария находится пользователь», переход — это сообщение или нажатие кнопки, которое двигает пользователя дальше. Пример простой записи на услугу: idle → awaiting_service → awaiting_date → awaiting_time → awaiting_phone → confirmation → done. Каждое состояние знает, какие апдейты валидны: если пользователь в awaiting_phone пришлёт фото, бот не должен ломаться — он мягко напомнит, что ждёт телефон. Формально FSM описывается пятёркой (S, s0, Σ, δ, F), но на практике важнее всего понятный граф состояний и явные переходы.

Где хранить FSM-состояние пользователя в боте MAX?+

Не в памяти процесса — любой деплой ломает все активные диалоги. Рабочие варианты: Redis (самый частый выбор, быстро, переживает перезапуски, есть TTL; ключ fsm:v1:user:{user_id}, значение — JSON со state и data), PostgreSQL (медленнее, но надёжнее, можно делать сложные запросы; подходит для длинных сценариев с историей), MongoDB (если уже есть в стеке и нужна гибкая схема), гибрид (Redis для горячего состояния, PostgreSQL для финальных данных и аудита). Для большинства проектов Redis с persistence (AOF + RDB) закрывает требования. Латентность Redis — 0.5-2 мс, Postgres — 2-10 мс, разница на FSM-операциях ощутимая под нагрузкой.

Как обрабатывать таймауты в FSM бота MAX?+

Через TTL на ключе во внешнем сторадже. Long TTL (сутки или неделя) — пользователь продолжит с того же места, хорошо для сложных длинных сценариев (анкета, КИК). Medium TTL (1-4 часа) — обычная регистрация, запись на услугу. Short TTL (15-30 минут) — быстрые сценарии: заказ, билет. Можно делать ступенчато: на 15-й минуте отправить напоминание «Продолжим?» с кнопками да/нет, через ещё 15 минут без ответа сбросить. В Redis TTL встроен через SET ... EX 1800. В Postgres — фоновый воркер раз в 10 минут с DELETE FROM fsm_state WHERE updated_at < NOW() - INTERVAL '30 minutes'.

Почему при нескольких репликах бота MAX нужен внешний сторадж?+

Без внешнего стораджа состояние живёт в памяти конкретной реплики. Балансировщик отправляет апдейты round-robin: пользователь ввёл имя, попал на реплику A, она запомнила. Отправил телефон, попал на реплику B, она ничего не знает и спрашивает имя заново. Решений два: sticky sessions (балансировщик закрепляет юзера за репликой по user_id, но при падении реплики её диалоги умирают и масштабироваться сложнее) или внешний сторадж — Redis либо Postgres (любая реплика обрабатывает любой апдейт, состояние шарится). Правильный путь — сразу закладывать внешнее хранилище, даже если пока одна реплика. Миграция позже больнее.

Как версионировать FSM при изменении сценария бота?+

Сценарии меняются. Если у пользователя в Redis висит старый state awaiting_color, а вы переименовали его в awaiting_color_v2, бот может упасть. Два решения. Сбросить все активные FSM при деплое мажорной версии — просто, но раздражает пользователей в середине диалога. Пометить версию в префиксе ключа (fsm:v2:user:{user_id} вместо fsm:v1:user:{user_id}) — старые состояния просто не подхватываются и таймаутятся по TTL за 30 минут. Никаких миграций БД не нужно. На практике сочетают: для критичных правок версионирование, для мелких — миграция данных в коде через data.get(key, default).

Как тестировать FSM бота MAX?+

FSM — отличный кандидат на юнит-тесты. Логика переходов чистая, без сетевых вызовов и MAX API. Тестируем функцию transition(state, event, data) -> (new_state, new_data) или асинхронные хендлеры через MemoryStorage. Прогоняем happy path (имя → телефон → email → подтверждение), невалидные вводы («не телефон» остаётся в awaiting_phone), глобальные команды (/cancel чистит стейт). Интеграционные тесты — с реальным Redis в docker-compose, проверяют, что сторадж не теряет данные и корректно работает TTL. Юнит-тесты гоняются на каждом PR, интеграционные — на ночном CI. Хорошо разделённый код позволяет покрыть FSM на 100% без моков MAX API.

Какие антипаттерны при работе с FSM в боте MAX?+

Шесть классических. Глобальное состояние на всех пользователей вместо изолированного по user_id. Shared mutable state между корутинами без блокировок — гонки при двух одновременных апдейтах от одного юзера, лечится per-user lock в Redis через SET NX EX 5. Хранение в chat_data фреймворка без персистентности — рестарт ломает всё. Смешивание FSM и бизнес-логики в одном классе — неподдерживаемая каша. Отсутствие валидации на каждом шаге — пользователь введёт «мама» вместо номера, заявка уйдёт битой. Игнорирование TTL — Redis забивается мёртвыми сессиями, RAM течёт. Также не стоит игнорировать команды (/start, /cancel) во время сценария — глобальные команды должны работать поверх FSM как middleware.