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

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)\) точным оптимумом исходной квадратичной задачи регулирования

Архитектура ET-DHP

На схеме показан один шаг управления: измеренное состояние \(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\):

  1. Проверка события. Сравнить \(\|x_k - x_{\mathrm{et}}\|\) с липшицевым порогом
\[ \rho \, \|x_{\mathrm{et}}\| \, \frac{1 - (2\rho)^{k - k_{\mathrm{trig}}}}{1 - 2\rho} \]

где \(x_{\mathrm{et}}\) и \(k_{\mathrm{trig}}\) — состояние и шаг, зафиксированные при последнем срабатывании триггера, а \(\rho \in (0, 0.5)\). Если порог превышен — срабатывание; иначе удерживать последнее управление и пропустить обучение.

  1. Якобианы модели. Прямой проход \((x, u)\) через предобученную модель и построчный автограф для извлечения \(F = \partial f/\partial x\), \(G = \partial f/\partial u\).

  2. Оптимальное управление в замкнутой форме (форма Modares–Lewis с ограничением):

\[ u^{*} = u_b \cdot \tanh\!\left(-\frac{\gamma}{2 u_b} R^{-1} G^{\top} \lambda(x_{k+1})\right) \]
  1. Цель для сопряжённых переменных. При функции стоимости \(r = x^{\top} Q x + Y(u)\) с интегральной стоимостью ограниченного управления
\[ Y(u) = 2 u_b^2 \, \mathrm{diag}(R) \cdot \bigl[\tanh(D)\cdot D + \tfrac{1}{2}\log(1 - \tanh^2 D)\bigr], \]

цель для \(\lambda\) равна \(\lambda_{\mathrm{target}} = \gamma F^{\top} \lambda(x_{k+1}) + \partial r/\partial x\).

  1. Градиентные шаги. 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-v0
  • LinearLongitudinalF16-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 — 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 (obs, ref, time_step) -> x̃ that builds the regulation state from the raw Gym observation and reference. If None (default), the state is assumed to already be in regulation form.

None
config ETDHPConfig | None

Optional :class:ETDHPConfig instance.

None

obs_to_state(obs, reference_signal, time_step)

Turn a raw env observation into the regulation state .

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 (N, n_state).

required
actions ndarray

Array of shape (N, n_control).

required
next_states ndarray

Array of shape (N, n_state).

required
batch_size int

Mini-batch size for Adam.

256
epochs int | None

Overrides cfg.model_epochs when set.

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 y_{k+1}.

required
reference_signal ndarray | None

Optional reference trajectory, forwarded to the state transform.

None
time_step int

Discrete step index k of the measurement (same index that was passed to :meth:predict).

0
dt float

Simulation step (s). Used only to advance the internal wall-clock that cfg.exploration_fn receives.

1.0

Returns:

Type Description
dict[str, float]

Dictionary of scalar metrics: triggered (1/0),

dict[str, float]

norm_diff, condition, actor_loss,

dict[str, float]

critic_loss.

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 reset/step/dt.

required
max_steps int | None

Optional cap on episode length.

None
reference_signal ndarray | None

Override the env's reference (defaults to env.reference_signal when present).

None

Returns:

Type Description
dict[str, float]

Summary dict: steps, triggers, ep_return.

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:train_episode.

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 when save_gradients=True).

Parameters:

Name Type Description Default
path Union[str, Path, None]

Base directory (None → CWD).

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 namespace/repo_name on the Hugging Face Hub.

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. "me/my-etdhp".

required
folder_path Union[str, Path]

Local folder produced by :meth:save.

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 (fit_plant_model). Online fine-tuning uses one epoch per triggered step when enabled via online_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 f_θ after the offline phase.

False
Q Sequence[float]

Diagonal weights of the quadratic state cost xᵀ Q x. Length must equal n_state.

(1.0,)
R Sequence[float]

Diagonal weights of the quadratic control cost. Length must equal n_control.

(1.0,)
gamma float

Discount factor. Default 1.0 (finite-horizon / regulation-style, as in the reference).

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:EventTrigger). Must lie in (0, 0.5).

0.1
trigger_floor float

Minimum trigger threshold (state units). Small positive values suppress noise-induced retriggers when x ≈ 0.

0.001
exploration_fn Callable[[float], ndarray] | None

Optional callable (time_sec) -> np.ndarray of length n_control. If provided, the returned vector is added to the actor's target control during training, which injects persistent excitation into the NN update without perturbing the control actually applied to the plant. None disables exploration.

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 tanh(D_nn) by this value.

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 x.

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:

  1. Forward — predict the next state.
  2. Backward — extract the Jacobians F = ∂f/∂x and G = ∂f/∂u via 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 x.

required
n_control int

Length of the control vector u.

required
hidden_sizes Sequence[int]

Sizes of the tanh hidden layers. Defaults to (10, 10) to match the reference implementation.

(10, 10)
init_scale float

Uniform-init bound for each weight (no bias). The reference uses 0.5.

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 ρ. Paper uses 0.1 — lower values give tighter tracking at the cost of more triggers.

0.1
trigger_first_step bool

If True (default), the very first call to :meth:should_trigger always returns True so the networks and control run once at k = 0.

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 x ≈ 0 (the multiplicative ‖x_et‖ term collapses to zero at the origin otherwise).

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.