如何快速从Pyomo Concrete Model提取结果？优化提取性能

问题背景 #

我正在从Pyomo实例提取结果，实例包含多个索引化Expression和Param，索引规模约675，共13个涉及求和/乘积的索引化表达式。目前的提取代码耗时甚至超过模型求解，怀疑是distance_traveled表达式导致性能瓶颈，想找无需逐个调用value()的高效提取或数值计算方法。

当前提取代码 #

from pyomo.core import ConcreteModel, Expression, Param, value, Var
import pandas as pd
def extract_values_from_instance(instance: ConcreteModel):
    model_attributes = {
        "model_vars": instance.component_map(ctype=Var),
        "model_param": instance.component_map(ctype=Param),
        "model_expr": instance.component_map(ctype=Expression),
    }
    LOGGER.info('Extracting model values')
    list_of_series = []
    scalars_dict = {}
    for values in model_attributes.values():
        for key, var in values.items():
            if var.is_indexed():
                list_of_series.append(
                    pd.Series(
                        {k: value(v) for k, v in var.extract_values().items()},
                        index=var.extract_values().keys(),
                        name=key,
                        dtype="float64",
                    )
                )
            else:
                scalars_dict[key] = var()
    return pd.concat(list_of_series, axis=1), scalars_dict

可疑表达式（distance_traveled） #

def distance_traveled_rule(model, current_time) -> float:
    if current_time == model.time_start.value:
        return model.initial_distance # Param
    distance_traveled = (
        model.initial_distance # Param
        + sum(              
            model.velocity[time_stamp] * interval
            for time_stamp in model.velocity
            if time_stamp < current_time
        )
    )
    return distance_traveled
model.distance_traveled = Expression(
    model.velocity.index_set(),
    rule=distance_traveled_rule,
    doc="Distance traveled at each time stamp",
)

优化方案 #

1. 消除extract_values()重复调用与冗余value() #

当前代码对同一个组件多次调用var.extract_values()，且额外套了一层value()——实际上extract_values()内部已经完成了求值（等价于自动调用value()），这是核心性能浪费点。修改后：

if var.is_indexed():
    extracted_data = var.extract_values()
    list_of_series.append(
        pd.Series(
            extracted_data,
            index=extracted_data.keys(),
            name=key,
            dtype="float64",
        )
    )

2. 重构distance_traveled的O(n²)求和逻辑 #

原规则对每个时间戳都遍历所有更早的时间点求和，时间复杂度为O(n²)，当n=675时会产生近45万次无效循环。改成增量累积或提前计算累积和：

方式一：利用有序时间戳的索引定位

def distance_traveled_rule(model, current_time):
    if current_time == model.time_start.value:
        return model.initial_distance
    # 预先排序时间戳集合
    sorted_times = sorted(model.velocity.index_set())
    idx = sorted_times.index(current_time)
    # 仅对前idx个时间戳求和
    cumulative_velocity = sum(model.velocity[t] for t in sorted_times[:idx])
    return model.initial_distance + cumulative_velocity * interval

方式二：求解后直接数值计算（性能最优）

如果模型已经求解完成，完全绕开Pyomo表达式的求值逻辑，直接用Pandas计算累积和：

# 提取速度的数值结果
velocity_vals = pd.Series(model.velocity.extract_values()).sort_index()
# 计算累积速度
cumulative_velocity = velocity_vals.cumsum()
# 直接生成距离结果
distance_vals = model.initial_distance.value + cumulative_velocity * interval
# 将结果加入提取列表
list_of_series.append(pd.Series(distance_vals, name="distance_traveled"))

这种方式直接跳过Pyomo表达式的循环求值，性能提升数量级。

3. 批量处理组件减少Pandas开销 #

多次创建pd.Series()和pd.concat()会带来额外开销，改为先收集所有索引化组件的字典，再一次性转成DataFrame：

def extract_values_from_instance(instance: ConcreteModel):
    model_attributes = {
        "model_vars": instance.component_map(ctype=Var),
        "model_param": instance.component_map(ctype=Param),
        "model_expr": instance.component_map(ctype=Expression),
    }
    LOGGER.info('Extracting model values')
    indexed_data = {}
    scalars_dict = {}
    for components in model_attributes.values():
        for key, var in components.items():
            if var.is_indexed():
                indexed_data[key] = var.extract_values()
            else:
                scalars_dict[key] = var()
    # 一次性生成DataFrame
    indexed_df = pd.DataFrame(indexed_data, dtype="float64")
    return indexed_df, scalars_dict

4. 过滤不必要的组件 #

如果只需要提取部分组件，在component_map()后添加过滤逻辑，减少处理量：

# 仅保留需要提取的组件名
wanted_keys = ["distance_traveled", "velocity", "your_other_component"]
model_expr = {k:v for k,v in instance.component_map(ctype=Expression).items() if k in wanted_keys}

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