Model Predictive Control (MPC)¶
MPC использует модель динамики для прогнозирования поведения системы и выбора оптимальной последовательности управлений с учётом ограничений. На каждом шаге решается задача оптимизации, применяется первое управление из оптимальной последовательности, затем окно сдвигается и цикл повторяется.
Теория (кратко)¶
- Дискретная динамика: \(x_{k+1} = f(x_k, u_k)\)
- Функция стоимости на горизонте \(N\):
- Приращение управления:
- Ограничения:
- Скользящий горизонт: решаем → применяем \(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)
Использование различных моделей динамики¶
Вы можете подключить различные архитектуры нейронных сетей:
Дополнительные функции стоимости¶
Режим слежения (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-энкодер для предсказания динамики |
Каждый пример демонстрирует полный пайплайн:
- Настройка среды — Создание среды B747 со ступенчатым референсом по тангажу (θ)
- Сбор данных — Сбор переходов с использованием богатых исследовательских сигналов
- Обучение динамики — Обучение нейросети предсказывать переходы состояний
- MPC rollout — Запуск управления с замкнутым контуром на обученной динамике
- Оценка — Анализ качества переходного процесса (перерегулирование, время установления и т.д.)
Ключевые результаты из примеров¶
| Модель | Перерегулирование | Время установления | Время нарастания | Статическая ошибка |
|---|---|---|---|---|
| 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"usestensoraerospace.signalsto 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=1andcontrol_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
|
nhead
|
int
|
Number of attention heads. Defaults to |
4
|
num_encoder_layers
|
int
|
Number of encoder layers. Defaults to |
2
|
dim_feedforward
|
int
|
Feed-forward layer dimension. Defaults to |
256
|
dropout
|
float
|
Dropout probability. Defaults to |
0.1
|
seq_len
|
int
|
Sequence length. Defaults to |
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). |
