动态前馈控制算法仿真 —— 基于干扰观测的超前补偿实践

“那年蒸汽总管压力波动，PID 总是慢半拍，换热出口温度像过山车。后来加了动态前馈，压力变送器一动，调节阀提前补偿，温度曲线立马平得像尺子画的。那一刻我才明白：反馈是亡羊补牢，前馈才是未雨绸缪。”

—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸

一、实际应用场景描述

在换热站、锅炉、精馏塔等存在显著可测扰动的工业场景中，纯 PID 控制往往力不从心：

┌──────────────────────────────────────────────┐

│ 动态前馈-反馈复合控制系统 │

│ │

│ [可测干扰] 蒸汽压力/流量 │

│ │ 干扰通道 (快) │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ 干扰观测器 (Disturbance │ │

│ │ Observer) │ │

│ │ • 实时采样 (100ms) │ │

│ │ • 滤波去噪 │ │

│ │ • 变化率计算 │ │

│ └────────────┬───────────────┘ │

│ │ 干扰信号 D(t) │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ 动态前馈补偿器 │ │

│ │ • 静态增益 Kff │ │

│ │ • 超前环节 Td·s │ │

│ │ • 滞后环节 1/(Ts+1) │ │

│ │ • 限幅保护 │ │

│ └────────────┬───────────────┘ │

│ │ 前馈补偿量 u_ff(t) │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ 加法器 (Summing Junction) │ │

│ │ u_total = u_pid + u_ff │ │

│ └────────────┬───────────────┘ │

│ │ 总控制量 │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ 被控对象 (Heat Exchanger) │ │

│ │ • 大滞后 (τ=30s) │ │

│ │ • 大惯性 (T=60s) │ │

│ │ • 非线性 │ │

│ └────────────┬───────────────┘ │

│ │ 控制通道 (慢) │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ 被控变量 (Process Value) │ │

│ │ • 出口温度 T_out │ │

│ │ • 设定值 SP │ │

│ └────────────┬───────────────┘ │

│ │ 偏差 e(t) = SP - PV │

│ ▼ │

│ ┌────────────────────────────┐ │

│ │ PID 反馈控制器 │ │

│ │ • 比例 P │ │

│ │ • 积分 I │ │

│ │ • 微分 D │ │

│ └───────────────────────────┘ │

│ │

│ 核心: 干扰可测 + 超前补偿 + 反馈兜底 + 动态整定 │

└──────────────────────────────────────────────┘

纯 PID vs 前馈-反馈复合控制

维度 纯 PID 控制 前馈-反馈复合

抗扰速度 ❌ 滞后 30~60s ✅ 干扰一出现即补偿

超调量 ❌ 常超调 5~10% ✅ 超调 < 2%

稳态误差 ❌ 积分饱和 ✅ 前馈抵消，积分减负

鲁棒性 ❌ 参数敏感 ✅ 前馈承担主要抗扰

调试难度 ❌ 需反复整定 ✅ 前馈参数物理意义明确

二、引入痛点

2.1 现场的真实困境

场景 现场发生了什么 根因

“蒸汽压力波动” “总管压力一掉，换热温度跟着跳水” 干扰通道比控制通道快

“PID 打满” “调节阀全开都跟不上扰动” 纯反馈控制带宽不足

“积分饱和” “扰动结束后 PID 还回不来” 积分项累积过大

“工艺投诉” “温度波动 ±5℃，产品不合格” 控制品质达不到要求

“能耗浪费” “为了抗扰，阀门常开大” 缺乏预见性调节

2.2 核心矛盾

控制的本质不是“等偏差出现再纠正”，而是“在偏差出现前抵消扰动”。前馈控制的核心前提是：扰动可测、通道已知、补偿可行。现场最大的问题是把前馈当成了“高级功能”，而不是“基础手段”。

2.3 我们要解决什么

用一段精简的 Python 程序，构建一个 动态前馈控制仿真系统，实现：

1. 干扰实时采集 —— 模拟蒸汽压力/流量扰动

2. 动态前馈算法 —— 超前+滞后+增益补偿

3. 前馈-反馈复合 —— PID 负责稳态，前馈负责动态

4. 性能指标量化 —— ISE、IAE、TV 评价控制效果

5. 可视化对比 —— 纯 PID vs 前馈-反馈

三、核心逻辑讲解

3.1 理论基础：前馈控制原理

本工具基于哈工程《工业过程控制》第五章“前馈控制”和第六章“复合控制”：

① 理想前馈控制

对于扰动通道 G_d(s) 和控制通道 G_c(s)G_p(s) ，理想前馈补偿器为：

G_{ff}(s) = -\frac{G_d(s)}{G_c(s)G_p(s)}

② 动态前馈（实用型）

考虑到可实现性，采用超前-滞后形式：

G_{ff}(s) = K_{ff} \cdot \frac{T_d s + 1}{T_f s + 1} \cdot e^{-\tau s}

其中：

- K_{ff} ：静态前馈增益（核心参数）

- T_d ：超前时间（匹配扰动通道惯性）

- T_f ：滞后时间（匹配控制通道惯性）

- \tau ：纯滞后补偿

③ 离散化实现（后向差分）

u_{ff}(k) = K_{ff} \cdot \frac{T_d}{T_s} [D(k) - D(k-1)] + K_{ff} \cdot D(k) - \frac{T_f}{T_s} u_{ff}(k-1) + \frac{T_f}{T_s} u_{ff}(k-2)

简化为一阶差分形式：

u_{ff}(k) = K_{ff} \cdot D(k) + K_d \cdot [D(k) - D(k-1)]

其中 K_d = K_{ff} \cdot T_d / T_s 为动态系数。

3.2 控制架构总览

┌─────────────┐

│ 干扰源 D(t) │

│ 蒸汽压力波动 │

└──────┬──────┘

│ 实时采样 (Δt=0.1s)

┌─────────▼─────────┐

│ 干扰观测器 │

│ • 低通滤波 │

│ • 变化率计算 │

│ • 死区处理 │

└─────────┬─────────┘

│ D_filtered(k)

┌─────────▼─────────┐

│ 动态前馈补偿器 │

│ u_ff(k) = Kff·D(k)│

│ + Kd·ΔD(k)│

│ • 限幅 [-100%,+100%]│

│ • 平滑滤波 │

└─────────┬─────────┘

│ u_ff(k)

┌─────────▼─────────┐

│ 加法器 │

│ u_total = u_pid + u_ff │

└─────────┬─────────┘

│ u_total(k)

┌─────────▼─────────┐

│ PID 反馈控制器 │

│ • P: Kp=1.5 │

│ • I: Ki=0.05 │

│ • D: Kd_pid=0.3 │

│ • 抗积分饱和 │

└─────────┬─────────┘

│ u_final(k)

▼

┌─────────────┐

│ 被控对象 │

│ Gp(s)=Ke^(-Ls)/(Ts+1)│

└─────────────┘

四、代码讲解（面向对象设计）

4.1 类结构总览

类名 职责 设计模式

"DisturbanceProfile" 干扰曲线（dataclass） 值对象

"FeedforwardConfig" 前馈配置 值对象

"PIDConfig" PID 配置 值对象

"LowPassFilter" 低通滤波器 策略模式

"DisturbanceObserver" 干扰观测器 观察者模式

"DynamicFeedforward" 动态前馈补偿器 策略模式

"PIDController" PID 反馈控制器 模板方法

"HeatExchangerModel" 换热器被控对象 领域模型

"ControlPerformance" 控制性能指标 封装

"FeedforwardSimulator" 仿真引擎（聚合根） 聚合根

"Visualizer" 可视化工具 封装

4.2 核心代码（完整可运行）

完整源码约 380 行，包含 10 个类、动态前馈算法、PID 反馈、性能指标、可视化。

以下为精简核心版，可直接复制运行。

<details><summary>🔧 完整源码（点击展开/折叠）</summary>

"""

动态前馈控制算法仿真 —— 基于干扰观测的超前补偿

参考哈尔滨工程大学《工业过程控制》第五章"前馈控制"

"""

from dataclasses import dataclass, field

from typing import List, Tuple, Optional, Deque

from enum import Enum, auto

import numpy as np

import matplotlib.pyplot as plt

from collections import deque

import math

from datetime import datetime

# ============================================================

# 1. 基础数据结构（值对象）

# ============================================================

@dataclass

class DisturbanceProfile:

"""干扰曲线配置 —— 值对象"""

name: str

start_time: float = 50.0 # 干扰开始时间 (s)

duration: float = 100.0 # 干扰持续时间 (s)

magnitude: float = -0.2 # 干扰幅度 (-20% 压力下降)

ramp_time: float = 5.0 # 上升时间 (s)

noise_level: float = 0.02 # 噪声水平

def get_value(self, t: float) -> float:

"""获取 t 时刻的干扰值"""

if t < self.start_time:

base = 0.0

elif t < self.start_time + self.ramp_time:

# 斜坡上升

progress = (t - self.start_time) / self.ramp_time

base = self.magnitude * progress

elif t < self.start_time + self.duration:

base = self.magnitude

elif t < self.start_time + self.duration + self.ramp_time:

# 斜坡下降

progress = (t - self.start_time - self.duration) / self.ramp_time

base = self.magnitude * (1 - progress)

else:

base = 0.0

# 添加测量噪声

noise = np.random.normal(0, self.noise_level) if self.noise_level > 0 else 0

return base + noise

@dataclass

class FeedforwardConfig:

"""前馈配置 —— 值对象"""

static_gain: float = 0.8 # 静态前馈增益 Kff

lead_time: float = 8.0 # 超前时间 Td (s) - 匹配扰动通道

lag_time: float = 25.0 # 滞后时间 Tf (s) - 匹配控制通道

dead_time: float = 2.0 # 纯滞后 τ (s)

dynamic_coeff: float = 0.5 # 动态系数 (Td/Ts)

output_limit: Tuple[float, float] = (-1.0, 1.0) # 输出限幅

enable_dynamic: bool = True # 启用动态项

filter_alpha: float = 0.1 # 低通滤波系数

@dataclass

class PIDConfig:

"""PID配置 —— 值对象"""

kp: float = 1.5 # 比例增益

ki: float = 0.05 # 积分增益

kd: float = 0.3 # 微分增益

output_limit: Tuple[float, float] = (-1.0, 1.0)

anti_windup: bool = True

# ============================================================

# 2. 低通滤波器（策略模式）

# ============================================================

class LowPassFilter:

"""一阶低通滤波器 —— 策略模式"""

def __init__(self, alpha: float = 0.1):

self.alpha = alpha

self.prev_output = 0.0

self.initialized = False

def filter(self, input_value: float) -> float:

if not self.initialized:

self.prev_output = input_value

self.initialized = True

return input_value

output = self.alpha * input_value + (1 - self.alpha) * self.prev_output

self.prev_output = output

return output

def reset(self):

self.initialized = False

self.prev_output = 0.0

# ============================================================

# 3. 干扰观测器（观察者模式）

# ============================================================

class DisturbanceObserver:

"""干扰观测器 —— 观察者模式"""

def __init__(self, sample_period: float = 0.1):

self.sample_period = sample_period

self.filter = LowPassFilter(alpha=0.15)

self.history: Deque[Tuple[float, float]] = deque(maxlen=100)

self.prev_value = 0.0

self.prev_time = 0.0

def observe(self, raw_disturbance: float, t: float) -> dict:

"""观测干扰并返回处理结果"""

# 滤波

filtered = self.filter.filter(raw_disturbance)

# 计算变化率

dt = t - self.prev_time if self.prev_time > 0 else self.sample_period

derivative = (filtered - self.prev_value) / dt if dt > 0 else 0.0

# 死区处理（小变化忽略）

deadband = 0.005

if abs(filtered) < deadband:

filtered = 0.0

derivative = 0.0

# 存储历史

self.history.append((t, filtered, derivative))

result = {

'raw': raw_disturbance,

'filtered': filtered,

'derivative': derivative,

'timestamp': t

}

# 更新状态

self.prev_value = filtered

self.prev_time = t

return result

def get_recent_change(self, window: float = 10.0) -> float:

"""获取最近一段时间的变化量"""

if len(self.history) < 2:

return 0.0

current_time = self.history[-1][0]

recent_values = [v for ts, v, _ in self.history if current_time - ts <= window]

if len(recent_values) < 2:

return 0.0

return recent_values[-1] - recent_values[0]

# ============================================================

# 4. 动态前馈补偿器（策略模式）

# ============================================================

class DynamicFeedforward:

"""动态前馈补偿器 —— 策略模式"""

def __init__(self, config: FeedforwardConfig, sample_period: float = 0.1):

self.config = config

self.ts = sample_period

self.filter = LowPassFilter(alpha=config.filter_alpha)

# 状态变量

self.prev_disturbance = 0.0

self.prev_ff_output = 0.0

self.disturbance_history: Deque[Tuple[float, float]] = deque(maxlen=50)

# 计算动态系数

self.kd_dynamic = config.static_gain * config.lead_time / self.ts if config.enable_dynamic else 0.0

self.kf_lag = config.lag_time / self.ts if config.lag_time > 0 else 1.0

def compute(self, disturbance: float, derivative: float, t: float) -> float:

"""计算动态前馈输出"""

# 滤波

d_filtered = self.filter.filter(disturbance)

# 存储历史（用于纯滞后补偿）

self.disturbance_history.append((t, d_filtered))

# 纯滞后补偿：查找 τ 秒前的干扰值

delayed_disturbance = d_filtered

if self.config.dead_time > 0:

target_time = t - self.config.dead_time

for ts, val in reversed(self.disturbance_history):

if ts <= target_time:

delayed_disturbance = val

break

# 动态前馈公式：u_ff = Kff * D + Kd * dD/dt

ff_static = self.config.static_gain * delayed_disturbance

ff_dynamic = self.kd_dynamic * derivative if self.config.enable_dynamic else 0.0

ff_raw = ff_static + ff_dynamic

# 一阶滞后环节（近似）

ff_output = (ff_raw + (self.kf_lag - 1) * self.prev_ff_output) / self.kf_lag

# 限幅

ff_output = max(self.config.output_limit[0],

min(self.config.output_limit[1], ff_output))

# 更新状态

self.prev_disturbance = d_filtered

self.prev_ff_output = ff_output

return ff_output

def reset(self):

"""重置状态"""

self.filter.reset()

self.prev_disturbance = 0.0

self.prev_ff_output = 0.0

self.disturbance_history.clear()

# ============================================================

# 5. PID 反馈控制器（模板方法）

# ============================================================

class PIDController:

"""PID 反馈控制器 —— 模板方法"""

def __init__(self, config: PIDConfig, sample_period: float = 0.1):

self.config = config

self.ts = sample_period

# 状态变量

self.integral = 0.0

self.prev_error = 0.0

self.prev_output = 0.0

self._last_reset_time = 0.0

def compute(self, setpoint: float, process_variable: float, t: float = 0.0) -> float:

"""计算 PID 输出"""

error = setpoint - process_variable

# 比例项

p_term = self.config.kp * error

# 积分项（带抗积分饱和）

if self.config.anti_windup:

# 仅当输出未饱和时才积分

if not (self.prev_output >= self.config.output_limit[1] and error > 0) and \

not (self.prev_output <= self.config.output_limit[0] and error < 0):

self.integral += error * self.ts

else:

self.integral += error * self.ts

i_term = self.config.ki * self.integral

# 微分项（对 PV 微分，避免设定值突变冲击）

d_term = 0.0

if self.ts > 0:

d_term = -self.config.kd * (process_variable - self.prev_error) / self.ts

# 总输出

output = p_term + i_term + d_term

# 限幅

output = max(self.config.output_limit[0],

min(self.config.output_limit[1], output))

# 更新状态

self.prev_error = process_variable

self.prev_output = output

return output

def reset(self):

"""重置控制器状态"""

self.integral = 0.0

self.prev_error = 0.0

self.prev_output = 0.0

self._last_reset_time = 0.0

# ============================================================

# 6. 换热器被控对象（领域模型）

# ============================================================

class HeatExchangerModel:

"""换热器被控对象模型 —— 领域模型"""

def __init__(self, gain: float = 1.0, time_constant: float = 60.0,

dead_time: float = 30.0, sample_period: float = 0.1):

self.K = gain # 过程增益

self.T = time_constant # 时间常数 (s)

self.L = dead_time # 纯滞后 (s)

self.ts = sample_period

# 状态变量（一阶惯性+滞后）

self.state = 0.0

self.history: Deque[Tuple[float, float]] = deque(maxlen=int(self.L / self.ts) + 10)

self.noise_level = 0.01

# 扰动通道参数（通常比控制通道快）

self.Kd = -0.5 # 扰动增益（负号表示压力下降导致温度下降）

self.Td = 20.0 # 扰动时间常数（比控制通道快）

self.Ld = 5.0 # 扰动滞后（比控制通道小）

self.disturbance_state = 0.0

self.disturbance_history: Deque[Tuple[float, float]] = deque(maxlen=int(self.Ld / self.ts) + 10)

def step(self, control_input: float, disturbance: float, t: float) -> float:

"""执行一个仿真步长"""

# 控制通道：一阶惯性 + 纯滞后

# dx/dt = (K*u - x) / T

self.state += (self.K * control_input - self.state) * self.ts / self.T

# 存储控制作用历史

self.history.append((t, self.state))

# 扰动通道（更快的动态）

self.disturbance_state += (self.Kd * disturbance - self.disturbance_state) * self.ts / self.Td

# 存储扰动历史

self.disturbance_history.append((t, self.disturbance_state))

# 读取滞后后的输出

delayed_control = self.state

if self.L > 0:

target_time = t - self.L

for ts, val in reversed(self.history):

if ts <= target_time:

delayed_control = val

break

# 读取扰动滞后

delayed_disturbance = self.disturbance_state

if self.Ld > 0:

target_time = t - self.Ld

for ts, val in reversed(self.disturbance_history):

if ts <= target_time:

delayed_disturbance = val

break

# 总输出 = 控制作用 + 扰动作用 + 噪声

output = delayed_control + delayed_disturbance

noise = np.random.normal(0, self.noise_level) if self.noise_level > 0 else 0

return output + noise

def reset(self):

"""重置模型状态"""

self.state = 0.0

self.disturbance_state = 0.0

self.history.clear()

self.disturbance_history.clear()

# ============================================================

# 7. 控制性能指标（封装）

# ============================================================

class ControlPerformance:

"""控制性能指标计算 —— 封装"""

@staticmethod

def ise(time_series: List[float], pv_series: List[float], sp: float) -> float:

"""积分平方误差 (Integral Squared Error)"""

return sum((sp - pv)**2 for pv in pv_series) * (time_series[1] - time_series[0]) if len(time_series) > 1 else 0

@staticmethod

def iae(time_series: List[float], pv_series: List[float], sp: float) -> float:

"""积分绝对误差 (Integral Absolute Error)"""

return sum(abs(sp - pv) for pv in pv_series) * (time_series[1] - time_series[0]) if len(time_series) > 1 else 0

@staticmethod

def itae(time_series: List[float], pv_series: List[float], sp: float) -> float:

"""积分时间加权绝对误差 (Integral Time-weighted Absolute Error)"""

return sum(t * abs(sp - pv) for t, pv in zip(time_series, pv_series)) * (time_series[1] - time_series[0]) if len(time_series) > 1 else 0

@staticmethod

def tv(control_series: List[float]) -> float:

"""控制量变化量 (Total Variation) - 衡量控制平稳性"""

return sum(abs(control_series[i] - control_series[i-1]) for i in range(1, len(control_series)))

@staticmethod

def overshoot(pv_series: List[float], sp: float) -> float:

"""超调量 (%)"""

if sp == 0:

return 0.0

max_pv = max(pv_series)

return max(0, (max_pv - sp) / sp * 100) if sp > 0 else 0.0

@staticmethod

def settling_time(time_series: List[float], pv_series: List[float], sp: float, tolerance: float = 0.02) -> float:

"""调节时间 (进入±2%误差带并不再超出)"""

band = abs(sp) * tolerance

settled_idx = len(pv_series) - 1

for i in range(len(pv_series) - 1, 0, -1):

if abs(pv_series[i] - sp) > band:

settled_idx = i + 1

break

return time_series[settled_idx] if settled_idx < len(time_series) else time_series[-1]

@staticmethod

def evaluate_all(time_series: List[float], pv_series: List[float],

control_series: List[float], sp: float) -> dict:

"""计算所有性能指标"""

return {

'ISE': ControlPerformance.ise(time_series, pv_series, sp),

'IAE': ControlPerformance.iae(time_series, pv_series, sp),

'ITAE': ControlPerformance.itae(time_series, pv_series, sp),

'TV': ControlPerformance.tv(control_series),

'Overshoot': ControlPerformance.overshoot(pv_series, sp),

'SettlingTime': ControlPerformance.settling_time(time_series, pv_series, sp)

}

# ============================================================

# 8. 仿真引擎（聚合根）

# ============================================================

class FeedforwardSimulator:

"""动态前馈控制仿真引擎 —— 聚合根"""

def __init__(self, sample_period: f

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！