name: 网络优化器 description: 用于解决图结构上的运输、分配和流问题的网络优化技能。 allowed-tools: Bash(*) Read Write Edit Glob Grep WebFetch metadata: author: babysitter-sdk version: “1.0.0” category: 运筹学 backlog-id: SK-IE-003

网络优化器

您是网络优化器 - 专门解决网络优化问题的技能，包括最短路径、最小生成树、最大流和分配问题。

概述

本技能支持AI驱动的网络优化，包括：

最短路径算法选择（Dijkstra、Bellman-Ford、Floyd-Warshall）
最小生成树生成
最大流/最小割分析
最小成本网络流建模
分配问题求解（匈牙利算法）
网络单纯形法实现
多商品流建模

前提条件

Python 3.8+ 并安装NetworkX
用于高级问题的Google OR-Tools
图论基础知识

能力

1. 最短路径算法

import networkx as nx

def shortest_path_analysis(G, source, target):
    """
    选择并应用合适的最短路径算法
    """
    # 检查负权重
    has_negative = any(d.get('weight', 1) < 0
                       for u, v, d in G.edges(data=True))

    if not has_negative:
        # 非负权重使用Dijkstra
        path = nx.dijkstra_path(G, source, target)
        length = nx.dijkstra_path_length(G, source, target)
    else:
        # 负权重使用Bellman-Ford
        path = nx.bellman_ford_path(G, source, target)
        length = nx.bellman_ford_path_length(G, source, target)

    return {
        "path": path,
        "length": length,
        "algorithm": "dijkstra" if not has_negative else "bellman_ford"
    }

# 全对最短路径
def all_pairs_shortest_paths(G):
    # 稠密图使用Floyd-Warshall
    if G.number_of_edges() > G.number_of_nodes()**2 / 4:
        return dict(nx.floyd_warshall(G))
    else:
        # 稀疏图使用Johnson
        return dict(nx.johnson(G))

2. 最小生成树

def minimum_spanning_tree(G, algorithm='kruskal'):
    """
    生成最小生成树
    """
    if algorithm == 'kruskal':
        mst = nx.minimum_spanning_tree(G, algorithm='kruskal')
    elif algorithm == 'prim':
        mst = nx.minimum_spanning_tree(G, algorithm='prim')

    total_weight = sum(d['weight'] for u, v, d in mst.edges(data=True))

    return {
        "tree": mst,
        "total_weight": total_weight,
        "edges": list(mst.edges(data=True))
    }

3. 最大流/最小割

def max_flow_min_cut(G, source, sink):
    """
    计算最大流和最小割
    """
    # 最大流
    flow_value, flow_dict = nx.maximum_flow(G, source, sink)

    # 最小割
    cut_value, partition = nx.minimum_cut(G, source, sink)

    # 识别割边
    reachable, non_reachable = partition
    cut_edges = [(u, v) for u in reachable for v in G[u]
                 if v in non_reachable]

    return {
        "max_flow": flow_value,
        "flow_dict": flow_dict,
        "min_cut_value": cut_value,
        "cut_edges": cut_edges,
        "source_side": list(reachable),
        "sink_side": list(non_reachable)
    }

4. 最小成本流

from ortools.graph.python import min_cost_flow

def min_cost_flow_problem(nodes, arcs):
    """
    解决最小成本网络流问题
    """
    smcf = min_cost_flow.SimpleMinCostFlow()

    # 添加弧：(起点, 终点, 容量, 单位成本)
    for start, end, capacity, cost in arcs:
        smcf.add_arc_with_capacity_and_unit_cost(
            start, end, capacity, cost
        )

    # 设置供应/需求
    for node, supply in nodes.items():
        smcf.set_node_supply(node, supply)

    status = smcf.solve()

    if status == smcf.OPTIMAL:
        result = {
            "status": "optimal",
            "total_cost": smcf.optimal_cost(),
            "flows": []
        }
        for i in range(smcf.num_arcs()):
            if smcf.flow(i) > 0:
                result["flows"].append({
                    "from": smcf.tail(i),
                    "to": smcf.head(i),
                    "flow": smcf.flow(i),
                    "cost": smcf.flow(i) * smcf.unit_cost(i)
                })
        return result

    return {"status": "infeasible"}

5. 分配问题（匈牙利算法）

from scipy.optimize import linear_sum_assignment

def assignment_problem(cost_matrix):
    """
    使用匈牙利算法解决分配问题
    """
    row_ind, col_ind = linear_sum_assignment(cost_matrix)

    total_cost = cost_matrix[row_ind, col_ind].sum()

    assignments = list(zip(row_ind.tolist(), col_ind.tolist()))

    return {
        "total_cost": total_cost,
        "assignments": assignments,
        "assignment_costs": cost_matrix[row_ind, col_ind].tolist()
    }

6. 多商品流

def multi_commodity_flow(G, commodities):
    """
    多商品流问题建模
    commodities: (源点, 汇点, 需求)列表
    """
    from ortools.linear_solver import pywraplp

    solver = pywraplp.Solver.CreateSolver('GLOP')

    # 每条边上每个商品的流变量
    flows = {}
    for k, (s, t, d) in enumerate(commodities):
        for u, v in G.edges():
            flows[k, u, v] = solver.NumVar(0, G[u][v]['capacity'],
                                           f'f_{k}_{u}_{v}')

    # 流量守恒
    for k, (s, t, d) in enumerate(commodities):
        for node in G.nodes():
            inflow = sum(flows[k, u, node] for u in G.predecessors(node))
            outflow = sum(flows[k, node, v] for v in G.successors(node))

            if node == s:
                solver.Add(outflow - inflow == d)
            elif node == t:
                solver.Add(inflow - outflow == d)
            else:
                solver.Add(inflow == outflow)

    # 容量约束（共享）
    for u, v in G.edges():
        solver.Add(sum(flows[k, u, v] for k in range(len(commodities)))
                  <= G[u][v]['capacity'])

    # 最小化总成本
    solver.Minimize(sum(
        flows[k, u, v] * G[u][v].get('cost', 1)
        for k in range(len(commodities))
        for u, v in G.edges()
    ))

    solver.Solve()
    return solver

流程集成

本技能与以下流程集成：

transportation-route-optimization.js
warehouse-layout-slotting-optimization.js
capacity-planning-analysis.js

输出格式

{
  "problem_type": "max_flow",
  "status": "optimal",
  "objective": 23.0,
  "solution": {
    "flow_paths": [
      {"path": ["s", "a", "b", "t"], "flow": 10},
      {"path": ["s", "c", "t"], "flow": 13}
    ]
  },
  "analysis": {
    "bottleneck_edges": [["a", "b"], ["c", "t"]],
    "recommendations": ["增加边(a,b)的容量"]
  }
}

工具/库

库	描述	使用场景
NetworkX	图分析	通用网络
OR-Tools	最小成本流	大规模问题
igraph	快速算法	性能要求高
SciPy	分配问题	匈牙利方法

最佳实践

选择合适的算法 - 根据问题结构匹配算法
处理不可行性 - 检查不连通组件
缩放权重 - 避免数值问题
可视化网络 - 辅助调试和沟通
测试边界情况 - 空图、单节点等

约束

求解前验证网络连通性
记录所有边权重和容量
适当处理负环
清晰报告不可行情况