带边投资成本的Python最小费用流求解器适配技术咨询

Hey Justus, great question—this is a classic variation of the min-cost flow problem called the fixed-charge network flow problem, which adds that per-edge fixed investment cost you’re dealing with. Neither NetworkX nor OR-Tools have built-in support for this out of the box, but we can adapt the standard min-cost flow model to handle it by splitting edges to model the fixed cost as a binary "enable this edge" decision. Let’s walk through how to do this.

Core Idea: Edge Splitting for Fixed Costs #

For every original edge (u, v) with:

• Fixed investment cost F (paid once if the edge is used at all)
• Per-unit flow cost c
• Maximum possible flow (we can set this to total demand, since you can’t move more than the total needed), we’ll split it into three parts using an auxiliary node k:

1. Enablement Edge: Add u → k with capacity 1, cost F. This represents paying the fixed cost to "unlock" the edge for use.
2. Flow Edge: Add k → v with capacity equal to total demand (or your upper limit), cost c. This handles the per-unit flow cost once the edge is enabled.
3. Block Direct Flow: We don’t add a direct u → v edge—this forces any flow from u to v to go through the enablement edge first, ensuring we only pay the fixed cost if the edge is actually used.

Implementation with OR-Tools #

OR-Tools’ min_cost_flow solver is flexible enough to handle this adapted model. Here’s a complete example tailored to your problem (suppliers and demand points in a complete graph):

from ortools.graph import pywrapgraph

def fixed_charge_min_cost_flow(suppliers, demands, edge_costs):
    # suppliers: dict {node_id: supply_amount}
    # demands: dict {node_id: demand_amount}
    # edge_costs: dict {(u, v): (fixed_cost, unit_cost)}
    
    min_cost_flow = pywrapgraph.SimpleMinCostFlow()
    node_id_map = {}
    next_node_id = 0
    
    # Assign unique IDs to original nodes
    all_nodes = set(suppliers.keys()).union(set(demands.keys()))
    for node in all_nodes:
        node_id_map[node] = next_node_id
        next_node_id += 1
    
    # Add auxiliary nodes for each edge and build the adapted graph
    for (u, v), (fixed_cost, unit_cost) in edge_costs.items():
        u_id = node_id_map[u]
        v_id = node_id_map[v]
        
        # Create auxiliary node for this edge
        aux_node_id = next_node_id
        next_node_id += 1
        
        # Add enablement edge (u -> aux): capacity 1, cost = fixed_cost
        min_cost_flow.AddArcWithCapacityAndUnitCost(u_id, aux_node_id, 1, fixed_cost)
        
        # Add flow edge (aux -> v): capacity = total demand, cost = unit_cost
        total_demand = sum(demands.values())
        min_cost_flow.AddArcWithCapacityAndUnitCost(aux_node_id, v_id, total_demand, unit_cost)
    
    # Set supplies and demands
    for node, supply in suppliers.items():
        min_cost_flow.SetNodeSupply(node_id_map[node], supply)
    for node, demand in demands.items():
        min_cost_flow.SetNodeSupply(node_id_map[node], -demand)
    
    # Solve the problem
    status = min_cost_flow.Solve()
    
    if status == min_cost_flow.OPTIMAL:
        print(f"Total cost: {min_cost_flow.OptimalCost()}")
        print("\nUsed edges and flows:")
        # Track which original edges are enabled
        enabled_edges = {}
        for arc in range(min_cost_flow.NumArcs()):
            # Check if this is an enablement edge (capacity 1)
            if min_cost_flow.Capacity(arc) == 1 and min_cost_flow.Flow(arc) == 1:
                u_orig = [k for k, v in node_id_map.items() if v == min_cost_flow.Tail(arc)][0]
                # Find the corresponding flow edge from aux node to v
                aux_node = min_cost_flow.Head(arc)
                for flow_arc in range(min_cost_flow.NumArcs()):
                    if min_cost_flow.Tail(flow_arc) == aux_node:
                        v_orig = [k for k, v in node_id_map.items() if v == min_cost_flow.Head(flow_arc)][0]
                        flow_amount = min_cost_flow.Flow(flow_arc)
                        enabled_edges[(u_orig, v_orig)] = flow_amount
                        print(f"Edge {u_orig} -> {v_orig}: Flow = {flow_amount}, Total cost = {fixed_cost + unit_cost * flow_amount}")
                        break
        return enabled_edges
    else:
        print(f"Solver failed with status: {status}")
        return None

# Example usage
if __name__ == "__main__":
    # Define suppliers (node S1, S2 with supply)
    suppliers = {"S1": 10, "S2": 15}
    # Define demand points (node D1, D2, D3 with demand)
    demands = {"D1": 8, "D2": 12, "D3": 5}
    # Define edge costs: (fixed cost, unit cost) for every possible edge in complete graph
    edge_costs = {
        ("S1", "D1"): (50, 2),
        ("S1", "D2"): (30, 3),
        ("S1", "D3"): (40, 1),
        ("S2", "D1"): (20, 4),
        ("S2", "D2"): (60, 1),
        ("S2", "D3"): (35, 2),
    }
    
    result = fixed_charge_min_cost_flow(suppliers, demands, edge_costs)

Key Details to Note #

• Auxiliary Nodes: Each original edge gets its own auxiliary node to separate the fixed cost decision from the flow cost.
• Enablement Edge Capacity: Setting capacity to 1 ensures we only pay the fixed cost once, regardless of how much flow we send through the edge.
• Flow Edge Capacity: We set this to total demand because you can’t send more flow than the total amount needed across the network—adjust this if you have per-edge flow limits.
• Result Parsing: We check for enablement edges with flow=1 to identify which original edges were activated, then map back to the flow amount through the corresponding flow edge.

Alternative with NetworkX #

If you prefer NetworkX, you can use the same edge-splitting approach with networkx.min_cost_flow. The main difference is manually managing node IDs and building the adapted graph, then calling the solver. The logic for splitting edges remains identical.

内容的提问来源于stack exchange，提问作者Justus von Rhein