Перейти к содержанию

Model Predictive Control (MPC)

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

MPC схема

Теория (кратко)

  • Дискретная динамика: \(x_{k+1} = f(x_k, u_k)\)
  • Функция стоимости на горизонте \(N\):
\[ J = \sum_{i=0}^{N-1} (x_{k+i} - x^{\mathrm{ref}}_{k+i})^\top Q (x_{k+i} - x^{\mathrm{ref}}_{k+i}) + u_{k+i}^\top R\, u_{k+i} + \Delta u_{k+i}^\top S\, \Delta u_{k+i} + \text{terminal\_weight} \cdot (x_{k+N}-x^{\mathrm{ref}}_{k+N})^\top Q (x_{k+N}-x^{\mathrm{ref}}_{k+N}) \]
  • Приращение управления:
\[ \Delta u_{k+i} = u_{k+i} - u_{k+i-1} \]
  • Ограничения:
\[ \begin{aligned} u_{\min} \le u_{k+i} \le u_{\max}, &\quad \Delta u_{\min} \le \Delta u_{k+i} \le \Delta u_{\max}, \\ \end{aligned} \]
  • Скользящий горизонт: решаем → применяем \(u_k\) → сдвигаем окно → повторяем
  • Устойчивость: терминальный вес, достаточный \(N\), допустимость

Архитектура

Модуль MPC состоит из:

Компонент Класс Описание
Низкоуровневый солвер MPC Проекционно-градиентный оптимизатор для дифференцируемой динамики
Высокоуровневый агент MPCAgent DSAC-подобная обёртка с обучаемой динамикой, буфером и обучением
Конфигурация весов MPCWeights Диагональные веса Q, R, S и терминальный вес
Ограничения MPCConstraints Box-ограничения для u и du
Доп. штрафы MPCTrackingExtraCostConfig, MPCStepResponseExtraCostConfig Штрафы за гладкость, перерегулирование, время установления
Модели динамики OneStepMLP, NARXDynamicsModel, TransformerDynamicsModel Нейросетевые модели для обучения динамики
Нормализатор MPCStandardScaler Нормализация признаков (mean/std)

Быстрый старт

Базовый MPC с пользовательской функцией динамики

import numpy as np
import torch
from tensoraerospace.agent.mpc import MPC, MPCWeights, MPCConstraints

state_dim = 4
action_dim = 1

# Определяем динамику: x_{t+1} = f(x_t, u_t)
def dynamics(x: torch.Tensor, u: torch.Tensor) -> torch.Tensor:
    # Простая линейная динамика для примера
    A = torch.eye(state_dim)
    B = torch.zeros(state_dim, action_dim)
    B[-1, 0] = 1.0  # управление влияет на последнее состояние
    return x @ A.T + u @ B.T

# Конфигурация весов
weights = MPCWeights(
    Q_diag=np.array([1.0, 1.0, 1.0, 10.0]),  # веса слежения за состоянием
    R_diag=np.array([0.01]),                   # штраф за управление
    S_diag=np.array([0.1]),                    # штраф за гладкость управления
    terminal_weight=2.0,
)

# Конфигурация ограничений
constraints = MPCConstraints(
    u_min=np.array([-1.0]),
    u_max=np.array([1.0]),
    du_min=np.array([-0.2]),
    du_max=np.array([0.2]),
)

# Создаём MPC солвер
mpc = MPC(
    dynamics=dynamics,
    state_dim=state_dim,
    action_dim=action_dim,
    horizon=20,
    weights=weights,
    constraints=constraints,
    iters=60,
    lr=0.05,
    optimizer="adam",
    warm_start=True,
)

# Решаем
x0 = np.zeros(state_dim)
x_ref = np.zeros((21, state_dim))  # опорная траектория (horizon+1)
x_ref[:, -1] = 0.1  # цель для последней компоненты состояния

result = mpc.solve(x0=x0, x_ref=x_ref, u_prev=None)
print("Первое управление:", result.u0)
print("Форма предсказанной траектории:", result.x_seq.shape)

MPCAgent с обучаемой динамикой (рекомендуется)

MPCAgent предоставляет полный рабочий процесс: сбор данных, обучение динамики и MPC-управление.

import gymnasium as gym
import numpy as np
from tensoraerospace.agent.mpc import (
    MPCAgent,
    MPCWeights,
    MPCConstraints,
    MPCStepResponseExtraCostConfig,
)

# Создаём среду
env = gym.make("LinearLongitudinalB747-v0", ...)

# Конфигурация весов
weights = MPCWeights(
    Q_diag=np.array([1.0, 1.0, 10.0, 100.0]),
    R_diag=np.array([0.01]),
    S_diag=np.array([0.5]),
    terminal_weight=1.0,
)

# Конфигурация ограничений
constraints = MPCConstraints(
    u_min=np.array([-0.3]),
    u_max=np.array([0.3]),
    du_min=np.array([-0.05]),
    du_max=np.array([0.05]),
)

# Доп. штрафы для качества переходного процесса
step_cfg = MPCStepResponseExtraCostConfig.from_degrees(
    tracked_idx=-1,        # последнее состояние = theta
    rate_idx=-2,           # предпоследнее = q (угловая скорость тангажа)
    dt=0.01,
    overshoot_limit_deg=0.05,
    settle_band_deg=0.1,
    settle_time_target_s=1.0,
)

# Создаём агента
agent = MPCAgent(
    env,
    horizon=30,
    weights=weights,
    constraints=constraints,
    tracking_type="step_response",
    step_response_config=step_cfg,
    hidden_layers=(256, 256),
    normalize=True,
    device="cuda",  # или "cpu"
)

# Собираем обучающие данные
agent.collect_data(num_episodes=50, exploration="signals")

# Обучаем модель динамики
agent.train_dynamics(epochs=10, batch_size=1024)

# Используем в цикле управления
obs, info = env.reset()
state = ...  # извлекаем внутреннее состояние из среды
x_ref = ...  # опорная траектория (horizon+1, state_dim)

action = agent.select_action(state, x_ref=x_ref)
obs, reward, done, truncated, info = env.step(action)

# Сохранение/загрузка чекпоинтов
path = agent.save("./runs")
agent.load(path)

Использование различных моделей динамики

Вы можете подключить различные архитектуры нейронных сетей:

from tensoraerospace.agent.mpc import OneStepMLP

model = OneStepMLP(
    input_dim=state_dim + action_dim,
    output_dim=state_dim,
    hidden_layers=(256, 256, 128),
    activation="relu",  # или "tanh", "gelu"
)

agent = MPCAgent(env, model=model, ...)
from tensoraerospace.agent.mpc import NARXDynamicsModel

model = NARXDynamicsModel(
    state_dim=state_dim,
    action_dim=action_dim,
    hidden_size=256,
    num_layers=3,
    state_lags=1,
    control_lags=1,
)

agent = MPCAgent(env, model=model, ...)
from tensoraerospace.agent.mpc import TransformerDynamicsModel

model = TransformerDynamicsModel(
    input_dim=state_dim + action_dim,
    output_dim=state_dim,
    d_model=64,
    nhead=4,
    num_encoder_layers=2,
    dim_feedforward=256,
    dropout=0.1,
)

agent = MPCAgent(env, model=model, ...)

Дополнительные функции стоимости

Режим слежения (tracking)

Добавляет штрафы за гладкость управления:

  • w_du: вес для \(\sum (\Delta u)^2\)
  • w_jerk: вес для \(\sum (\Delta^2 u)^2\)
from tensoraerospace.agent.mpc import MPCTrackingExtraCostConfig

cfg = MPCTrackingExtraCostConfig(w_du=50.0, w_jerk=10.0)
agent = MPCAgent(env, tracking_type="tracking", tracking_config=cfg, ...)

Режим переходного процесса (step_response)

Добавляет штрафы за перерегулирование, время установления, колебания:

from tensoraerospace.agent.mpc import MPCStepResponseExtraCostConfig

cfg = MPCStepResponseExtraCostConfig.from_degrees(
    tracked_idx=-1,               # индекс отслеживаемого состояния (напр., theta)
    rate_idx=-2,                  # индекс скорости (напр., q)
    dt=0.01,                      # шаг по времени
    overshoot_limit_deg=0.05,     # макс. перерегулирование в градусах
    settle_band_deg=0.10,         # ширина полосы установления
    settle_time_target_s=1.0,     # целевое время установления
    w_overshoot=8000.0,           # вес штрафа за перерегулирование
    w_settle=8000.0,              # вес штрафа за установление
    w_osc=500.0,                  # вес штрафа за колебания
    w_jerk=50.0,                  # вес штрафа за рывки
)

agent = MPCAgent(env, tracking_type="step_response", step_response_config=cfg, ...)

Можно переключать режимы во время работы:

agent.set_tracking_type("tracking", tracking_config=tracking_cfg)
# или
agent.set_tracking_type("step_response", step_response_config=step_cfg)

Сбор данных

MPCAgent.collect_data() поддерживает две стратегии исследования:

Стратегия Описание
"random" Случайные действия через env.action_space.sample()
"signals" Богатая библиотека сигналов: ступеньки, рампы, синусоиды, chirp, дублеты и др.
agent.collect_data(
    num_episodes=50,
    max_steps=1000,
    exploration="signals",
    signal_kinds=["random_steps", "sinusoid", "chirp", "doublet"],
    action_amplitude_frac=0.8,
)

Доступные типы сигналов: random_steps, unit_step, multi_step, ramp, sinusoid, multisine, chirp, square_wave, triangular_wave, sawtooth, doublet, pulse, gaussian_pulse, exponential, damped_sinusoid.

Гиперпараметры

MPC Solver (MPC)

Параметр Описание По умолчанию
horizon Горизонт предсказания 20
iters Итерации оптимизации за один вызов solve 60
lr Скорость обучения 0.05
optimizer "adam" или "sgd" "adam"
warm_start Повторное использование предыдущего решения True
track_best Отслеживание лучшего решения при оптимизации True
compile_dynamics Использовать torch.compile (PyTorch 2.x) False

MPCAgent

Параметр Описание По умолчанию
hidden_layers Размеры скрытых слоёв MLP (256, 256)
normalize Нормализация входов/выходов True
dynamics_lr Скорость обучения модели динамики 1e-3
grad_clip_norm Обрезка градиентов 1.0
memory_capacity Размер replay buffer 200_000
model_predict_delta Предсказывать \(\Delta x\) вместо \(x'\) True

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

  • Используйте exploration="signals" для лучшего покрытия пространства состояний-действий
  • Начните с horizon=20-30 и увеличивайте при необходимости
  • Включите normalize=True для нейросетевой динамики
  • Используйте tracking_type="step_response" для аэрокосмических задач управления
  • Для управления в реальном времени рассмотрите compile_dynamics=True на GPU

Примеры

Полные end-to-end примеры, демонстрирующие MPC с различными моделями динамики на задаче продольного управления B747:

Пример Модель динамики Описание
MPC + MLP OneStepMLP Стандартное обучение динамики на MLP с отслеживанием переходного процесса
MPC + NARX NARXDynamicsModel Нелинейная авторегрессия с экзогенными входами
MPC + Transformer TransformerDynamicsModel Transformer-энкодер для предсказания динамики

Каждый пример демонстрирует полный пайплайн:

  1. Настройка среды — Создание среды B747 со ступенчатым референсом по тангажу (θ)
  2. Сбор данных — Сбор переходов с использованием богатых исследовательских сигналов
  3. Обучение динамики — Обучение нейросети предсказывать переходы состояний
  4. MPC rollout — Запуск управления с замкнутым контуром на обученной динамике
  5. Оценка — Анализ качества переходного процесса (перерегулирование, время установления и т.д.)

Ключевые результаты из примеров

Модель Перерегулирование Время установления Время нарастания Статическая ошибка
MLP ~0.30% ~1.7с ~1.1с ~0.001
NARX ~-1.9% ~3.0с ~1.0с ~0.026
Transformer ~-0.10% ~1.5с ~0.8с ~0.009

Запуск примеров

Примеры — это Jupyter notebooks, расположенные в example/mpc_controllers/. Запустите их, чтобы увидеть полные логи обучения, графики и отчёты бенчмарков.

Документация API

MPC(*, dynamics, state_dim, action_dim, horizon=20, weights, constraints=None, extra_cost_fn=None, iters=60, lr=0.05, optimizer='adam', device=None, dtype=torch.float32, warm_start=True, track_best=True, best_check_every=1, compile_dynamics=False, compile_mode='reduce-overhead', seed=None)

Projected-gradient MPC over a differentiable dynamics model.

This solver optimizes a control sequence U using torch/autograd and applies hard constraints via projection (clamp + sequential rate limiting).

reset()

Reset warm-start state.

solve(*, x0, x_ref=None, u_prev=None)

Solve MPC problem for the current state.

Parameters:

Name Type Description Default
x0 TensorLike

Current state, shape (state_dim,) or (state_dim, 1) or (1, state_dim).

required
x_ref TensorLike | None

Optional reference trajectory for states. Expected shape is (horizon+1, state_dim) or (horizon, state_dim). If provided as (horizon, state_dim), it is interpreted as targets for x_{t+1} and a terminal target is appended by repeating the last row.

None
u_prev TensorLike | None

Optional previous control input, shape (action_dim,).

None

Returns:

Name Type Description
MPCSolveResult MPCSolveResult

contains the first control input and the predicted trajectory.

MPCAgent(env, *, state_dim=None, action_dim=None, horizon=20, weights=None, constraints=None, tracking_type='tracking', tracking_config=None, step_response_config=None, extra_cost_fn=None, iters=60, mpc_lr=0.05, mpc_optimizer='adam', warm_start=True, mpc_track_best=True, mpc_best_check_every=1, mpc_compile_dynamics=False, mpc_compile_mode='reduce-overhead', model=None, model_predict_delta=True, hidden_layers=(256, 256), activation='relu', normalize=True, dynamics_lr=0.001, weight_decay=0.0, grad_clip_norm=1.0, memory_capacity=200000, obs_to_state=None, action_to_env=None, action_from_env=None, device='cpu', dtype=torch.float32, seed=0)

Bases: BaseRLModel

DSAC-like wrapper around MPC with learned dynamics.

Goals: - Work with different Gymnasium-like environments (infer dims and bounds) - Accept a neural dynamics model at init (or build a default MLP) - Provide DSAC-ish ergonomics: buffer, collect_data(), train_dynamics(), select_action()

set_tracking_type(tracking_type, *, tracking_config=None, step_response_config=None)

Switch extra-cost mode (tracking vs step_response).

to_device(device)

Move model, normalizers, and MPC to a new device (DSAC-style).

get_env()

Return current env.

get_param_env()

Return env/policy metadata for HuggingFace-style checkpoints.

reset()

Reset MPC warm-start and previous action memory.

select_action(state, *, x_ref=None)

Compute action (env units) using MPC over learned dynamics.

collect_data(*, num_episodes=10, max_steps=None, exploration='random', signal_kinds=None, dt=None, action_amplitude_frac=0.8)

Collect (x, u, x_next) transitions into self.memory.

Notes
  • For exploration="random" uses env.action_space.sample().
  • For exploration="signals" uses tensoraerospace.signals to generate time-series actions (works for continuous Box actions).

fit_normalizers(*, num_samples=50000)

Fit x/u/y normalizers from a random subset of the replay buffer.

train_dynamics(*, epochs=5, batch_size=1024, steps_per_epoch=None, loss='mse', force_refit_normalizers=False)

Train the dynamics model on transitions stored in the replay buffer.

The model is trained on samples from self.memory.

Parameters:

Name Type Description Default
force_refit_normalizers bool

If True, re-fit the normalizers even if they have already been fitted. Default is False — normalizers are fit only on the first call to avoid changing the loss landscape mid-training.

False

save(path=None, save_gradients=True)

Save MPC agent in HuggingFace-style layout (config + weights).

from_pretrained(repo_name, access_token=None, version=None, load_gradients=False) classmethod

Load checkpoint from local dir or HuggingFace Hub.

push_to_hub(repo_name, access_token=None, save_path=None, include_gradients=False)

Save checkpoint and upload it to HuggingFace Hub.

MPCWeights(Q_diag, R_diag, S_diag=None, terminal_weight=1.0) dataclass

Quadratic weights for the standard MPC objective.

The objective is

sum_{t=0..N-1} ||x_{t+1} - x_ref_{t+1}||Q^2 + ||u_t||_R^2 + ||u_t - u||_S^2 + terminal_weight * ||x_N - x_ref_N||_Q^2

Q, R, S are interpreted as diagonal weights (vectors). This keeps the API simple and fast enough for small horizons without extra dependencies.

MPCConstraints(u_min=None, u_max=None, du_min=None, du_max=None) dataclass

Box constraints for control and rate limits.

All bounds are interpreted element-wise (per control dimension).

MPCSolveResult(u0, u_seq, x_seq, final_cost, iters) dataclass

Result bundle for MPC solve.

MPCTrackingExtraCostConfig(w_du=0.0, w_jerk=0.0) dataclass

Extra cost for generic reference tracking.

This is applied in addition to the quadratic MPC objective. Values are dimensionless weights.

MPCStepResponseExtraCostConfig(tracked_idx=-1, rate_idx=None, dt=0.1, ref_change_threshold=float(np.deg2rad(0.1)), min_step_amp=float(np.deg2rad(0.5)), overshoot_limit=float(np.deg2rad(0.05)), settle_band=float(np.deg2rad(0.1)), settle_band_min=float(np.deg2rad(0.05)), settle_band_ratio=0.01, settle_time_target_s=1.0, rate_settle=float(np.deg2rad(0.25)), w_overshoot=8000.0, w_time=800.0, w_settle=8000.0, w_sse_steady=40000.0, w_osc=500.0, w_jerk=50.0, w_du_steady=80.0, w_jerk_steady=800.0) dataclass

Extra cost tuned for step response (overshoot/settling/osc/jerk).

All thresholds must be in the SAME UNITS as the tracked state component inside x_seq/x_ref (for B747 internal model this is radians).

MPCStandardScaler(mean, std) dataclass

Simple per-feature standardization helper (mean/std).

The scaler lives on a torch device and is used both for training and for MPC rollouts through a learned dynamics model.

OneStepMLP(*, input_dim, output_dim, hidden_layers=(256, 256), activation='relu')

Bases: Module

A small MLP for one-step dynamics learning.

Expected IO

in: concatenated [x, u] of shape (B, state_dim + action_dim) out: either delta-x or x_next of shape (B, state_dim)

forward(xu)

Forward pass.

NARXDynamicsModel(*, state_dim, action_dim, hidden_size=256, num_layers=2, state_lags=1, control_lags=1)

Bases: Module

MPCAgent-compatible NARX dynamics model.

MPCAgent expects learned dynamics modules with signature:

y = model(xu), where xu = concat([x, u])

This wrapper builds an internal :class:~tensoraerospace.agent.mpc.narx.NARX and provides the required forward(xu) interface.

Notes
  • For the current MPC pipeline, this is typically used with state_lags=1 and control_lags=1 (one-step model).
  • If you want true NARX with lags > 1, you must provide an augmented state/action history vector as input to MPC (not handled implicitly).

forward(xu)

Forward pass.

Parameters:

Name Type Description Default
xu Tensor

Concatenated input of shape (B, state_dimstate_lags + action_dimcontrol_lags).

required

Returns:

Type Description
Tensor

Predicted next-state (or delta-state) of shape (B, state_dim).

NARX(input_size, hidden_size, output_size, num_layers, state_lags, control_lags)

Bases: Module

NARX neural network for learning system dynamics.

The model uses lagged (historical) state and control inputs to predict the next state.

Parameters:

Name Type Description Default
input_size int

Input size (concatenated lagged states and controls).

required
hidden_size int

Hidden layer size.

required
output_size int

Output size (predicted state dimension).

required
num_layers int

Number of hidden layers.

required
state_lags int

Number of state lags used as input.

required
control_lags int

Number of control lags used as input.

required

Initialize the NARX network.

Parameters:

Name Type Description Default
input_size int

Input size.

required
hidden_size int

Hidden layer size.

required
output_size int

Output size.

required
num_layers int

Number of hidden layers.

required
state_lags int

Number of state lags.

required
control_lags int

Number of control lags.

required

forward(state, control)

Run a forward pass.

Parameters:

Name Type Description Default
state Tensor

Lagged state tensor.

required
control Tensor

Lagged control tensor.

required

Returns:

Type Description
Tensor

torch.Tensor: Predicted next state.

TransformerDynamicsModel(input_dim, output_dim, d_model=64, nhead=4, num_encoder_layers=2, dim_feedforward=256, dropout=0.1, seq_len=1)

Bases: Module

Transformer-based model for learning system dynamics.

Predicts the next system state from a sequence of state+action inputs using a Transformer encoder.

Parameters:

Name Type Description Default
input_dim int

Input dimension (state + control).

required
output_dim int

Output dimension (next state).

required
d_model int

Transformer model dimension. Defaults to 64.

64
nhead int

Number of attention heads. Defaults to 4.

4
num_encoder_layers int

Number of encoder layers. Defaults to 2.

2
dim_feedforward int

Feed-forward layer dimension. Defaults to 256.

256
dropout float

Dropout probability. Defaults to 0.1.

0.1
seq_len int

Sequence length. Defaults to 1.

1

Initialize the transformer dynamics model.

Parameters:

Name Type Description Default
input_dim int

Input dimension.

required
output_dim int

Output dimension.

required
d_model int

Model dimension.

64
nhead int

Number of attention heads.

4
num_encoder_layers int

Number of encoder layers.

2
dim_feedforward int

Feed-forward dimension.

256
dropout float

Dropout probability.

0.1
seq_len int

Sequence length.

1

forward(x)

Forward pass through transformer model.

Parameters:

Name Type Description Default
x Tensor

Input tensor of shape (batch_size, input_dim).

required

Returns:

Type Description
Tensor

torch.Tensor: Predicted next state of shape (batch_size, output_dim).