简述随机游走模型

一维随机游走问题:设一个质点(随机游走者)沿着一条直线运动,单位时间内只能运动一个单位长度,且只能停留在该直线上的整数点,假设在时刻t,该质点位于直线上的点i,那么在时刻t+1,该质点的位置有三种可能:

①以p的概率跳到整数点i-1

②或以q的概率跳到点i+1

③或以r=1-p-q的概率继续停留在点i

由于每一步的结果都是独立的,且每种情况发生的概率之和都为1,则该过程服从伯努利分布,称为贝努利随机游走过程。当p=q=0.5时,即质点在下一时刻到达其相邻点的概率是相等的,称为简单的随机游走。

基于随机游走的图像分割算法

随机游走算法是一种基于图论的分割算法,属于一种交互式的图像分割。它的分割思想是,以图像的像素为图的顶点,相邻像素之间的四邻域或八邻域关系为图的边,并根据像素属性及相邻像素之间特征的相似性定义图中各边的权值,以此构建网络图,然后由通过用户手工指定前景和背景标记,即前景物体和背景物体的种子像素,以边上的权重为转移概率,未标记像素节点为初始点,计算每个未标记节点首次到达各种子像素的概率,根据概率大小,划分未标记节点,得到最终分割结果。

以下内容来自参考文献:Grady L .Random Walks for Image Segmentation[J].IEEE Transactions on Pattern Analysis & Machine Intelligence, 2006, 28:1768-1783.DOI:10.1109/TPAMI.2006.233.

L1,L2,L3三个种子点分别由用户交互输入,作为标记的种子点。现要把图像分割成对应的三部分。

计算图中任意一点vi与其各个邻接顶点连接边的权重

对于图中任意一点vi的概率,其满足随机游走概率公式

加边界约束条件:以已标记的K类顶点作为边界约束条件,求解各未知点游走到L1的概率,则以PL1=1,PL2=0,PL3=0,作为约束条件,可求得个未知点的概率,如下图所示:

同理,到达L2的概率为

到达L3的概率为

每一个未标记点,根据获得的对K类标记的隶属度值进行判断,若未标记点到达第k类的概率最大,则将未标记节点vi判别为属于类别k,完成分割。

python代码

import numpy as npimport networkx as nximport randomclass Graph():def __init__(self, nx_G, is_directed, p, q):self.G = nx_Gself.is_directed = is_directedself.p = pself.q = qdef node2vec_walk(self, walk_length, start_node):'''Simulate a random walk starting from start node.'''G = self.Galias_nodes = self.alias_nodesalias_edges = self.alias_edgeswalk = [start_node]while len(walk)  0:# 如果序列中仅有一个结点,即第一次游走# alias_nodes中保存了alias_setup的[alias, accept],通过alias_draw返回采样的下一个索引号if len(walk) == 1:walk.append(cur_nbrs[alias_draw(alias_nodes[cur][0], alias_nodes[cur][1])])else:# 当前游走结点的前一个结点和下一个节点prev = walk[-2]# 使用alias_edges中记录的[alias, accept],来采样邻居中的下一个节点next = cur_nbrs[alias_draw(alias_edges[(prev, cur)][0], alias_edges[(prev, cur)][1])]walk.append(next)else:breakreturn walkdef simulate_walks(self, num_walks, walk_length):'''Repeatedly simulate random walks from each node.'''G = self.Gwalks = []nodes = list(G.nodes())# nodes采样一次为一个epoch,此处就是num_walks个epochprint('Walk iteration:')for walk_iter in range(num_walks):print(str(walk_iter+1), '/', str(num_walks))random.shuffle(nodes)for node in nodes:walks.append(self.node2vec_walk(walk_length=walk_length, start_node=node))return walksdef get_alias_edge(self, src, dst):'''Get the alias edge setup lists for a given edge.:return alias_setup(): 在上一次访问顶点 t ,当前访问顶点为 v 时到下一个顶点 x 的未归一化转移概率。:param src:随机游走序列种的上一个结点:param dst:当前结点参数p控制重复访问刚刚访问过的顶点的概率。若p较大,则访问刚刚访问过的顶点的概率会变低。参数q控制着游走是向外还是向内:若q>1,随机游走倾向于访问和上一次的t接近的顶点(偏向BFS);若q<1,倾向于访问远离t的顶点(偏向DFS)'''G = self.Gp = self.pq = self.qunnormalized_probs = []for dst_nbr in sorted(G.neighbors(dst)):if dst_nbr == src:unnormalized_probs.append(G[dst][dst_nbr]['weight']/p)elif G.has_edge(dst_nbr, src):unnormalized_probs.append(G[dst][dst_nbr]['weight'])else:unnormalized_probs.append(G[dst][dst_nbr]['weight']/q)norm_const = sum(unnormalized_probs)normalized_probs =[float(u_prob)/norm_const for u_prob in unnormalized_probs]return alias_setup(normalized_probs)def preprocess_transition_probs(self):'''Preprocessing of transition probabilities for guiding the random walks.用于引导随机游走的预处理,得到马尔可夫转移概率矩阵。'''G = self.Gis_directed = self.is_directedalias_nodes = {}# G.neighbors(node) 与顶点相邻的所有顶点,更方便更快的访问adjacency字典用: G[cur]for node in G.nodes():# 根据邻居节点的权重,计算转移概率unnormalized_probs = [G[node][nbr]['weight'] for nbr in sorted(G.neighbors(node))]norm_const = sum(unnormalized_probs)# 计算当前节点到邻居节点的转移概率,其实就是权重归一化normalized_probs =[float(u_prob)/norm_const for u_prob in unnormalized_probs]# 设置alias table,保存每个节点的accept[i]和alias[i],为后面alias采样做准备。alias_nodes[node] = alias_setup(normalized_probs)alias_edges = {}triads = {}# 保存每条边的accept[i]和alias[i]if is_directed:for edge in G.edges():alias_edges[edge] = self.get_alias_edge(edge[0], edge[1])else:for edge in G.edges():alias_edges[edge] = self.get_alias_edge(edge[0], edge[1])alias_edges[(edge[1], edge[0])] = self.get_alias_edge(edge[1], edge[0])self.alias_nodes = alias_nodesself.alias_edges = alias_edgesreturndef alias_setup(probs):'''Compute utility lists for non-uniform sampling from discrete distributions.Refer to https://hips.seas.harvard.edu/blog/2013/03/03/the-alias-method-efficient-sampling-with-many-discrete-outcomes/for details:param probs: 指定的采样结果概率分布列表。期望按这个概率列表来采样每个随机变量X。:return J: alias[i]表示第i列中不是事件i的另一个事件的编号。:return p: accept[i]表示事件i占第i列矩形的面积的比例。'''K = len(probs)# q表示:accept数组q = np.zeros(K)# J表示:alias数组J = np.zeros(K, dtype=np.int)# Alias方法将整个概率分布压成一个 1*N 的矩形,每个事件转换为矩形中的面积。# 将面积大于1的事件多出的面积补充到面积小于1对应的事件中,以确保每一个小方格的面积为1,# 同时,保证每一方格至多存储两个事件。smaller = [] # 面积小于1的事件larger = []# 面积大于1的事件for kk, prob in enumerate(probs):q[kk] = K*probif q[kk]  0 and len(larger) > 0:small = smaller.pop()large = larger.pop()J[small] = large# 其实是 q[large] - (1.0 - q[small]),把大的削去(1.0 - q[small])填充到小的上q[large] = q[large] + q[small] - 1.0# 大的剩下的面积,放到下一轮继续倒腾if q[large] < 1.0:smaller.append(large)else:larger.append(large)return J, qdef alias_draw(J, q):'''Draw sample from a non-uniform discrete distribution using alias sampling.参考:https://zhuanlan.zhihu.com/p/54867139:param q: accept数组,表示事件i占第i列矩形的面积的比例;:param J: alias数组,表示alias矩形的第i列中不是事件i的另一个事件的编号,也就是填充的那一列的序号;生成一个随机数 kk in [0, K],另一个随机数 x in [0,1],如果 x < accept[kk],表示接受事件kk,返回kk,否则拒绝事件kk,返回alias[kk]'''K = len(J)kk = int(np.floor(np.random.rand()*K))if np.random.rand() < q[kk]:return kkelse:return J[kk]