Event-Triggered Dual Heuristic Programming (ET-DHP)¶
ET-DHP — адаптивный оптимальный регулятор из семейства Dual Heuristic Programming, расширенный схемой событийного триггера. Между срабатываниями триггера привод удерживает последнее управляющее воздействие, а обновления actor/critic не выполняются, что отделяет вычислительную частоту регулятора от частоты дискретизации датчиков/симуляции — экономия вычислений на задачах стабилизации может достигать порядка при сохранении качества регулирования замкнутого контура. См. также нелинейную модель F-16: NonlinearLongitudinalF16.
Ключевые идеи¶
- Супервизор на основе событий: правило липшицевского типа сравнивает измеренное состояние с состоянием, зафиксированным при последнем срабатывании; обновления выполняются только при превышении растущего, но ограниченного порога
- Нейросетевая модель объекта: предобученный двухслойный MLP предсказывает \(x_{k+1} = f(x_k, u_k)\); автограф через модель даёт якобианы \(F = \partial f/\partial x\) и \(G = \partial f/\partial u\), используемые в целевых функциях актора и критика
- Ограниченный актор: детерминированная политика \(u = u_b \cdot \tanh(D(x))\) соблюдает покомпонентные ограничения на привод даже во время начальной случайной фазы; \(u(0)=0\) по построению (без bias-слоёв), что обеспечивает точную неподвижную точку регулятора
- Критик сопряжённых переменных: критик напрямую регрессирует \(\lambda(x) = \partial J/\partial x\) (форма DHP), позволяя использовать чистое матрично-векторное обновление актора без скалярной головы \(J\)
- Стоимость ограниченного управления Abu-Khalaf–Lewis: интегральный член \(Y(u)\), добавленный к стоимости шага, делает \(u = u_b \cdot \tanh(D)\) точным оптимумом исходной квадратичной задачи регулирования
На схеме показан один шаг управления: измеренное состояние \(x_k\) сравнивается с состоянием при последнем срабатывании \(x_{\mathrm{et}}\); при срабатывании триггера модель объекта выдаёт \(F\) и \(G\) через autograd, критик вычисляет \(\lambda(x_{k+1})\), собирается замкнутая форма \(u^{*}\), и actor/critic делают шаги SGD. Между срабатываниями удерживается последнее значение \(u_{k-1}\), а градиенты не вычисляются.
Отличия от близких методов¶
| Аспект | HDP | DHP | ET-DHP |
|---|---|---|---|
| Выход критика | \(J(x)\) | \(\lambda(x) = \partial J/\partial x\) | \(\lambda(x)\) |
| Модель объекта | Известная / отсутствует | Аналитическая или NN | Предобученная NN |
| Дискретизация | Временная | Временная | Событийная |
| Ограничения актора | Часто отсутствуют | Часто отсутствуют | \(u_b \cdot \tanh(D)\) |
| Функция стоимости | Квадратичная | Квадратичная | Квадратичная + интеграл ограниченного управления |
Состав ET-DHP¶
| Компонент | Роль | Реализация |
|---|---|---|
| PlantModelNN | Одношаговый предиктор \(x_{k+1} = f(x_k, u_k)\); источник якобианов \(F\), \(G\) | tensoraerospace.agent.et_dhp.PlantModelNN |
| ETDHPActor | Ограниченная детерминированная политика \(u_b \cdot \tanh(D(x))\) | tensoraerospace.agent.et_dhp.ETDHPActor |
| ETDHPCritic | Сеть сопряжённых переменных \(\lambda(x) = \partial J/\partial x\) | tensoraerospace.agent.et_dhp.ETDHPCritic |
| EventTrigger | Липшицевое правило срабатывания обновлений | tensoraerospace.agent.et_dhp.EventTrigger |
| ETDHPAgent | Оркестрация всех компонент, интерфейс predict/learn | tensoraerospace.agent.et_dhp.ETDHPAgent |
Алгоритм¶
На каждом шаге \(k\) при измерении \(x_k\):
- Проверка события. Сравнить \(\|x_k - x_{\mathrm{et}}\|\) с липшицевым порогом
где \(x_{\mathrm{et}}\) и \(k_{\mathrm{trig}}\) — состояние и шаг, зафиксированные при последнем срабатывании триггера, а \(\rho \in (0, 0.5)\). Если порог превышен — срабатывание; иначе удерживать последнее управление и пропустить обучение.
-
Якобианы модели. Прямой проход \((x, u)\) через предобученную модель и построчный автограф для извлечения \(F = \partial f/\partial x\), \(G = \partial f/\partial u\).
-
Оптимальное управление в замкнутой форме (форма Modares–Lewis с ограничением):
- Цель для сопряжённых переменных. При функции стоимости \(r = x^{\top} Q x + Y(u)\) с интегральной стоимостью ограниченного управления
цель для \(\lambda\) равна \(\lambda_{\mathrm{target}} = \gamma F^{\top} \lambda(x_{k+1}) + \partial r/\partial x\).
- Градиентные шаги. SGD по актору относительно \(\mathrm{MSE}(u, u^{*})\) и по критику относительно \(\mathrm{MSE}(\lambda(x), \lambda_{\mathrm{target}})\).
Быстрый старт¶
import numpy as np
from tensoraerospace.agent.et_dhp import ETDHPAgent, ETDHPConfig
# Преобразование в регуляционное состояние: из исходного наблюдения
# в x_tilde, которое должно стремиться к нулю в рабочей точке.
def state_transform(obs, reference_signal, time_step):
return np.degrees(np.asarray(obs).reshape(-1)) # пример: градусы
cfg = ETDHPConfig(
actor_hidden=(24, 24),
critic_hidden=(24, 24),
model_hidden=(24, 24),
actor_lr=5e-3,
critic_lr=5e-3,
model_lr=5e-3,
model_epochs=300,
Q=[10.0, 0.2, 0.0, 0.0],
R=[0.5],
gamma=0.95,
num_epochs_per_trigger=5,
u_bound=5.0,
rho=0.15,
trigger_floor=0.05,
weight_init_scale=0.3,
seed=0,
)
agent = ETDHPAgent(
n_state=4,
n_control=1,
state_transform=state_transform,
config=cfg,
)
# 1. Предобучение модели объекта на оффлайн-данных с PE-сигналом.
agent.fit_plant_model(states_arr, actions_arr, next_states_arr,
batch_size=128, verbose=True)
# 2. Онлайн замкнутый контур с событийным триггером.
obs, _ = env.reset()
agent.reset()
for k in range(number_time_steps - 2):
agent.predict(obs, reference_signal, k)
u_cmd = agent.last_action()
obs_next, _, done, _, _ = env.step(u_cmd)
metrics = agent.learn(obs_next, reference_signal, k, dt=dt)
obs = obs_next
if done:
break
Tip
Неподвижная точка актора — \(u(0) = 0\). Для задач слежения проектируйте state_transform так, чтобы идеальное слежение соответствовало нулю регуляционного состояния (например, вычитайте задающий сигнал из измеренного состояния).
Гиперпараметры¶
Общие¶
| Параметр | По умолчанию | Описание |
|---|---|---|
gamma |
1.0 | Коэффициент дисконтирования |
num_epochs_per_trigger |
10 | Внутренние шаги SGD на одно срабатывание |
weight_init_scale |
0.5 | Граница равномерной инициализации весов всех сетей |
seed |
None | Зерно ГСЧ для воспроизводимости |
device |
"cpu" |
Устройство PyTorch |
Актор¶
| Параметр | По умолчанию | Описание |
|---|---|---|
actor_hidden |
(10, 10) | Размеры скрытых слоёв |
actor_lr |
1e-3 | Скорость обучения SGD |
u_bound |
1.0 | Покомпонентное ограничение привода |
Критик¶
| Параметр | По умолчанию | Описание |
|---|---|---|
critic_hidden |
(10, 10) | Размеры скрытых слоёв |
critic_lr |
1e-3 | Скорость обучения SGD |
Модель объекта¶
| Параметр | По умолчанию | Описание |
|---|---|---|
model_hidden |
(10, 10) | Размеры скрытых слоёв |
model_lr |
1e-3 | Скорость обучения Adam для оффлайн-обучения |
model_epochs |
200 | Эпохи оффлайн-предобучения |
online_model_fit |
False | Продолжать дообучение модели после оффлайн-фазы |
Функция стоимости¶
| Параметр | По умолчанию | Описание |
|---|---|---|
Q |
(1.0,) | Диагональные веса стоимости состояния \(x^{\top} Q x\); длина = n_state |
R |
(1.0,) | Диагональные веса стоимости управления; длина = n_control |
Событийный триггер¶
| Параметр | По умолчанию | Описание |
|---|---|---|
rho |
0.1 | Липшицевая константа \(\in (0, 0.5)\); меньше ⇒ больше срабатываний, точнее слежение |
trigger_floor |
1e-3 | Минимальный порог (в единицах состояния) для подавления шумовых срабатываний |
Исследование¶
| Параметр | По умолчанию | Описание |
|---|---|---|
exploration_fn |
None | Опциональный callable (time_sec) -> array, добавляющий PE в цель актора |
Поддерживаемые окружения¶
NonlinearLongitudinalF16-v0LinearLongitudinalF16-v0
Документация API¶
ETDHPAgent(n_state, n_control, state_transform=None, config=None, log_dir=None, wandb_project=None, wandb_entity=None, wandb_run_name=None, wandb_tags=None, wandb_config=None)
¶
Event-triggered DHP agent with bounded actor and costate critic.
The agent operates on a regulation state x̃ — typically the
tracking error y − y_ref possibly augmented with unmeasured
channels regulated to zero (elevator rate, etc.). The caller is
responsible for turning the raw environment observation into that
regulation state via :meth:obs_to_state, overridable by passing
a custom state_transform on construction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_state
|
int
|
Length of the regulation state vector |
required |
n_control
|
int
|
Number of control channels. |
required |
state_transform
|
Callable[[ndarray, ndarray | None, int], ndarray] | None
|
Callable |
None
|
config
|
ETDHPConfig | None
|
Optional :class: |
None
|
obs_to_state(obs, reference_signal, time_step)
¶
Turn a raw env observation into the regulation state x̃.
When no custom transform is supplied this is just the identity,
which makes sense for pure stabilisation tasks. For tracking
tasks the user can pass a state_transform that returns the
tracking error (possibly padded with zero-regulated channels).
reset()
¶
Re-arm the event trigger for a new episode.
Does not touch learned weights so training accumulates across episodes, consistent with the rest of the tensoraerospace agents.
fit_plant_model(states, actions, next_states, *, batch_size=256, epochs=None, verbose=False)
¶
Train :attr:plant_model on offline transitions (x, u, x').
The agent expects the state tuples in regulation form (i.e.
already transformed by :meth:obs_to_state). Typically a PE
roll-out is collected on the real env, converted via
obs_to_state and then handed here.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
states
|
ndarray
|
Array of shape |
required |
actions
|
ndarray
|
Array of shape |
required |
next_states
|
ndarray
|
Array of shape |
required |
batch_size
|
int
|
Mini-batch size for Adam. |
256
|
epochs
|
int | None
|
Overrides |
None
|
verbose
|
bool
|
Print per-epoch MSE when True. |
False
|
Returns:
| Type | Description |
|---|---|
list[float]
|
Per-epoch MSE loss trace. |
predict(obs, reference_signal=None, time_step=0, *, deterministic=True)
¶
Compute the control action for the current measurement.
deterministic is accepted for API consistency with other
tensoraerospace agents but is ignored — exploration is handled
inside :meth:learn via cfg.exploration_fn, so the control
that reaches the plant is always the actor's current best.
learn(next_obs, reference_signal=None, time_step=0, *, dt=1.0)
¶
Run the event-triggered update step.
Must be called after :meth:predict and the environment
step. When the event-trigger threshold is not crossed this
method is essentially free — only the Lipschitz test runs, no
autograd and no optimiser touch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
next_obs
|
ndarray
|
Environment measurement |
required |
reference_signal
|
ndarray | None
|
Optional reference trajectory, forwarded to the state transform. |
None
|
time_step
|
int
|
Discrete step index |
0
|
dt
|
float
|
Simulation step (s). Used only to advance the internal
wall-clock that |
1.0
|
Returns:
| Type | Description |
|---|---|
dict[str, float]
|
Dictionary of scalar metrics: |
dict[str, float]
|
|
dict[str, float]
|
|
last_action()
¶
Return the control that the env should actually apply.
Between triggers this is the same action that was last produced
by :meth:predict; on triggered steps it reflects the freshly
updated actor.
train_episode(env, *, max_steps=None, reference_signal=None)
¶
Run one episode of ET-DHP training against a Gym env.
This is a thin wrapper intended for scripts/notebooks; the
more flexible path is to call :meth:predict / :meth:learn
directly and forward last_action to the environment.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Any
|
Gymnasium env exposing |
required |
max_steps
|
int | None
|
Optional cap on episode length. |
None
|
reference_signal
|
ndarray | None
|
Override the env's reference (defaults
to |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, float]
|
Summary dict: |
train(env, *, num_episodes=1, max_steps=None, reference_signal=None)
¶
Train ET-DHP for num_episodes episodes (unified interface).
Thin wrapper around :meth:train_episode that loops the requested
number of episodes, flushes the metrics writer, and asserts the
unified-metrics contract once the session has ended.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
env
|
Any
|
Gymnasium env. |
required |
num_episodes
|
int
|
Episode budget. |
1
|
max_steps
|
int | None
|
Optional per-episode step cap. |
None
|
reference_signal
|
ndarray | None
|
Optional reference signal forwarded to
:meth: |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, float]
|
Aggregated session-level summary. |
get_param_env()
¶
Build a JSON-serialisable config for :meth:save.
The state_transform callable is never serialised — on load the
caller must supply it again via :meth:from_pretrained /
:meth:_load_from_dir. This matches how Gymnasium treats other
non-picklable env-side components.
save(path=None, *, save_gradients=False)
¶
Write the agent to a directory.
Files produced
config.json— constructor kwargs + serialised :class:ETDHPConfig.actor.pth/critic.pth/plant_model.pth— PyTorch state dicts.event_trigger.json— supervisor counters and the state captured at the last trigger.actor_optim.pth/critic_optim.pth/model_optim.pth— optimiser state dicts (only whensave_gradients=True).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
Union[str, Path, None]
|
Base directory ( |
None
|
save_gradients
|
bool
|
Persist optimiser states for resume. |
False
|
Returns:
| Type | Description |
|---|---|
str
|
Absolute path to the created run directory. |
from_pretrained(repo_name, access_token=None, version=None, state_transform=None, load_gradients=False)
classmethod
¶
Load an agent from a local directory or the Hugging Face Hub.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
repo_name
|
str
|
Local folder path, or |
required |
access_token
|
Optional[str]
|
Hub access token for private repos. |
None
|
version
|
Optional[str]
|
Hub revision / branch / tag. |
None
|
state_transform
|
Callable[[ndarray, ndarray | None, int], ndarray] | None
|
Re-attach the observation-to-regulation transform used during training (optional but required for tracking tasks). |
None
|
load_gradients
|
bool
|
Also restore optimiser state dicts. |
False
|
Returns:
| Name | Type | Description |
|---|---|---|
ETDHPAgent |
'ETDHPAgent'
|
Reconstructed agent. |
publish_to_hub(repo_name, folder_path, access_token=None)
¶
Upload a :meth:save directory to the Hugging Face Hub.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
repo_name
|
str
|
Target repository id, e.g. |
required |
folder_path
|
Union[str, Path]
|
Local folder produced by :meth: |
required |
access_token
|
Optional[str]
|
Hub access token. |
None
|
ETDHPConfig(actor_hidden=(10, 10), critic_hidden=(10, 10), model_hidden=(10, 10), actor_lr=0.001, critic_lr=0.001, model_lr=0.001, model_epochs=200, online_model_fit=False, Q=(1.0,), R=(1.0,), gamma=1.0, num_epochs_per_trigger=10, u_bound=1.0, rho=0.1, trigger_floor=0.001, exploration_fn=None, weight_init_scale=0.5, device='cpu', seed=None, history=dict())
dataclass
¶
Hyper-parameters for :class:ETDHPAgent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
actor_hidden
|
Sequence[int]
|
Tanh hidden-layer sizes for the actor MLP. |
(10, 10)
|
critic_hidden
|
Sequence[int]
|
Tanh hidden-layer sizes for the critic MLP. |
(10, 10)
|
model_hidden
|
Sequence[int]
|
Tanh hidden-layer sizes for the plant-model MLP. |
(10, 10)
|
actor_lr
|
float
|
SGD learning rate for the actor. |
0.001
|
critic_lr
|
float
|
SGD learning rate for the critic. |
0.001
|
model_lr
|
float
|
Adam learning rate for the plant-model pre-training. |
0.001
|
model_epochs
|
int
|
Number of epochs for the offline plant-model fit
( |
200
|
online_model_fit
|
bool
|
Whether to keep adapting the plant model on
every triggered step. Disabled by default — the reference
paper freezes |
False
|
Q
|
Sequence[float]
|
Diagonal weights of the quadratic state cost |
(1.0,)
|
R
|
Sequence[float]
|
Diagonal weights of the quadratic control cost. Length must
equal |
(1.0,)
|
gamma
|
float
|
Discount factor. Default |
1.0
|
num_epochs_per_trigger
|
int
|
Inner-loop actor/critic iterations run each time the event trigger fires. The reference uses 10 on the nonlinear F-16-like plant. |
10
|
u_bound
|
float
|
Per-channel actuator bound fed to the bounding layer of the actor. Interpreted in the same units as the action the env expects. |
1.0
|
rho
|
float
|
Lipschitz constant for the event trigger (see
:class: |
0.1
|
trigger_floor
|
float
|
Minimum trigger threshold (state units). Small
positive values suppress noise-induced retriggers when
|
0.001
|
exploration_fn
|
Callable[[float], ndarray] | None
|
Optional callable |
None
|
weight_init_scale
|
float
|
Uniform-init bound for each NN weight. |
0.5
|
device
|
str
|
Torch device. |
'cpu'
|
seed
|
int | None
|
Optional seed for numpy and torch RNGs. |
None
|
ETDHPActor(n_state, n_control, hidden_sizes=(10, 10), u_bound=1.0, init_scale=0.5)
¶
Bases: Module
Bounded deterministic policy u = u_b · tanh(D_nn(x)).
The D_nn pre-bound output is returned alongside the bounded
control because the ET-DHP reward contains a "bounding" term that
depends on the raw logits — see :func:bounded_integral_cost in
:mod:tensoraerospace.agent.et_dhp.model.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_state
|
int
|
Length of the state vector (actor input). |
required |
n_control
|
int
|
Number of actuators. |
required |
hidden_sizes
|
Sequence[int]
|
Tanh hidden-layer sizes. |
(10, 10)
|
u_bound
|
float
|
Per-channel absolute bound on the output. The bounding
layer rescales |
1.0
|
init_scale
|
float
|
Uniform-init bound for each weight. |
0.5
|
forward(state)
¶
Return (u_bounded, D_nn).
D_nn is the pre-saturation output of the last linear layer.
The bounded control is u_bound · tanh(D_nn).
ETDHPCritic(n_state, hidden_sizes=(10, 10), init_scale=0.5)
¶
Bases: Module
Costate critic λ(x) = ∂J/∂x for DHP.
Directly regresses the partial derivative of the cost-to-go with
respect to the state (same dimensionality as x) so the actor
update can take a plain matrix-vector form and skip the scalar
J-head altogether — this is the original DHP formulation from
Werbos/Prokhorov.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_state
|
int
|
Length of the state vector |
required |
hidden_sizes
|
Sequence[int]
|
Tanh hidden-layer sizes. |
(10, 10)
|
init_scale
|
float
|
Uniform-init bound for each weight. |
0.5
|
forward(state)
¶
Return λ(x) of shape (n_state,) (or batched).
PlantModelNN(n_state, n_control, hidden_sizes=(10, 10), init_scale=0.5)
¶
Bases: Module
One-step predictor x_{k+1} = f(x_k, u_k) for ET-DHP.
The network realises a discrete-time approximation of the plant dynamics. It is queried twice during each learning sweep:
- Forward — predict the next state.
- Backward — extract the Jacobians
F = ∂f/∂xandG = ∂f/∂uvia autograd; these feed directly into the critic target and the optimal-control formula used by the actor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
n_state
|
int
|
Length of the state vector |
required |
n_control
|
int
|
Length of the control vector |
required |
hidden_sizes
|
Sequence[int]
|
Sizes of the tanh hidden layers. Defaults to
|
(10, 10)
|
init_scale
|
float
|
Uniform-init bound for each weight (no bias). The
reference uses |
0.5
|
forward(xu)
¶
Return the predicted next state from a concatenated [x; u].
EventTrigger(rho=0.1, trigger_first_step=True, min_floor=0.0)
¶
Stateful Lipschitz-based event-trigger for ET-DHP.
Keeps track of the reference state captured at the last trigger and
compares its distance to the incoming measured state against a
growing threshold that saturates once (2ρ)^Δk → 0.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rho
|
float
|
Lipschitz constant |
0.1
|
trigger_first_step
|
bool
|
If True (default), the very first call to
:meth: |
True
|
min_floor
|
float
|
Lower bound on the threshold. Adds a small absolute
slack so the trigger does not fire on pure measurement
noise when |
0.0
|
reset()
¶
Re-arm the trigger for a new episode.
threshold(step)
¶
Compute the current Lipschitz threshold at step.
Returns the floor when the trigger has never fired yet (so the first evaluation still runs).
should_trigger(state_measured, step)
¶
Return True when the networks must be re-evaluated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state_measured
|
ndarray
|
Current (possibly noisy) state measurement. |
required |
step
|
int
|
Current discrete time index. |
required |
Источники¶
- Sun, B., Liu, C., Dally, K., van Kampen, E.-J. (2022). Intelligent Aircraft Stabilization Control with Event-Triggered Scheme. CEAS EuroGNC 2022.
- Abu-Khalaf, M., Lewis, F. L. (2005). Nearly optimal control laws for nonlinear systems with saturating actuators using a neural network HJB approach. Automatica, 41(5), 779–791.
- Modares, H., Lewis, F. L. (2014). Optimal tracking control of nonlinear partially-unknown constrained-input systems using integral reinforcement learning. Automatica, 50(7), 1780–1792.