GraphSAGE 代碼解析(一) - unsupervised_train.py

原創文章～轉載請注明出處哦。其他部分內容參見以下鏈接～

GraphSAGE 代碼解析(二) - layers.py

GraphSAGE 代碼解析(三) - aggregators.py

GraphSAGE 代碼解析(四) - models.py

GraphSAGE代碼詳解

example_data:

1. toy-ppi-G.json 圖的信息

{ 
  directed: false
  graph : {
　　　　　         {name: disjoint_union(,) }
　　　　　　　　　　　nodes:  [
　　　　　　　　　　              {　 
                                test: false
　　　　　　　　　　　              id: 0
　　　　　　　　　　　              features: [ ... ]
　　　　　　　　　　　              val: false
　　　　　　　　　　                lable: [ ... ]
　　　　　　　　　　　　           }
　　　　　　　　　　　　           {...}
　　　　　　　　　　               ...
　　　　　　　　　　　　      ]

　　　　　　　　　　　 links: [
　　　　　　　　　　　　           {　 
                                test_removed: false
　　　　　　　　　　　　            train_removed: false
　　　　　　　　　　　　            target: 800 # 指向的節點id（默認從小節點指向大節點）
　　　　　　　　　　　　            source: 0   # 從0節點按順序展示
　　　　　　　　　　               }
　　　　　　　　　　               {...}
　　　       　　　　　            ...
　　　　　　　　　　          ]
　　　　　　}
}

2. toy-ppi-class_map.json

3. toy-ppi-feats.npy 預訓練好得到的features

4. toy-ppi-id_map.json 節點編號與序號的一一對應；數據格式為：{"0": 0, "1": 1,..., "14754": 14754}

5. toy-ppi-walks.txt 

    從一點出發隨機游走到鄰居節點的情況，對於每個點取198次（即可能有重復情況）

    例如：0    708 表示從0點走到708點。

1. __init__.py

from __future__ import print_function  
#即使在python2.X，使用print就得像python3.X那樣加括號使用。

from __future__ import division          
# 導入python未來支持的語言特征division(精確除法)，
# 當我們沒有在程序中導入該特征時，"/"操作符執行的是截斷除法(Truncating Division)；
# 當我們導入精確除法之后，"/"執行的是精確除法, "//"執行截斷除除法

2. unsupervised_train.py

if __name__ == '__main__':
　　tf.app.run()
# https://blog.csdn.net/fxjzzyo/article/details/80466321
# tf.app.run()的作用：通過處理flag解析，然后執行main函數
# 如果你的代碼中的入口函數不叫main()，而是一個其他名字的函數，如test()，則你應該這樣寫入口tf.app.run(test())
# 如果你的代碼中的入口函數叫main()，則你就可以把入口寫成tf.app.run()

def main(argv=None):
　　print("Loading training data..")
　　train_data = load_data(FLAGS.train_prefix, load_walks=True)
　　# load_data函數在graphsage.utils中定義

　　print("Done loading training data..")
　　train(train_data)
　　# train函數在該文件中定義def train(train_data, test_data=None)

3. utils.py - func: load_data

(1) 讀入id_map, class_map

if isinstance(G.nodes()[0], int):
        def conversion(n): return int(n)
    else:
        def conversion(n): return n

a. isinstance() 函數來判斷一個對象是否是一個已知的類型，類似 type()。 

isinstance(object, classinfo)

參數
object -- 實例對象。
classinfo -- 可以是直接或間接類名、基本類型或者由它們組成的元組。

返回值
如果對象的類型與參數二的類型（classinfo）相同則返回 True，否則返回 False。

>>>a = 2
>>> isinstance (a,int)
True
>>> isinstance (a,str)
False
>>> isinstance (a,(str,int,list))    # 是元組中的一個返回 True
True

type() 與 isinstance() 區別:

type() 不會認為子類是一種父類類型，不考慮繼承關系。
isinstance() 會認為子類是一種父類類型，考慮繼承關系。
如果要判斷兩個類型是否相同推薦使用 isinstance()。

class A:
    pass
 
class B(A):
    pass
 
isinstance(A(), A)    # returns True
type(A()) == A        # returns True
isinstance(B(), A)    # returns True
type(B()) == A        # returns False

b. G.nodes()

返回的是圖中節點n與節點屬性nodedata。https://networkx.github.io/documentation/stable/reference/classes/generated/networkx.Graph.nodes.html

例子：

>>> G = nx.path_graph(3)
>>> list(G.nodes)
[0, 1, 2]
>>> list(G)
[0, 1, 2]

獲取nodedata:

>>> G.add_node(1, time='5pm')
>>> G.nodes[0]['foo'] = 'bar'
>>> list(G.nodes(data=True))
[(0, {'foo': 'bar'}), (1, {'time': '5pm'}), (2, {})]
>>> list(G.nodes.data())
[(0, {'foo': 'bar'}), (1, {'time': '5pm'}), (2, {})]

>>> list(G.nodes(data='foo'))
[(0, 'bar'), (1, None), (2, None)]

>>> list(G.nodes(data='time'))
[(0, None), (1, '5pm'), (2, None)]

>>> list(G.nodes(data='time', default='Not Available'))
[(0, 'Not Available'), (1, '5pm'), (2, 'Not Available')]

If some of your nodes have an attribute and the rest are assumed to have a default attribute value you can create a dictionary from node/attribute pairs using the default keyword argument to guarantee the value is never None:

>>> G = nx.Graph()
>>> G.add_node(0)
>>> G.add_node(1, weight=2)
>>> G.add_node(2, weight=3)
>>> dict(G.nodes(data='weight', default=1))
{0: 1, 1: 2, 2: 3}

----------------------------

在utils.py中，判斷G.nodes()[0] 是否為int型（即不帶nodedata）。

若為int型，則將n轉為int型；否則直接返回n.

b. conversion() 函數

id_map = json.load(open(prefix + "-id_map.json"))
id_map = {conversion(k): int(v) for k, v in id_map.items()}

前面定義的conversion()函數在id_map這里用到了,把外存中的文件內容讀到內存中，用dict類型的id_map存儲。

id_map.json文件中數據格式為：{"0": 0, "1": 1,..., "14754": 14754}，也即id_map的迭代中k為str類型，v為int型。數據文件中G.nodes()[0] 顯然是帶nodedata的，也就算一般采用 def conversion(n): return n，返回的n為類型的（就是前面形參k的類型）；

但是為什么當G.nodes()[0] 不帶nodedata時，要返回int(n)？

c. class_map:  {"0": [.0,1,..], "1": [.0,1,..]...} ?含義？

list(class_map.values())： [ [...], [...], ... ,[...] ]

list(class_map.values())[0]: 表示取第一個[...] =>含義？ 

if isinstance(list(class_map.values())[0], list):
    def lab_conversion(n): return n else: def lab_conversion(n): return int(n)

(2) Remove node

# Remove all nodes that do not have val/test annotations
    # (necessary because of networkx weirdness with the Reddit data)
    broken_count = 0
    for node in G.nodes():
        if not 'val' in G.node[node] or not 'test' in G.node[node]:
            G.remove_node(node)
            broken_count += 1

這里刪除的節點是不具有'val'，'test'屬性 的節點，而不是'val'，'test' 屬性值為None的節點。

區分開 if not 'val' in G.node[node] 和 if not G.node[n]['val']的不同意義。

broken_count  記錄刪去的沒有val 或者 test的屬性的節點的數目。

e. G.edges()

for edge in G.edges():
        if (G.node[edge[0]]['val'] or G.node[edge[1]]['val'] or
                G.node[edge[0]]['test'] or G.node[edge[1]]['test']):
            G[edge[0]][edge[1]]['train_removed'] = True
        else:
            G[edge[0]][edge[1]]['train_removed'] = False

G.edges() 得到edge_list, [( , ), ( , ), ... ( , )].list中每一個元素是所表示邊的兩個節點信息。若設置data = True，則會顯示邊的權重等屬性信息。

>>> G = nx.Graph()   # or DiGraph, MultiGraph, MultiDiGraph, etc
>>> G.add_path([0,1,2])
>>> G.add_edge(2,3,weight=5)
>>> G.edges()
[(0, 1), (1, 2), (2, 3)]
>>> G.edges(data=True) # default edge data is {} (empty dictionary)
[(0, 1, {}), (1, 2, {}), (2, 3, {'weight': 5})]
>>> list(G.edges_iter(data='weight', default=1))
[(0, 1, 1), (1, 2, 1), (2, 3, 5)]
>>> G.edges([0,3])
[(0, 1), (3, 2)]
>>> G.edges(0)
[(0, 1)]

代碼中edge對edges迭代，每次去list中的一個元組，而edge[0], edge[1]則分別表示兩個頂點。

若兩個頂點中至少有一個的val/test不為空，則將該邊的'train_removed'設為True，否則為False.

該操作為保證'train_removed'不為空。

(3) 獲取訓練數據features並標准化

if normalize and not feats is None:
        from sklearn.preprocessing import StandardScaler
        train_ids = np.array([id_map[n] for n in G.nodes(
        ) if not G.node[n]['val'] and not G.node[n]['test']])
        train_feats = feats[train_ids]
        scaler = StandardScaler()
        scaler.fit(train_feats)
        feats = scaler.transform(feats)

這里if not feats is None 等價於 if feats is not None.

將val,test均為None的node選為訓練數據，通過id_map獲取其在feature表中的索引值，添加到train_ids數組中。根據索引train_ids，train_fests獲取這些nodes的features.

StandardScaler的用法：

http://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.StandardScaler.html

Methods:

fit(X[, y]) : Compute the mean and std to be used for later scaling.

transform(X[, y, copy]) : Perform standardization by centering and scaling

fit_transform(X[, y]) : Fit to data, then transform it.

例子：

>>> from sklearn.preprocessing import StandardScaler
>>> data = [[0, 0], [0, 0], [1, 1], [1, 1]]
>>> scaler = StandardScaler()
>>> print(scaler.fit(data))
StandardScaler(copy=True, with_mean=True, with_std=True)
>>> print(scaler.mean_)
[0.5 0.5]
>>> print(scaler.transform(data))
[[-1. -1.]
 [-1. -1.]
 [ 1.  1.]
 [ 1.  1.]]
>>> print(scaler.transform([[2, 2]]))
[[3. 3.]]

# 計算得
# 均值[0.5, 0.5], 
# 方差：1/4 * [(0 - 0.5)^2 * 2 + (1 - 0.5)^2 * 2] = 1/4 = 0.25
# 標准差：0.5
# 對於[2,2] transform 標准化之后: (2 - 0.5) / 0.5 = 3

(4) Load walks

在unsupervised_train.py的main函數中：

train_data = load_data(FLAGS.train_prefix, load_walks=True)

load_walks = True，需要執行utils.py中的load_walks操作。

if load_walks:  # false by default
        with open(prefix + "-walks.txt") as fp:
            for line in fp:
                walks.append(map(conversion, line.split()))

map() 的用法：http://www.runoob.com/python/python-func-map.html

map(function, iterable, ...)

map() 會根據提供的函數對指定序列做映射。

第一個參數 function 以參數序列中的每一個元素調用 function 函數，返回包含每次 function 函數返回值的新列表。

例子：

>>>def square(x) :            # 計算平方數
...     return x ** 2
... 
>>> map(square, [1,2,3,4,5])   # 計算列表各個元素的平方
[1, 4, 9, 16, 25]
>>> map(lambda x: x ** 2, [1, 2, 3, 4, 5])  # 使用 lambda 匿名函數
[1, 4, 9, 16, 25]
 
# 提供了兩個列表，對相同位置的列表數據進行相加
>>> map(lambda x, y: x + y, [1, 3, 5, 7, 9], [2, 4, 6, 8, 10])
[3, 7, 11, 15, 19]

walks初始化為[]， 之后append的是游走的節點對的對象。

例子：walks.txt:

0    708
0    3163
0    276

def conversion(n): return n
walks = []
with open("walks.txt") as fp:
    for line in fp:
        print(line.split())
        walks.append(map(conversion, line.split()))
print(walks) 
print(len(walks))

輸出：

['0', '708']
['0', '3163']
['0', '276']
[<map object at 0x7f5bc0d68da0>, <map object at 0x7f5bc0d68e48>, <map object at 0x7f5bc0d68f28>]
3

(5) 函數返回值

return G, feats, id_map, walks, class_map

------------------------------------------------------------------------------------

4. unsupervised_train.py - func: train(train_data)

def train(train_data, test_data=None):

這里的train_data是上文所述的load_data函數的返回值。

變量含義：

G = train_data[0]    # 圖
features = train_data[1]    # 訓練數據的features
id_map = train_data[2]     # "n" : n
context_pairs = train_data[3] if FLAGS.random_context else None #random walk的點對

if not features is None:
    # pad with dummy zero vector
    features = np.vstack([features, np.zeros((features.shape[1],))])

這里vstack為features添加列一行0向量，用於WX + b中與b相加。

placeholders = construct_placeholders()
# def construct_placeholders()定義的placeholders包含：
# batch1, batch2, neg_samples, dropout, batch_size

minibatch是EdgeMinibatchIterator的一個實例，轉至minibatch.py看class EdgeMinibatchIterator(object)的定義。

5. minibatch.py - class EdgeMinibatchIterator

https://www.cnblogs.com/shiyublog/p/9902423.html

6. unsupervised_train.py - func train

繼續回來看unsupervised_trian.py 中的train函數

變量：

adj_info_ph = tf.placeholder(tf.int32, shape=minibatch.adj.shape)
adj_info = tf.Variable(adj_info_ph, trainable=False, name="adj_info")

adj_info記錄鄰居信息，是一個矩陣，矩陣每一行對應每一個節點的鄰居節點編號數組。

（1）選擇模型

接下來根據輸入參數判斷選擇6種模型（graphsage_mean，gcn，graphsage_seq，graphsage_maxpool，graphsage_meanpool，n2v）中的哪一種。

以graphsage開頭的幾種是graphsage的幾種變體，由於aggregator不同而不同。可以通過設定SampleAndAggregate()中的aggregator_type進行選擇。默認為mean.

其中gcn與graphsage的參數不同在於：

gcn的aggregator中進行列concat的操作，因此其維數是graphsage的二倍。

a. graphsage_maxpool 

sampler = UniformNeighborSampler(adj_info)

首先看UniformNeighborSampler，該類用於sample節點的鄰居，在neigh_samplers.py中。

neigh_samplers.py

class UniformNeighborSampler(Layer):
    """
    Uniformly samples neighbors.
    Assumes that adj lists are padded with random re-sampling
    """
    def __init__(self, adj_info, **kwargs):
        super(UniformNeighborSampler, self).__init__(**kwargs)
        self.adj_info = adj_info

    def _call(self, inputs):
        ids, num_samples = inputs
        adj_lists = tf.nn.embedding_lookup(self.adj_info, ids) 
        adj_lists = tf.transpose(tf.random_shuffle(tf.transpose(adj_lists)))
        adj_lists = tf.slice(adj_lists, [0,0], [-1, num_samples])
        return adj_lists

1.  tf.nn.embedding_lookup 用於根據ids在adj_info中找到各個對應位的向量。

2. adj_lists = tf.transpose(tf.random_shuffle(tf.transpose(adj_lists)))

    adj_lists = tf.slice(adj_lists, [0,0], [-1, num_samples]) 的過程見下：

id0 id1 id2...   --transpose--> id0 [...]  --shuffle--> id1 [...]  --transpose--> id1 id2 id0 --slice--> id1 id2

[]    []    []                                id1 [...]                     id2 [...]                         []     []    []                  []    []

                                              id2 [...]                     id0 [...]

均勻：shuffle打亂0維的順序，即打亂行順序，以此使下面采樣可以“均勻”。為了使用shuffle函數，需要在shuffle前后transpose一下。

采樣：slice之后，相當於隨機挑選了num_samples個樣本，並保留了這些樣本的全部屬性特征。

3. 最后的adj_lists即為均勻采樣后的表示鄰居信息的矩陣。

---------------------------------------------------

回到unsupervised_train.py 的train()函數.

sampler = UniformNeighborSampler(adj_info)

sampler獲取均勻采樣后的鄰居節點信息。

---------------------------------------------------

layer_infos = [SAGEInfo("node", sampler, FLAGS.samples_1, FLAGS.dim_1),
               SAGEInfo("node", sampler, FLAGS.samples_2, FLAGS.dim_2)]

其中SAGEInfo在models.py中。

models.py 

https://www.cnblogs.com/shiyublog/p/9879875.html

# SAGEInfo is a namedtuple that specifies the parameters 
# of the recursive GraphSAGE layers
SAGEInfo = namedtuple("SAGEInfo",
    ['layer_name', # name of the layer (to get feature embedding etc.)
     'neigh_sampler', # callable neigh_sampler constructor
     'num_samples',
     'output_dim' # the output (i.e., hidden) dimension
    ])

namedtuple 命名元組，可以給tuple命名，用法見下：

https://www.cnblogs.com/chenlin163/p/7259061.html

import collections

MyTupleClass = collections.namedtuple('MyTupleClass',['name', 'age', 'job'])
obj = MyTupleClass("Tomsom",12,'Cooker')
print(obj.name)
print(obj.age)
print(obj.job)

# Output:
# Tomsom
# 12
# Cooker
#############################

Person=collections.namedtuple('Person','name age gender') 
# 以空格分開，表示這個namedtuple有三個元素

print( 'Type of Person:',type(Person))
Bob=Person(name='Bob',age=30,gender='male')
print( 'Representation:',Bob)
Jane=Person(name='Jane',age=29,gender='female')
print( 'Field by Name:',Jane.name)
for people in [Bob,Jane]:
    print ("%s is %d years old %s" % people)

# Output:
# Type of Person: <class 'type'>
# Representation: Person(name='Bob', age=30, gender='male')
# Field by Name: Jane
# Bob is 30 years old male
# Jane is 29 years old female
#############################

# 在使用namedtyuple的時候要注意其中的名稱不能使用Python的關鍵字，如class def等
# 不能有重復的元素名稱，比如：不能有兩個’age age’。如果出現這些情況，程序會報錯。
# 但是，在實際使用的時候可能無法避免這種情況，
# 比如:可能我們的元素名稱是從數據庫里讀出來的記錄，這樣很難保證一定不會出現Python關鍵字。
# 這種情況下的解決辦法是將namedtuple的重命名模式打開，
# 這樣如果遇到Python關鍵字或者有重復元素名 時，自動進行重命名。

with_class=collections.namedtuple('Person','name age class gender',rename=True)
print with_class._fields
two_ages=collections.namedtuple('Person','name age gender age',rename=True)
print two_ages._fields

# Output:
# ('name', 'age', '_2', 'gender')
# ('name', 'age', 'gender', '_3')

# 使用rename=True的方式打開重命名選項。
# 可以看到第一個集合中的class被重命名為 ‘_2' ； 
# 第二個集合中重復的age被重命名為 ‘_3'
# namedtuple在重命名的時候使用了下划線 _ 加元素所在索引數的方式進行重命名
##############################

# 附兩段官方文檔代碼實例:
# 1) namedtuple基本用法
>>> # Basic example
>>> Point = namedtuple('Point', ['x', 'y'])
>>> p = Point(11, y=22) # instantiate with positional or keyword arguments
>>> p[0] + p[1] # indexable like the plain tuple (11, 22)
33
>>> x, y = p # unpack like a regular tuple
>>> x, y
(11, 22)
>>> p.x + p.y # fields also accessible by name
33
>>> p # readable __repr__ with a name=value style
Point(x=11, y=22)

# 2) namedtuple結合csv和sqlite用法
EmployeeRecord = namedtuple('EmployeeRecord', 'name, age, title, department, paygrade')
import csv
for emp in map(EmployeeRecord._make, csv.reader(open("employees.csv", "rb"))):
print(emp.name, emp.title)

import sqlite3
conn = sqlite3.connect('/companydata')
cursor = conn.cursor()
cursor.execute('SELECT name, age, title, department, paygrade FROM employees')
for emp in map(EmployeeRecord._make, cursor.fetchall()):
print(emp.name, emp.title)

對於FLAGS.dim_1與FLAGS.dim_2，定義為：

flags.DEFINE_integer(
    'dim_1', 128, 'Size of output dim (final is 2x this, if using concat)')
flags.DEFINE_integer(
    'dim_2', 128, 'Size of output dim (final is 2x this, if using concat)')

若GCN，因為有concat操作，故使用2x.

對於FLAGS.samples_1與FLAGS.samples_2，定義為：

1 flags.DEFINE_integer('samples_1', 25, 'number of samples in layer 1')
2 flags.DEFINE_integer('samples_2', 10, 'number of users samples in layer 2')

對應論文中的K = 1 ，第一層S1 = 25； K = 2 ，第二層S2 = 10。

----------------------------------------------------------

model = SampleAndAggregate(placeholders,
                           features,
                           adj_info,
                           minibatch.deg,
                           layer_infos=layer_infos,
                           aggregator_type="maxpool",
                           model_size=FLAGS.model_size,
                           identity_dim=FLAGS.identity_dim,
                           logging=True)

SampleAndAggregate在models.py中。

class SampleAndAggregate(GeneralizedModel)主要包含的函數有：

1. def __init__(self, placeholders, features, adj, degrees, layer_infos, concat=True, aggregator_type="mean",  model_size="small", identity_dim=0, **kwargs)

2. def sample(self, inputs, layer_infos, batch_size=None)

3. def aggregate(self, samples, input_features, dims, num_samples, support_sizes, batch_size=None,
aggregators=None, name=None, concat=False, model_size="small")

4. def _build(self)

5. def build(self)

6. def _loss(self)

7. def _accuracy(self)

---------------------------------------------------------------

(2) Session

Config

config = tf.ConfigProto(log_device_placement=FLAGS.log_device_placement)
# 參數初始化為False:
# tf.app.flags.DEFINE_boolean('log_device_placement', False,
#                     """Whether to log device placement.""")

config.gpu_options.allow_growth = True
# 控制GPU資源使用率
# 使用allow_growth option，剛一開始分配少量的GPU容量，然后按需慢慢的增加，
# 由於不會釋放內存，所以會導致碎片

#config.gpu_options.per_process_gpu_memory_fraction = GPU_MEM_FRACTION
# 設置每個GPU應該拿出多少容量給進程使用，
# per_process_gpu_memory_fraction =0.4代表 40%

config.allow_soft_placement = True
# 自動選擇運行設備
# 在tf中，通過命令 "with tf.device('/cpu:0'):",允許手動設置操作運行的設備。
# 如果手動設置的設備不存在或者不可用，就會導致tf程序等待或異常，
# 為了防止這種情況，可以設置tf.ConfigProto()中參數allow_soft_placement=True，
# 允許tf自動選擇一個存在並且可用的設備來運行操作。

 Initialize session

# Initialize session
sess = tf.Session(config=config)
merged = tf.summary.merge_all()
# tf.summary()能夠保存訓練過程以及參數分布圖並在tensorboard顯示。
# merge_all 可以將所有summary全部保存到磁盤，以便tensorboard顯示。
# 如果沒有特殊要求，一般用這一句就可一顯示訓練時的各種信息了

summary_writer = tf.summary.FileWriter(log_dir(), sess.graph)
# 指定一個文件用來保存圖。
# 格式：tf.summary.FileWritter(path,sess.graph)
# 可以調用其add_summary（）方法將訓練過程數據保存在filewriter指定的文件中

Init variables

sess.run(tf.global_variables_initializer(),
     feed_dict={adj_info_ph: minibatch.adj})

---------------------------------------------------------

(4) Train model

feed_dict = minibatch.next_minibatch_feed_dict()

next_minibatch_feed_dict() 在minibatch.py的class EdgeMinibatchIterator(object)中定義。

def next_minibatch_feed_dict(self):
    start_idx = self.batch_num * self.batch_size
    self.batch_num += 1
    end_idx = min(start_idx + self.batch_size, len(self.train_edges))
    batch_edges = self.train_edges[start_idx: end_idx]
    return self.batch_feed_dict(batch_edges)

函數中獲取下個edgeminibatch的起始與終止序號，將batch后的邊的信息傳給batch_feed_dict(self, batch_edges)函數，更新placeholders中的batch1, batch2, batch_size信息。

def batch_feed_dict(self, batch_edges):
    batch1 = []
    batch2 = []
    for node1, node2 in batch_edges:
        batch1.append(self.id2idx[node1])
        batch2.append(self.id2idx[node2])

    feed_dict = dict()
    feed_dict.update({self.placeholders['batch_size']: len(batch_edges)})
    feed_dict.update({self.placeholders['batch1']: batch1})
    feed_dict.update({self.placeholders['batch2']: batch2})

    return feed_dict

也即next_minibatch_feed_dict()返回的是下一個edge minibatch的placeholders信息。

=======================================

                  

感謝您的打賞！

（夢想還是要有的，萬一您喜歡我的文章呢）