Back openDesk Edu for a sovereign, open-source education β every vote counts.
Vote nowModern recommender systems produce predictions that users cannot interrogate. Collaborative filtering captures behavioural signals but offers no reasoning. Large Language Models generate fluent explanations but hallucinate and are poorly grounded in user history.
X-KGRank (Explainable Knowledge Graph RAG) unifies structural collaborative filtering with LLM-based explanation. From the MovieLens-1M dataset, it constructs a heterogeneous knowledge graph of 9,762 nodes and 999,264 edges. It trains a LightGCN ranker with content-aware SBERT initialization and applies popularity selective routing that grounds long-tail items in knowledge-graph paths while serving popular items from pre-trained knowledge.
The result: NDCG@10 = 0.2956, +17.1% over popularity baseline, and a critical finding β smaller models match larger models on explanation quality but fabricate facts more often.
Traditional recommender systems:
User β Collaborative Filtering β Recommendation
(matrix factorization, LightGCN, etc.)
Output: "You might like Movie X"
Explanation: None (or "because you liked Y")
LLM-based recommenders:
User β LLM β Recommendation + Explanation
Output: "You might like Movie X because..."
Explanation: "Based on your love of sci-fi and strong female leads..."
Problem: Hallucinated, not grounded in actual history
The gap: Collaborative filtering is accurate but unexplainable. LLMs are explainable but hallucinate.
X-KGRank bridges this gap by grounding explanations in knowledge graph structure.
# From MovieLens-1M (6,040 users, 3,704 items, 988,129 interactions)
class X_KGRankKG:
def __init__(self, ratings_data):
self.graph = Neo4jGraph()
# Create user nodes
for user_id in ratings_data.users:
self.graph.create_node('User', user_id, metadata={
'rating_count': len(ratings_data.get_user_ratings(user_id))
})
# Create item nodes
for item_id in ratings_data.items:
self.graph.create_node('Item', item_id, metadata={
'title': ratings_data.get_item_title(item_id),
'genres': ratings_data.get_item_genres(item_id)
})
# Create rating edges
for user_id, item_id, rating in ratings_data.interactions:
self.graph.create_edge('User', user_id, 'RATED', 'Item', item_id, {
'rating': rating,
'timestamp': ratings_data.get_timestamp(user_id, item_id)
})
# Create genre edges
for item_id, genres in ratings_data.item_genres.items():
for genre in genres:
self.graph.create_edge('Item', item_id, 'HAS_GENRE', 'Genre', genre)
# Create co-rating edges (popularity signal)
item_counts = ratings_data.get_item_popularity()
popularity_threshold = 100 # Items with >100 ratings are "popular"
for item_id, count in item_counts.items():
if count > popularity_threshold:
self.graph.create_edge('Item', item_id, 'CO_RATED', 'Item', item_id, {
'co_rating_count': count
})
Graph statistics:
class LightGCN_Ranker:
def __init__(self, graph, n_factors=128, n_layers=3):
self.graph = graph
self.n_factors = n_factors
self.n_layers = n_layers
# Initialize with SBERT embeddings (content-aware)
self.item_embeddings = self._init_with_sbert()
self.user_embeddings = self._init_user_embeddings()
# LightGCN layers
self.layers = [
LightGCNLayer(n_factors, n_factors)
for _ in range(n_layers)
]
# Rating-weighted BPR objective
self.loss = BPR_Loss(rating_weighted=True)
def _init_with_sbert(self):
# Load pre-trained SBERT for item titles
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = {}
for item_id, title in self.graph.get_items():
embeddings[item_id] = model.encode(title)
return embeddings
def forward(self, user_id, item_id):
# Get embeddings
user_emb = self.user_embeddings[user_id]
item_emb = self.item_embeddings[item_id]
# Propagate through LightGCN layers
for layer in self.layers:
user_emb = layer.propagate(user_emb, self.graph.user_item_edges)
item_emb = layer.propagate(item_emb, self.graph.item_user_edges)
# Predict rating
prediction = (user_emb * item_emb).sum()
return prediction
Key innovation: SBERT initialization provides content awareness even for cold-start items.
class PopularitySelectiveRouter:
def __init__(self, graph, threshold=100):
self.graph = graph
self.threshold = threshold # Items with >100 ratings are "popular"
# Pre-compute popular item knowledge
self.popular_items = self._get_popular_items()
self.popular_embeddings = self._embed_popular_items()
def route(self, item_id):
"""Decide: use KG path or pre-trained knowledge"""
popularity = self.graph.get_item_popularity(item_id)
if popularity > self.threshold:
# Popular item: use pre-trained knowledge (faster)
return 'pretrained'
else:
# Long-tail item: ground in KG paths (more accurate)
return 'kg_path'
def _get_popular_items(self):
# Query Neo4j for popular items
query = """
MATCH (i:Item)<-[r:RATED]-()
WITH i, count(r) as rating_count
WHERE rating_count > $threshold
RETURN i.item_id as item_id, rating_count
"""
result = self.graph.query(query, threshold=self.threshold)
return [r['item_id'] for r in result]
Key insight: 50% reduction in KG-augmented generations by serving popular items from pre-trained knowledge.
class LLM_ReRanker:
def __init__(self, llm, kg):
self.llm = llm
self.graph = kg # Knowledge graph for path finding
def rerank(self, candidates, user_history):
"""Re-rank candidates with LLM scoring"""
scored = []
for item in candidates:
# Generate explanation
explanation = self._generate_explanation(item, user_history)
# Score explanation quality
quality = self._score_explanation(explanation)
scored.append({
'item': item,
'explanation': explanation,
'quality': quality
})
# Sort by quality
scored.sort(key=lambda x: x['quality'], reverse=True)
return scored
def _generate_explanation(self, item, user_history):
# Query KG for paths between user history and item
paths = self.graph.find_paths(
start=user_history.items,
end=item,
max_hops=3
)
# Generate explanation from paths
prompt = f"""
User has rated: {user_history.top_items}
Recommended item: {item.title}
Knowledge graph paths:
{self._format_paths(paths)}
Generate a natural explanation for this recommendation.
"""
explanation = self.llm.generate(prompt)
return explanation
def _format_paths(self, paths):
"""Format KG paths into human-readable text for the LLM prompt."""
lines = []
for path in paths:
hops = " β ".join(str(node) for node in path)
lines.append(f" - {hops}")
return "\n".join(lines)
def _score_explanation(self, explanation):
# Heuristic scoring (0-1)
# Based on: specificity, grounding, coherence
score = 0
# Check for specific item references
if len(explanation.split()) > 20:
score += 0.3
# Check for user history mentions
if 'because you' in explanation.lower():
score += 0.3
# Check for knowledge grounding
if 'similar to' in explanation.lower() or 'also' in explanation.lower():
score += 0.4
return score
X-KGRank was evaluated on MovieLens-1M test set under a 99-sample protocol:
| Metric | Popularity Baseline | X-KGRank | Improvement |
|---|---|---|---|
| NDCG@10 | 0.2525 | 0.2956 | +17.1% |
| Recall@10 | 0.4615 | 0.5371 | +16.4% |
| NDCG@20 | 0.2983 | 0.3449 | +15.6% |
| MRR | 0.2124 | 0.2435 | +14.6% |
LLM explanation quality (across three backbones, 16 cases):
| Model | Parameter Count | Explanation Quality | Fact Fabrication Rate |
|---|---|---|---|
| Qwen2.5 | 1.5B | 0.97 | 12% |
| Mistral | 7B | 0.94 | 8% |
| Llama3 | 8B | 0.95 | 7% |
Key findings:
// Find paths for explanation generation
MATCH (u:User {user_id: $user_id})-[:RATED]->(liked:Item)
MATCH (liked)-[:HAS_GENRE]->(genre)<-[:HAS_GENRE]-(recommended:Item {item_id: $item_id})
RETURN liked.title as liked_item, genre.name as shared_genre, recommended.title as recommended_item
// Get similar items via co-rating
MATCH (i1:Item {item_id: $item_id})-[:CO_RATED]-(i2:Item)
RETURN i2.title as similar_item, i2.item_id as item_id
ORDER BY i2.rating_count DESC
LIMIT 10
// User profile aggregation
MATCH (u:User {user_id: $user_id})-[:RATED]->(item:Item)-[:HAS_GENRE]->(genre:Genre)
RETURN genre.name as genre, count(*) as count
ORDER BY count DESC
class CostOptimisedX_KGRank:
def __init__(self, kg, ranker, llm, router, llm_reranker):
self.kg = kg
self.ranker = ranker
self.llm = llm
self.router = router # PopularitySelectiveRouter
self.llm_reranker = llm_reranker # LLM_ReRanker
self.cache = LRUCache(maxsize=10000)
def recommend(self, user_id, n=10):
# Check cache
cache_key = f"recommend:{user_id}:{n}"
if cache_key in self.cache:
return self.cache[cache_key]
# Get candidates via LightGCN
candidates = self.ranker.get_candidates(user_id, n=50)
# Route: popular vs long-tail
routed = []
for item in candidates:
route = self.router.route(item.id)
if route == 'pretrained':
# Fast path: use pre-trained embedding
score = self.ranker.predict_pretrained(user_id, item.id)
else:
# Slow path: KG grounding
paths = self.kg.find_paths(user_id, item.id, max_hops=3)
score = self.ranker.predict_kg(user_id, item.id, paths)
routed.append((item, score))
# Re-rank with LLM (only top 10)
reranked = self.llm_reranker.rerank(routed[:10], user_id)
# Cache result
self.cache[cache_key] = reranked
return reranked
# Batch processing for offline recommendations
def batch_recommendations(users, batch_size=1000):
for i in range(0, len(users), batch_size):
batch = users[i:i+batch_size]
# Parallel LightGCN inference
candidates = parallel_lightgcn_predict(batch)
# Parallel KG path finding
paths = parallel_kg_query(candidates)
# Serial LLM re-ranking (bottleneck)
reranked = llm_rerank_batch(candidates, paths)
# Store results
store_recommendations(reranked)
X-KGRank reveals three trends:
Pure collaborative filtering or pure LLM recommendation are both suboptimal. Hybrid systems that combine structural signals with generative explanation will dominate.
Smaller models are cost-effective but less factual. Knowledge graph grounding narrows the gap but doesn't eliminate it.
Popular items can be served from pre-trained knowledge. Long-tail items require KG grounding. Selective routing cuts costs by 50%.
X-KGRank solves the explainability problem by:
For recommendation systems, the implication is clear: explanation requires grounding. Ungrounded LLMs hallucinate.