PyMC 常微分方程(ODE)贝叶斯建模指南:DifferentialEquation 从 API 到梯度实现
2026/9/16 19:07:21 网站建设 项目流程

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)可以明确两点官方定位:

  1. 该模块提供"工具"级的 ODE 求解支持,而不是一个独立的完整求解器套件;
  2. 由于所用求解器的特性,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_statesn_theta关键字专用参数*之后),t0默认取0

核心参数一览

参数类型含义说明
funccallable定义微分方程的函数必须可调用,签名必须为func(y, t, p),其中y(n_states,)状态向量,t为标量时刻,p(n_theta,)参数向量
timesarray求解并返回数值解的时刻数组构造时会被转为 tuple 存储,同时内部会把t0插入到数组最前面作为积分起点
n_statesint系统状态维度标量微分方程取 1;向量系统等于方程组个数,必须 ≥ 1
n_thetaint微分方程中未知参数的个数必须为正整数(> 0)
t0float初始条件对应的时刻默认 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的返回值有明确约束,它必须是以下三者之一:

  1. 单个pt.TensorVariable(会经pt.atleast_1d转成一维);
  2. Pythonlisttuple(逐元素pt.stack成一维张量);
  3. 其他类型直接抛出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)

要点拆解

  • 参数向量thetay0可以是随机变量:这里把alphay0都声明为先验分布,ode_model(theta=[alpha], y0=[y0])会在每次采样迭代中用当前参数值重新求解 ODE;
  • 数值解直接充当观测分布的均值forward的形状为(15, 1),与yobs对齐,因此可作为mu参与似然计算;
  • 维度一致性theta长度必须等于n_thetay0长度必须等于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做符号变换,构造新的微分方程系统:

  1. ypt声明为float64的 PyTensor 符号变量(注释明确指出:即使floatX=float32也使用 float64,因为 SciPy 积分器始终使用 float64);
  2. 计算状态雅可比 $J = \partial f / \partial y$ 与参数雅可比 $\partial f / \partial p$;
  3. 构造灵敏度的时间导数:

$$\frac{d}{dt}\left(\frac{\partial y}{\partial p}\right) = J \cdot \frac{\partial y}{\partial p} + \frac{\partial f}{\partial p}$$

  1. 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 的反向传播规则:

  1. 从缓存_output_sensitivities中取出与当前输入对应的符号灵敏度sens(形状(n_times, n_states, n_p),其中n_p = n_states + n_theta);
  2. 对每个参数 $p_j$,计算pt.sum(sens[:, :, j] * ograds),即灵敏度与上游梯度的逐元素乘积之和;
  3. 返回两个梯度张量:前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_nodegrad共用该缓存,避免在同一计算图中重复构造灵敏度张量。

八、小结

围绕 docs/source/api/ode.rst 所索引的pymc.ode模块,本文完成了从 API 到实现的全链路梳理:

  1. 建模入口DifferentialEquation(func, times, n_states, n_theta, t0=0),五个参数决定了 ODE 系统的全部语义,构造期即有严格校验;
  2. 使用范式ode_model(theta=[...], y0=[...])返回(n_times, n_states)数值解,可直接作为观测分布均值,随机变量参数使其适配 MCMC 推断;
  3. 梯度原理:通过 pymc/ode/utils.py 的augment_systemmake_sens_ic构造增广系统,一次积分同时得到状态与灵敏度,grad方法将其转译为反向传播所需的梯度;
  4. 工程细节:输入形状以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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询