PageRank 算法是一種經(jīng)典的網(wǎng)頁排名算法。基本思想是，每個(gè)節(jié)點(diǎn)首先賦相等的初值。接下來，根據(jù)鏈接關(guān)系將值傳播到鏈接去的節(jié)點(diǎn)。如此迭代直到收斂。

需要特殊處理的地方是，出度為 0 的節(jié)點(diǎn)需要將值保存到自己。

為了避免自私的節(jié)點(diǎn)不引用別人，從而大量積累自己的值，進(jìn)行平滑處理。給每一個(gè)節(jié)點(diǎn)乘以縮減因子 ，再將每個(gè)節(jié)點(diǎn)加上相等的 。注意到這種平滑不改變總值。也即任何時(shí)刻所有節(jié)點(diǎn)的值之和恒為 1 。

與之相關(guān)的還有 特征向量中心度 eigenvector centrality ，其區(qū)別是，不處理出度為 0 的點(diǎn)，也不進(jìn)行平滑。而在每一步進(jìn)行正規(guī)化。此外，特征向量也可以使用入度作為標(biāo)準(zhǔn)，僅需將連接矩陣轉(zhuǎn)置即可。

這里給出一種簡潔的三合一 Haskell 實(shí)現(xiàn)。不使用任何復(fù)雜的庫函數(shù)，僅用 80 行。從中可以看到 Haskell 的簡潔和抽象能力。

三種算法的核心都是不斷迭代直到收斂。將這一邏輯抽象出來得到：

converge :: Eq a => (a -> a) -> a -> a
converge f v = fst $ until theSame update (v, f v)
  where
    theSame (x, y) = x == y
    update (x, y) = (y, f y)

這里用到了庫函數(shù) until :: (a -> Bool) -> (a -> a) -> a -> a 。這個(gè)函數(shù)接收一個(gè)判斷函數(shù)，一個(gè)更新函數(shù)和初值。當(dāng)判斷函數(shù)返回假時(shí)，會(huì)應(yīng)用更新函數(shù)。當(dāng)判斷函數(shù)返回真時(shí)，返回最終值。

converge 函數(shù)實(shí)際上要構(gòu)造一個(gè)流（stream），即 v : f v : f (f v) : f (f (f v)) : ... 。當(dāng)流的兩個(gè)連續(xù)元素相等時(shí)，我們找到了 f 這個(gè)函數(shù)的不動(dòng)點(diǎn)，也就是最終的收斂值。

因?yàn)橹恍枰容^前兩個(gè)元素，所以我們使用兩個(gè)元素的元組（tuple）作為保存的狀態(tài)。until 的判斷函數(shù)就是兩個(gè)元素是否相等。更新函數(shù)是拋棄第一個(gè)元素，對第二個(gè)元素應(yīng)用 f 。

接下來不同算法的區(qū)別，僅在更新函數(shù)不同。

對于 pageRank 來說，就是不斷乘以連接矩陣：

pageRank :: [[Value]] -> [Value] -> [Value]
pageRank a vs = head $ converge (`matmul` a') [vs]
  where
    a' = compensate a

其中 matmul :: (Num a) => [[a]] -> [[a]] -> [[a]] 是矩陣乘法，將在下面給出實(shí)現(xiàn)。

注意到，首先將初值用列表改成 (n, 1) 的行向量，因此每次迭代改為右乘連接矩陣。最后使用 head 再轉(zhuǎn)變成一維列表 (n,) 。下面各個(gè)算法做同樣的處理。

compensate 函數(shù)實(shí)現(xiàn)兩個(gè)功能，對于出度不為 0 的節(jié)點(diǎn)，將因子 1 平均分配到每個(gè)非零節(jié)點(diǎn)上；對于出度為 0 的節(jié)點(diǎn)，將 1 分配到自己的位置上（矩陣對角線）。

compensate :: [[Value]] -> [[Value]]
compensate = map procOut . zip [0 ..]
  where
    procOut (i, l) =
      if any (/= 0) l
        then distribute l
        else oneAt i l
    distribute l =
      let v = 1.0 / (sum l)
       in map
            (\x ->
               if x == 0
                 then x
                 else v)
            l
    oneAt i l =
      let (x, _:ys) = splitAt i l
       in x ++ 1.0 : ys

平滑處理可以改為對連接矩陣進(jìn)行修改：

smooth :: Value -> [[Value]] -> [[Value]]
smooth s m = map (map interpolate) m
  where
    interpolate a = s * a + (1.0 - s) / fromIntegral n
    n = length m

對每一個(gè)元素，都用因子 s 縮減，再加上補(bǔ)償。

那么平滑后的 PageRank 算法如下：

smoothPageRank :: Value -> [[Value]] -> [Value] -> [Value]
smoothPageRank s a vs = head $ converge (`matmul` a') $ [vs]
  where
    a' = smooth s . compensate $ a

對于特征向量中心性，需要實(shí)現(xiàn)正規(guī)化：

normalize :: (Fractional a, Ord a) => [a] -> [a]
normalize vs =
  let m = maximum . (map abs) $ vs
   in map (/ m) vs

即將一個(gè)行向量的每個(gè)元素除以最大值。

那么特征向量中心性可以實(shí)現(xiàn)如下：

eiginCentr :: [[Value]] -> [Value] -> [Value]
eiginCentr a vs =
  head $ converge ((map normalize) . (`matmul` a)) [vs]

以上已經(jīng)實(shí)現(xiàn)了三個(gè)算法的核心部分。接下來給出輔助函數(shù)的直觀定義。

矩陣乘法：

dot :: (Num a) => [a] -> [a] -> a
dot x y = sum $ zipWith (*) x y

matmul :: (Num a) => [[a]] -> [[a]] -> [[a]]
matmul a b = map rowMul a
  where
    b' = transpose b
    rowMul r = map (dot r) b'

類型轉(zhuǎn)換：

type Value = Double

aFromIntegral :: (Integral a) => [[a]] -> [[Value]]
aFromIntegral = map (map fromIntegral)

生成初始平均分配值：

normalDist :: Int -> [Value]
normalDist n = replicate n $ 1.0 / fromIntegral n

圖從邊表示轉(zhuǎn)化為連接矩陣表示：

edgeToAdj :: (Integral a) => [(a, a)] -> [[a]]
edgeToAdj es = [[query i j | j <- [0 .. upper]] | i <- [0 .. upper]]
  where
    (ls, rs) = unzip es
    vs = ls ++ rs
    upper = maximum vs -- lower bound = 0
    query i j =
      if elem (i, j) es
        then 1
        else 0

其實(shí)這里使用 ST monad 更好一點(diǎn)，僅需要 的時(shí)間復(fù)雜度。這里用的是直接搜索，需要 的時(shí)間復(fù)雜度。

以上代碼實(shí)現(xiàn)了所有三個(gè)算法的功能，僅用了 80 行代碼。完整代碼見 gist 。

使用下圖進(jìn)行測試：

Network Example

-- Test Graph 2
tg2e =
  [ (0, 8)
  , (1, 6)
  , (1, 10)
  , (1, 11)
  , (2, 1)
  , (2, 10)
  , (2, 11)
  , (3, 15)
  , (3, 17)
  , (4, 1)
  , (4, 6)
  , (4, 15)
  , (5, 7)
  , (5, 8)
  , (5, 16)
  , (6, 5)
  , (6, 8)
  , (6, 16)
  , (7, 5)
  , (7, 13)
  , (7, 15)
  , (8, 16)
  , (8, 5)
  , (8, 6)
  , (9, 11)
  , (9, 10)
  , (9, 2)
  , (10, 9)
  , (10, 11)
  , (10, 13)
  , (11, 9)
  , (11, 10)
  , (11, 15)
  , (12, 13)
  , (12, 15)
  , (12, 16)
  , (13, 14)
  , (13, 15)
  , (13, 16)
  , (14, 13)
  , (14, 12)
  , (14, 15)
  , (15, 1)
  , (15, 9)
  , (15, 11)
  , (16, 7)
  , (16, 8)
  , (16, 13)
  ]

tg2 = edgeToAdj tg2e

tg2spr = smoothPageRank 0.8 (aFromIntegral tg2) (normalDist . length $ tg2)

printTg2spr :: IO ()
printTg2spr = mapM_ (printf "%.3f\n") tg2spr

測試結(jié)果如下：

$ stack ghci
λ> :load pagerank.hs
[1 of 1] Compiling Main             ( pagerank.hs, interpreted )
Ok, one module loaded.
λ> printTg2spr
0.011
0.049
0.034
0.011
0.011
0.054
0.045
0.048
0.069
0.087
0.084
0.104
0.020
0.083
0.033
0.095
0.083
0.078
λ>

符合預(yù)期。

連矩陣乘法都從頭開始寫，到整個(gè)算法完成，僅需要 80 行代碼。核心就是 converge 函數(shù)的抽象。這個(gè)例子很好地體現(xiàn)了 Haskell 作為函數(shù)式語言的優(yōu)點(diǎn)。