llm_kernel/graph/algo/
pagerank.rs

1//! Weighted PageRank centrality.
2//!
3//! Standard power iteration with dangling-node redistribution. Edges carry
4//! weights (relationship strength), so rank flows along an out-edge in
5//! proportion to that edge's weight divided by the source's total out-weight —
6//! stronger relations pass more rank, matching the `smart_recall` boost
7//! semantics.
8//!
9//! The iterative math is pure-Rust over a [`CsrGraph`] and backend-agnostic,
10//! so the SQLite and PostgreSQL recall paths can import the same
11//! [`pagerank_default`] and stay drift-free (see `recall.rs`'s `compute_recency`
12//! for the same principle).
13
14use rusqlite::Connection;
15
16use crate::error::Result;
17
18use super::csr::CsrGraph;
19
20/// damping factor (rank a node keeps vs. teleports); the canonical 0.85.
21pub const PAGERANK_DAMPING: f64 = 0.85;
22/// Maximum power-iteration count before giving up on further convergence.
23pub const PAGERANK_ITERS: usize = 100;
24/// L1-norm convergence threshold; iteration stops once the rank vector stops
25/// moving more than this between steps.
26pub const PAGERANK_EPS: f64 = 1e-6;
27
28/// Weighted PageRank over `g`, returning one score per node index.
29///
30/// Scores sum to 1.0 (normalized to absorb floating-point drift). Dangling
31/// nodes — those with no out-edges, or all-zero out-weights — have their rank
32/// mass redistributed uniformly each iteration so rank does not leak.
33pub fn pagerank(g: &CsrGraph, damping: f64, iters: usize) -> Vec<f64> {
34    let n = g.node_count();
35    if n == 0 {
36        return Vec::new();
37    }
38    let base = (1.0 - damping) / n as f64;
39    let mut r = vec![1.0 / n as f64; n];
40    let mut next = vec![0.0_f64; n];
41
42    for _ in 0..iters {
43        for slot in next.iter_mut() {
44            *slot = 0.0;
45        }
46
47        // Distribute rank from non-dangling nodes along weighted out-edges,
48        // accumulating dangling mass for uniform redistribution.
49        let mut dangling_mass = 0.0_f64;
50        for i in g.node_indices() {
51            let out_sum = g.out_weight_sum(i);
52            if out_sum <= 0.0 {
53                dangling_mass += r[i as usize];
54                continue;
55            }
56            let share = damping * r[i as usize] / out_sum;
57            for (j, w) in g.out_neighbors(i) {
58                next[j as usize] += share * w;
59            }
60        }
61
62        // Teleport base + dangling redistribution is uniform across all nodes.
63        let uniform = base + damping * dangling_mass / n as f64;
64        for slot in next.iter_mut() {
65            *slot += uniform;
66        }
67
68        // L1 convergence check.
69        let mut delta = 0.0_f64;
70        for j in 0..n {
71            delta += (next[j] - r[j]).abs();
72        }
73        std::mem::swap(&mut r, &mut next);
74        if delta < PAGERANK_EPS {
75            break;
76        }
77    }
78
79    // Normalize so Σ == 1.0 exactly (absorbs accumulated float drift).
80    let sum: f64 = r.iter().sum();
81    if sum > 0.0 {
82        for v in &mut r {
83            *v /= sum;
84        }
85    }
86    r
87}
88
89/// PageRank with the canonical defaults (`PAGERANK_DAMPING`, `PAGERANK_ITERS`).
90pub fn pagerank_default(g: &CsrGraph) -> Vec<f64> {
91    pagerank(g, PAGERANK_DAMPING, PAGERANK_ITERS)
92}
93
94/// Build a CSR snapshot from `conn` and return `(node_id, score)` pairs in
95/// node-index order.
96pub fn pagerank_scores(conn: &Connection) -> Result<Vec<(String, f64)>> {
97    let g = CsrGraph::build_csr(conn)?;
98    let scores = pagerank_default(&g);
99    Ok(g.node_indices()
100        .map(|i| (g.node_id(i).to_string(), scores[i as usize]))
101        .collect())
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use crate::graph::algo::csr::CsrGraph;
108    use crate::graph::schema::init_graph_schema;
109    use crate::graph::store::{append_edge, upsert_node};
110    use crate::graph::types::{GraphEdge, GraphNode};
111    use rusqlite::Connection;
112
113    fn mem_db() -> Connection {
114        let conn = Connection::open_in_memory().unwrap();
115        init_graph_schema(&conn).unwrap();
116        conn
117    }
118
119    fn edge(id: &str, s: &str, t: &str, w: f64) -> GraphEdge {
120        GraphEdge {
121            id: id.into(),
122            source: s.into(),
123            target: t.into(),
124            relation: "related".into(),
125            weight: w,
126            ts: "2026-01-01T00:00:00Z".into(),
127        }
128    }
129
130    /// Diamond: A→B, A→C, B→D, C→D. D is a sink. B, C symmetric.
131    fn diamond_csr() -> CsrGraph {
132        let nodes: Vec<String> = ["A", "B", "C", "D"].iter().map(|s| s.to_string()).collect();
133        let edges = vec![
134            edge("e1", "A", "B", 1.0),
135            edge("e2", "A", "C", 1.0),
136            edge("e3", "B", "D", 1.0),
137            edge("e4", "C", "D", 1.0),
138        ];
139        CsrGraph::from_edges(&nodes, &edges)
140    }
141
142    #[test]
143    fn empty_graph_returns_empty() {
144        let g = CsrGraph::from_edges(&[], &[]);
145        assert!(pagerank_default(&g).is_empty());
146    }
147
148    #[test]
149    fn single_node_scores_one() {
150        let nodes: Vec<String> = vec!["A".to_string()];
151        let g = CsrGraph::from_edges(&nodes, &[]);
152        let pr = pagerank_default(&g);
153        assert_eq!(pr.len(), 1);
154        assert!((pr[0] - 1.0).abs() < 1e-9);
155    }
156
157    #[test]
158    fn diamond_sink_dominates_and_symmetric() {
159        let g = diamond_csr();
160        let pr = pagerank_default(&g);
161        let a = g.node_index("A").unwrap() as usize;
162        let b = g.node_index("B").unwrap() as usize;
163        let c = g.node_index("C").unwrap() as usize;
164        let d = g.node_index("D").unwrap() as usize;
165
166        // Normalization invariant.
167        assert!(
168            (pr.iter().sum::<f64>() - 1.0).abs() < 1e-9,
169            "scores must sum to 1.0"
170        );
171        // All positive.
172        assert!(pr.iter().all(|&v| v > 0.0));
173        // D (sink) outranks A (pure source).
174        assert!(pr[d] > pr[a], "sink D should outrank source A");
175        // B and C are structurally symmetric.
176        assert!(
177            (pr[b] - pr[c]).abs() < 1e-9,
178            "symmetric nodes B,C must match"
179        );
180        // D is the global maximum.
181        let max_idx = (0..pr.len())
182            .max_by(|&a, &b| {
183                pr[a]
184                    .partial_cmp(&pr[b])
185                    .unwrap_or(std::cmp::Ordering::Equal)
186            })
187            .unwrap();
188        assert_eq!(max_idx, d);
189    }
190
191    #[test]
192    fn dangling_mass_conserved() {
193        // A→B, B is dangling. Rank must not leak: Σ == 1.0 after convergence.
194        let nodes: Vec<String> = ["A", "B"].iter().map(|s| s.to_string()).collect();
195        let edges = vec![edge("e1", "A", "B", 1.0)];
196        let g = CsrGraph::from_edges(&nodes, &edges);
197        let pr = pagerank_default(&g);
198        assert!((pr.iter().sum::<f64>() - 1.0).abs() < 1e-9);
199        // B receives A's rank plus teleport — outranks A.
200        let a = g.node_index("A").unwrap() as usize;
201        let b = g.node_index("B").unwrap() as usize;
202        assert!(pr[b] > pr[a]);
203    }
204
205    #[test]
206    fn weighted_edges_flow_more_rank() {
207        // Two parallel two-hop paths A→hub→sink: one via a strong edge, one weak.
208        // The sink reachable through the stronger link accrues more rank.
209        let nodes: Vec<String> = ["A", "P", "Q", "Sp", "Sq"]
210            .iter()
211            .map(|s| s.to_string())
212            .collect();
213        let edges = vec![
214            edge("e1", "A", "P", 0.9), // strong path
215            edge("e2", "A", "Q", 0.1), // weak path
216            edge("e3", "P", "Sp", 1.0),
217            edge("e4", "Q", "Sq", 1.0),
218        ];
219        let g = CsrGraph::from_edges(&nodes, &edges);
220        let pr = pagerank_default(&g);
221        let sp = g.node_index("Sp").unwrap() as usize;
222        let sq = g.node_index("Sq").unwrap() as usize;
223        assert!(
224            pr[sp] > pr[sq],
225            "sink on the stronger A→P edge should outrank the weaker A→Q sink"
226        );
227        assert!((pr.iter().sum::<f64>() - 1.0).abs() < 1e-9);
228    }
229
230    #[test]
231    fn pagerank_scores_from_sqlite() {
232        let conn = mem_db();
233        for id in ["A", "B", "C"] {
234            let n = GraphNode {
235                id: id.into(),
236                node_type: "concept".into(),
237                title: id.into(),
238                body: String::new(),
239                tags: vec![],
240                projects: vec![],
241                agents: vec![],
242                created: "2026-01-01T00:00:00Z".into(),
243                updated: "2026-01-01T00:00:00Z".into(),
244                importance: 0.5,
245                access_count: 0,
246                accessed_at: String::new(),
247            };
248            upsert_node(&conn, &n).unwrap();
249        }
250        append_edge(&conn, &edge("e1", "A", "B", 1.0)).unwrap();
251        append_edge(&conn, &edge("e2", "A", "C", 1.0)).unwrap();
252
253        let scores = pagerank_scores(&conn).unwrap();
254        assert_eq!(scores.len(), 3);
255        let map: std::collections::HashMap<String, f64> = scores.into_iter().collect();
256        let sum: f64 = map.values().sum();
257        assert!((sum - 1.0).abs() < 1e-9);
258    }
259}
llm_kernel/graph/algo/pagerank.rs

llm_kernel/graph/algo/
pagerank.rs
