multiprocess进程池内Scipy.optimize.curve_fit的异常处理问题

问题现象 #

• 目标为对图像多列数据并行执行批量一维高斯拟合，选用multiprocess而非标准库multiprocessing，用于解决数组参数跨进程传递时的序列化（pickle/dill）问题，基础流程可正常运行。
• 序列化函数内调用curve_fit时触发异常：注释掉curve_fit逻辑直接返回初始猜测参数时程序正常；普通for循环中可正常捕获的拟合不收敛RuntimeError，在进程池环境下无法捕获。
• 拟合不收敛时，进程池调用get方法抛出错误，溯源到scipy.optimize.minpack.py的_lmdif函数调用，报错信息为：

minpack.error: Result from function call is not a proper array of floats.
ValueError: object too deep for desired array

• 已尝试操作：新增捕获ValueError类型异常，问题未解决。

核心问题排查与修复点 #

• 数组维度不匹配是报错直接原因
  np.split(framey, sizey[1], axis=1)切分得到的单列数据是形状为(行数, 1)的二维数组，curve_fit依赖的minpack底层Fortran接口要求传入的因变量必须是和自变量同形状的一维浮点数组。单进程场景下numpy会自动做隐式展平容错，但子进程中序列化后的数组传入底层接口时不会触发这个隐式转换，直接抛出数组嵌套过深、返回值不是合法浮点数组的错误。这类错误发生在拟合函数调用阶段，不属于拟合不收敛触发的RuntimeError，仅捕获RuntimeError无法拦截。
  修复方式：切分得到的每列数据传入拟合逻辑前，调用.flatten()或.ravel()强制转换为一维数组。

• 入口守卫位置错误
  原代码将if __name__ == '__main__'写在processframe函数内部，子进程启动导入模块时会直接跳过守卫包裹的逻辑，导致拟合函数定义、进程池初始化逻辑完全不执行，还会触发函数作用域丢失的序列化问题。入口守卫必须放在模块最外层，不能嵌套在业务函数内部。

• 进程池写法存在语法错误
  pool.close、pool.join是对象方法，必须加括号调用（即pool.close()、pool.join()），否则两行代码不会生效，会导致进程池资源泄漏、任务未完成就提前返回结果。另外不要在pool.map中传入lambda表达式，即使是支持dill序列化的multiprocess模块，对跨进程传递嵌套lambda的兼容性也很差，直接将单任务处理逻辑定义为顶层函数传入即可。

• 异常捕获范围不足
  除了拟合不收敛触发的RuntimeError，还需要捕获ValueError、TypeError、np.linalg.LinAlgError等参数不合法、矩阵运算失败触发的异常，避免单列数据拟合失败直接中断整个进程池任务。

• 嵌套函数定义增加序列化风险
  原代码在多层函数内部反复定义gguess1d、gauss1d等拟合工具函数，跨进程传递时会反复序列化闭包作用域，既增加性能开销，也容易触发作用域变量丢失的问题。将这类工具函数提到模块顶层定义即可规避该问题。

修复后参考代码 #

import numpy as np
import scipy.ndimage as nd
from scipy.optimize import curve_fit
import multiprocess

# 顶层定义拟合辅助函数，避免闭包序列化问题
def gguess1d(a):
    x = np.linspace(1, np.amax(a.shape), np.amax(a.shape), endpoint=True)
    fata = nd.gaussian_filter1d(a, 2)
    GuessY0 = np.amin(a)
    GuessA = np.amax(a) - GuessY0
    X0A = np.amax(fata)
    X0Y0 = np.amin(fata)
    X0form = np.multiply(fata + X0Y0, (fata + X0Y0 > (X0A + X0Y0)/3))
    GuessX0 = np.sum(np.multiply(X0form, x)) / (1 + np.sum(X0form))
    valid_mask = a - GuessY0 > 2
    GuessSigma = np.sqrt(
        np.sum(np.square(np.multiply(x - GuessX0, np.multiply(a - GuessY0, valid_mask)))) 
        / (1 + np.sum(valid_mask))
    ) / 4
    solution = np.asarray([GuessY0, GuessA, GuessX0, GuessSigma])
    return np.nan_to_num(np.real(solution.flatten()))

def gauss1d(x, aa, bb, cc, dd, size, therange):
    return (
        aa + bb * np.exp(-2*(x-cc)**2 / (np.abs(dd)+0.5)**2)
        + 2000*(cc<1)*(1-cc)**2
        + 2000*(cc>size)*(cc-size)**2
        + 2000*(np.abs(aa-bb)>3*therange)*(np.abs(aa-bb) - 3*therange**2)
    )

def fit_single_column(col):
    a = col.flatten()
    therange = np.amax(a) - np.amin(a)
    size = np.amax(a.shape)
    x = np.linspace(1, size, size, endpoint=True)
    guess = np.nan_to_num(np.real(gguess1d(a))).flatten()
    try:
        # 闭包变量作为显式参数传入，规避序列化作用域问题
        params, _ = curve_fit(
            lambda x, aa, bb, cc, dd: gauss1d(x, aa, bb, cc, dd, size, therange),
            x, a, p0=guess
        )
        params[3] = np.abs(params[3]) + 0.5
    # 覆盖所有拟合阶段可能触发的异常
    except (RuntimeError, ValueError, TypeError, np.linalg.LinAlgError):
        params = np.array([1,1,1,1])
    return np.nan_to_num(np.real(np.asarray(params.flatten())))

def processframe(framey, n_workers=2):
    sizey = framey.shape
    # 切分列后直接转一维数组，解决维度不匹配问题
    columnsy = [col.flatten() for col in np.split(framey, sizey[1], axis=1)]
    # 用上下文管理器自动管理进程池生命周期
    with multiprocess.Pool(n_workers) as pool:
        paramsy = pool.map(fit_single_column, columnsy)
    return np.transpose(np.stack(paramsy, axis=1))

# 入口守卫放在模块最外层
if __name__ == '__main__':
    # 测试示例
    test_img = np.random.randn(100, 50)
    fit_res = processframe(test_img)
    print(f"拟合结果形状：{fit_res.shape}")

内容的提问来源于stack exchange，提问作者Matt Reed