PyMC 常微分方程(ODE)贝叶斯建模指南:DifferentialEquation 从 API 到梯度实现
【免费下载链接】pymcBayesian Modeling and Probabilistic Programming in Python项目地址: https://gitcode.com/GitHub_Trending/py/pymc
本指南围绕 PyMC 官方 API 文档 docs/source/api/ode.rst 所对应的pymc.ode模块展开,系统讲解如何在贝叶斯模型中嵌入常微分方程(ODE)动力学:从DifferentialEquation的数学形式与全部构造参数,到把 ODE 数值解作为观测分布均值参与 MCMC 采样的完整实战流程,并深入源码剖析其基于"增广系统 + 灵敏度矩阵"的自动梯度实现原理。读完本文,你将能够在 PyMC 中直接定义带未知参数的 ODE 系统,并将其作为随机模型的确定性前向内核完成参数推断。
一、模块定位:pymc.ode是什么
pymc.ode是 PyMC 中用于"对常微分方程执行贝叶斯推理"的工具子模块,在 pymc/ode/init.py 中导出,并经 pymc/init.py 以pm.ode挂载到顶层命名空间,同时把唯一的公开类DifferentialEquation直接暴露出来。
它的核心用途是:把形式如下的初值问题嵌入到一个可微分的概率模型图中
$$\frac{dy}{dt} = f(y, t, p), \quad y(t_0) = y_0$$
其中 $y$ 是系统的状态向量(如药物浓度、种群数量、化学反应物浓度),$p$ 是动力学参数(如速率常数、增长率)。在贝叶斯框架下,这些 $p$ 正是我们希望通过观测数据 $y_{\text{obs}}$ 反推其后验分布的未知量。
从模块 docstring(见 pymc/ode/init.py)可以明确两点官方定位:
- 该模块提供"工具"级的 ODE 求解支持,而不是一个独立的完整求解器套件;
- 由于所用求解器的特性,ODE 求解过程可能较慢,模块文档明确提示:基于 PyMC 的另一个库
sunode使用 SUNDIALS 套件实现了 Adams 方法与 BDF(向后差分公式),速度远超pm.ode的默认实现——如果你的模型对求解速度敏感,这是一个值得了解的替代方向。
二、DifferentialEquation:把 ODE 变成 PyTensor 算子
DifferentialEquation是整个pymc.ode模块唯一公开的核心类,定义在 pymc/ode/ode.py。从源码结构看,它继承自pytensor.graph.op.Op,即一个自定义 PyTensor 算子:向计算图输入初始条件y0与参数向量theta,输出各观测时刻的数值解(以及可选的灵敏度张量)。这正是它能够无缝嵌入pm.Model()、参与梯度计算与 HMC 采样的根本原因。
类的 docstring 给出了完整的数学形式与构造签名:
dy/dt = f(y, t, p) y(t0) = y0构造函数为:
DifferentialEquation(func, times, *, n_states, n_theta, t0=0)注意n_states、n_theta是关键字专用参数(*之后),t0默认取0。
核心参数一览
| 参数 | 类型 | 含义 | 说明 |
|---|---|---|---|
func | callable | 定义微分方程的函数 | 必须可调用,签名必须为func(y, t, p),其中y为(n_states,)状态向量,t为标量时刻,p为(n_theta,)参数向量 |
times | array | 求解并返回数值解的时刻数组 | 构造时会被转为 tuple 存储,同时内部会把t0插入到数组最前面作为积分起点 |
n_states | int | 系统状态维度 | 标量微分方程取 1;向量系统等于方程组个数,必须 ≥ 1 |
n_theta | int | 微分方程中未知参数的个数 | 必须为正整数(> 0) |
t0 | float | 初始条件对应的时刻 | 默认 0,即 $y(t_0)=y_0$ |
构造时的参数校验
__init__(pymc/ode/ode.py)会对参数做前置检查,违反任一条件都会抛出ValueError:
func不可调用 →"Argument func must be callable."n_states < 1→"Argument n_states must be at least 1."n_theta <= 0→"Argument n_theta must be positive."
这些错误消息在 tests/ode/test_ode.py 的TestErrors中均有对应的单元测试逐一验证。
三、快速上手:Logistic 微分方程
DifferentialEquationdocstring 中给出了一个最小可用示例——Logistic 微分方程,这是理解整个 API 的最佳起点:
import numpy as np from pymc.ode import DifferentialEquation def odefunc(y, t, p): # Logistic differential equation: dy/dt = p0 * y * (1 - y) return p[0] * y[0] * (1 - y[0]) times = np.arange(0.5, 5, 0.5) ode_model = DifferentialEquation(func=odefunc, times=times, n_states=1, n_theta=1, t0=0)这里odefunc接收三个参数:当前状态y(一维数组)、当前时刻t、参数向量p,返回导数值。随后调用算子即可得到所有times时刻的数值解:
solution = ode_model(y0=[0.1], theta=[0.5]) # 形状为 (n_times, n_states)返回值形状为(len(times), n_states):行对应每个观测时刻,列对应每个状态变量。
关于func返回值类型的约定
augment_system(见 pymc/ode/utils.py)对func的返回值有明确约束,它必须是以下三者之一:
- 单个
pt.TensorVariable(会经pt.atleast_1d转成一维); - Python
list或tuple(逐元素pt.stack成一维张量); - 其他类型直接抛出
TypeError(提示"Unexpected type, ... returned by ode_func.")。
同时返回值维度必须是 0 维或 1 维,若返回 2 维张量会抛出ValueError。对应的边界测试覆盖了返回 set、dict、2 维 list、2 维 tensor 四种非法情况(tests/ode/test_ode.py)。
四、把 ODE 嵌入贝叶斯模型:完整可运行示例
DifferentialEquation的真正价值在于"数值解可以出现在概率密度函数中",从而让 MCMC 对动力学参数进行推断。tests/ode/test_ode.py 中的test_scalar_ode_1_param给出了一个完整、可直接复现的建模流程,这也是 ODE 贝叶斯推断的标准范式:
import pymc as pm import pytensor from pymc.ode import DifferentialEquation from pymc.testing import fast_unstable_sampling_mode def system(y, t, p): # dy/dt = exp(-t) - p0 * y return np.exp(-t) - p[0] * y[0] times = np.array([0.5, 1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0, 5.5, 6.0, 6.5, 7.0, 7.5]) yobs = np.array( [0.31, 0.57, 0.51, 0.55, 0.47, 0.42, 0.38, 0.3, 0.26, 0.22, 0.22, 0.14, 0.14, 0.09, 0.1] )[:, np.newaxis] ode_model = DifferentialEquation(func=system, t0=0, times=times, n_states=1, n_theta=1) with pm.Model() as model: alpha = pm.HalfCauchy("alpha", 1) # 动力学参数先验 y0 = pm.LogNormal("y0", 0, 1) # 初始条件先验 sigma = pm.HalfCauchy("sigma", 1) # 观测噪声 forward = ode_model(theta=[alpha], y0=[y0]) # 前向求解,作为均值 y = pm.LogNormal("y", mu=pm.math.log(forward), sigma=sigma, observed=yobs) idata = pm.sample(50, tune=0, chains=1)要点拆解
- 参数向量
theta与y0可以是随机变量:这里把alpha和y0都声明为先验分布,ode_model(theta=[alpha], y0=[y0])会在每次采样迭代中用当前参数值重新求解 ODE; - 数值解直接充当观测分布的均值:
forward的形状为(15, 1),与yobs对齐,因此可作为mu参与似然计算; - 维度一致性:
theta长度必须等于n_theta,y0长度必须等于n_states,否则__call__(pymc/ode/ode.py)会抛出pymc.exceptions.ShapeError,错误消息形如"Length of theta is wrong. (actual (2,) != expected (1,))"(对应测试见 tests/ode/test_ode.py); - 向量系统同理:
tests/ode/test_ode.py中的test_vector_ode_1_param演示了经典的 SIR 类两状态模型,odefunc返回[ds, di]列表,y0=[0.99, 0.01],n_states=2。
五、梯度从哪来:增广系统与灵敏度矩阵
贝叶斯采样(尤其是 NUTS/HMC)要求整个模型图可微,但scipy.integrate.odeint只是数值积分器,本身不提供解析梯度。pymc.ode的巧妙之处在于:把"状态 + 状态对参数的偏导数(灵敏度)"拼成一个增广系统,一次性积分求解,从而通过前向灵敏度分析获得精确梯度。
5.1 增广系统augment_system
核心实现在 pymc/ode/utils.py。它对用户定义的ode_func做符号变换,构造新的微分方程系统:
- 把
y、p、t声明为float64的 PyTensor 符号变量(注释明确指出:即使floatX=float32也使用 float64,因为 SciPy 积分器始终使用 float64); - 计算状态雅可比 $J = \partial f / \partial y$ 与参数雅可比 $\partial f / \partial p$;
- 构造灵敏度的时间导数:
$$\frac{d}{dt}\left(\frac{\partial y}{\partial p}\right) = J \cdot \frac{\partial y}{\partial p} + \frac{\partial f}{\partial p}$$
- 用
pytensor.function编译成可调用的system,输入(y, t, p, dydp_vec),输出(t_yhat, ddt_dydp)。
这样原 ODE 与灵敏度 ODE 被"缝合"成一个整体,在DifferentialEquation._system(pymc/ode/ode.py)中统一喂给odeint。
5.2 灵敏度初始条件make_sens_ic
灵敏度矩阵的初值并非任意设定,而是由 pymc/ode/utils.py 的make_sens_ic生成,具有固定的结构:
- 形状为
(n_states, n_states + n_theta)的矩阵; - 前
n_states列(对应参数向量中的初始条件部分)为单位矩阵 $I$,因为 $\partial y_0 / \partial y_0 = I$; - 后
n_theta列(对应 ODE 参数部分)全为 0,因为 $y_0$ 与动力学参数无关; - 最终
ravel成一维向量参与增广积分。
对应的形状断言测试见 tests/ode/test_ode.py,覆盖了标量/向量系统与 1~3 个参数的多种组合。例如 1 状态 1 参数系统,灵敏度初值恒为[1, 0]。
5.3grad方法:把灵敏度变成梯度
DifferentialEquation.grad(pymc/ode/ode.py)实现了 PyTensor Op 的反向传播规则:
- 从缓存
_output_sensitivities中取出与当前输入对应的符号灵敏度sens(形状(n_times, n_states, n_p),其中n_p = n_states + n_theta); - 对每个参数 $p_j$,计算
pt.sum(sens[:, :, j] * ograds),即灵敏度与上游梯度的逐元素乘积之和; - 返回两个梯度张量:前
n_states个对应y0,后n_theta个对应theta。
grad的正确性由 tests/ode/test_utils.py 验证:对dy/dt = exp(-t) - p0*y,用解析解的导数与数值积分灵敏度对比,rtol=1e-5内一致。
5.4 一次前向积分,两种输出
_simulate(pymc/ode/ode.py)把所有状态与灵敏度拼接为初始向量后调用scipy.integrate.odeint,随后拆分:
- 状态解
y:形状(n_times, n_states); - 灵敏度
sens:形状(n_times, n_states, n_p)。
infer_shape(pymc/ode/ode.py)据此声明输出形状。默认__call__只返回状态解;传入return_sens=True可同时拿到灵敏度,供调试或自定义梯度使用:
states, sens = ode_model(y0=[0.1], theta=[0.5], return_sens=True)六、数值精度与求解器说明
- 求解器:积分由
scipy.integrate.odeint完成(pymc/ode/ode.py),并在积分前把时间数组改写为t0插入后的_augmented_times(pymc/ode/ode.py),因此times无需包含t0; - 精度:数值解精度可由 tests/ode/test_ode.py 的
test_simulate佐证——对dy/dt = exp(-t) - a*y的解析解对比,rtol=1e-5内一致; - logp 一致性:
test_logp_scalar_ode(tests/ode/test_ode.py)把 ODE 前向解手动代入scipy.stats.norm.logpdf计算"手工 logp",与pm.Model().compile_logp()的结果做allclose对比,确保 ODE 算子接入概率图后似然计算完全正确; - 性能提醒:模块 docstring 明确提示 ODE 求解可能偏慢,对速度敏感的场景可考虑基于 SUNDIALS 的
sunode方案(实现了 Adams 方法与 BDF 公式)。
七、算子相等性与缓存机制
DifferentialEquation通过__props__ = ("func", "times", "n_states", "n_theta", "t0")(pymc/ode/ode.py)声明其"身份属性",PyTensor 据此判断两个 Op 是否等价:函数体、求解时刻、状态数、参数数、初值时刻全部相同的两个实例相等。这一点被test_op_equality(tests/ode/test_ode.py)验证——相同配置的实例相等、不同times的实例不相等。
此外,实例内部维护_apply_nodes与_output_sensitivities两个字典,以输入哈希为键缓存符号灵敏度(pymc/ode/ode.py),make_node与grad共用该缓存,避免在同一计算图中重复构造灵敏度张量。
八、小结
围绕 docs/source/api/ode.rst 所索引的pymc.ode模块,本文完成了从 API 到实现的全链路梳理:
- 建模入口:
DifferentialEquation(func, times, n_states, n_theta, t0=0),五个参数决定了 ODE 系统的全部语义,构造期即有严格校验; - 使用范式:
ode_model(theta=[...], y0=[...])返回(n_times, n_states)数值解,可直接作为观测分布均值,随机变量参数使其适配 MCMC 推断; - 梯度原理:通过 pymc/ode/utils.py 的
augment_system与make_sens_ic构造增广系统,一次积分同时得到状态与灵敏度,grad方法将其转译为反向传播所需的梯度; - 工程细节:输入形状以
ShapeError严格校验,Op 相等性由__props__定义,求解精度与 logp 正确性均有测试用例背书。
如需深入阅读源码,推荐按以下顺序研读:核心算子 pymc/ode/ode.py、灵敏度工具 pymc/ode/utils.py、集成测试 tests/ode/test_ode.py、梯度数值验证 tests/ode/test_utils.py,并可从 docs/source/api.rst 了解整个 API 文档的组织结构。
【免费下载链接】pymcBayesian Modeling and Probabilistic Programming in Python项目地址: https://gitcode.com/GitHub_Trending/py/pymc
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考