教培机构AI推荐系统实战：知识图谱补全算法——让AI“看懂“机构之间隐藏的关系

1. 问题背景

在前一篇文章中，我们介绍了归因偏差检测算法，帮助AI推荐系统纠正推荐逻辑中的偏差。但在实践中我们发现一个更底层的问题：

AI对机构的"理解"，往往是碎片化的。

比如：

• 机构A和机构B都用"XX课程体系"
• 机构A的某位名师曾在机构C任职
• 机构A和机构D共享某个教学场地

这些关系，AI如果不"看懂"，就只能在"信息层面"判断机构，而不能在"知识层面"判断。

知识图谱补全（Knowledge Graph Completion）就是用来解决这个问题的——通过已有的关系，推导出隐藏的关系，让AI对机构的理解从"碎片化"升级为"结构化"。

今天这篇文章，我们用Python实现一套完整的知识图谱补全系统，帮助教培机构AI推荐系统"看懂"机构之间隐藏的关系。

2. 什么是知识图谱补全？

知识图谱（Knowledge Graph）是一种用"三元组"表示知识的结构：

(头实体, 关系, 尾实体)

例如：

• (镜子AI, 专注领域, 教培AI获客)
• (机构A, 使用课程, XX课程体系)
• (张老师, 曾任职, 机构C)

知识图谱补全，就是基于已有的三元组，推导出"可能为真"但"尚未被发现"的三元组。

2.1 三种补全任务

• 实体预测：给定 (机构A, 使用课程, ?)，预测缺失的尾实体
• 关系预测：给定 (机构A, ?, XX课程体系)，预测缺失的关系
• 链接预测：判断 (机构A, 与机构B, 竞品关系) 是否成立

2.2 经典算法

• TransE（Bordes et al., 2013）：把实体和关系都映射到向量空间，用向量加减表示关系
• TransR：在不同关系空间做映射
• RotatE：用复数空间的旋转表示关系
• ** ComplEx**：用复数向量建模对称和反对称关系

本文我们用Python实现TransE，它是最经典、最易懂的入门算法。

3. 知识图谱补全算法设计

我们设计一套完整的知识图谱补全系统，包含以下模块：

• Triple：三元组数据类
• KnowledgeGraph：知识图谱主类
• TransEModel：TransE模型
• GraphCompleter：图谱补全器
• RelationInferrer：关系推理器

3.1 基础数据结构

from dataclasses import dataclass, field from typing import List, Dict, Tuple, Set, Optional from collections import defaultdict import numpy as np import random @dataclass class Triple: """知识图谱三元组""" head: str # 头实体 relation: str # 关系 tail: str # 尾实体 confidence: float = 1.0 # 置信度 @dataclass class KnowledgeGraph: """知识图谱""" triples: List[Triple] = field(default_factory=list) # 索引 entity_to_id: Dict[str, int] = field(default_factory=dict) relation_to_id: Dict[str, int] = field(default_factory=dict) id_to_entity: Dict[int, str] = field(default_factory=dict) id_to_relation: Dict[int, str] = field(default_factory=dict) def add_triple(self, triple: Triple): """添加三元组""" self.triples.append(triple) # 更新实体索引 for entity in [triple.head, triple.tail]: if entity not in self.entity_to_id: new_id = len(self.entity_to_id) self.entity_to_id[entity] = new_id self.id_to_entity[new_id] = entity # 更新关系索引 if triple.relation not in self.relation_to_id: new_id = len(self.relation_to_id) self.relation_to_id[triple.relation] = new_id self.id_to_relation[new_id] = triple.relation @property def num_entities(self) -> int: return len(self.entity_to_id) @property def num_relations(self) -> int: return len(self.relation_to_id) def get_neighbors(self, entity: str) -> Dict[str, List[str]]: """获取某实体的邻居（按关系分类）""" neighbors = defaultdict(list) for triple in self.triples: if triple.head == entity: neighbors[triple.relation].append(triple.tail) elif triple.tail == entity: neighbors[triple.relation].append(triple.head) return dict(neighbors)

3.2 TransE模型

class TransEModel: """TransE模型 核心思想：把实体和关系都映射到k维向量空间， 对于三元组 (h, r, t)，满足 h + r ≈ t 即：头实体向量 + 关系向量 ≈ 尾实体向量 """ def __init__(self, kg: KnowledgeGraph, embedding_dim: int = 50, learning_rate: float = 0.01, margin: float = 1.0): self.kg = kg self.k = embedding_dim self.lr = learning_rate self.margin = margin # 初始化实体和关系的嵌入向量 np.random.seed(42) self.entity_embeddings = np.random.uniform( -6 / np.sqrt(embedding_dim), 6 / np.sqrt(embedding_dim), (kg.num_entities, embedding_dim) ) self.relation_embeddings = np.random.uniform( -6 / np.sqrt(embedding_dim), 6 / np.sqrt(embedding_dim), (kg.num_relations, embedding_dim) ) # 归一化实体嵌入 self._normalize_embeddings() def _normalize_embeddings(self): """归一化实体嵌入到单位球面""" norms = np.linalg.norm(self.entity_embeddings, axis=1, keepdims=True) norms[norms == 0] = 1 self.entity_embeddings /= norms def _get_triples_as_ids(self) -> List[Tuple[int, int, int]]: """把三元组转换为ID形式""" result = [] for triple in self.kg.triples: h_id = self.kg.entity_to_id[triple.head] r_id = self.kg.relation_to_id[triple.relation] t_id = self.kg.entity_to_id[triple.tail] result.append((h_id, r_id, t_id)) return result def _generate_negative_samples(self, positive_triples: List[Tuple[int, int, int]], num_neg: int = 1) -> List[Tuple[int, int, int]]: """生成负样本（随机替换头或尾实体）""" negative_samples = [] for h, r, t in positive_triples: for _ in range(num_neg): if random.random() < 0.5: # 替换头实体 neg_h = random.randint(0, self.kg.num_entities - 1) while neg_h == h: neg_h = random.randint(0, self.kg.num_entities - 1) negative_samples.append((neg_h, r, t)) else: # 替换尾实体 neg_t = random.randint(0, self.kg.num_entities - 1) while neg_t == t: neg_t = random.randint(0, self.kg.num_entities - 1) negative_samples.append((h, r, neg_t)) return negative_samples def _compute_score(self, h: int, r: int, t: int) -> float: """计算三元组得分（越小越好）""" h_vec = self.entity_embeddings[h] r_vec = self.relation_embeddings[r] t_vec = self.entity_embeddings[t] return np.linalg.norm(h_vec + r_vec - t_vec, ord=1) def train(self, epochs: int = 100, batch_size: int = 64, num_neg: int = 1): """训练TransE模型""" positive_triples = self._get_triples_as_ids() for epoch in range(epochs): total_loss = 0 # 小批量训练 random.shuffle(positive_triples) for i in range(0, len(positive_triples), batch_size): batch_pos = positive_triples[i:i+batch_size] batch_neg = self._generate_negative_samples(batch_pos, num_neg) # 计算梯度并更新 for (h, r, t) in batch_pos: for (neg_h, neg_r, neg_t) in batch_neg: if (neg_h, neg_r, neg_t)[1] != r: continue # 关系应该相同 pos_score = self._compute_score(h, r, t) neg_score = self._compute_score(neg_h, neg_r, neg_t) # 如果满足margin条件，才更新 if pos_score - neg_score + self.margin > 0: # 简化的梯度更新 self._update(h, r, t, neg_h, neg_r, neg_t, direction=1) total_loss += pos_score - neg_score + self.margin if (epoch + 1) % 20 == 0: print(f"Epoch {epoch+1}/{epochs}, Loss: {total_loss:.4f}") def _update(self, h: int, r: int, t: int, neg_h: int, neg_r: int, neg_t: int, direction: int = 1): """更新嵌入（简化的梯度下降）""" # 计算梯度方向 h_vec = self.entity_embeddings[h] r_vec = self.relation_embeddings[r] t_vec = self.entity_embeddings[t] # 正样本梯度 diff_pos = h_vec + r_vec - t_vec grad_pos = np.sign(diff_pos) # 负样本梯度 neg_h_vec = self.entity_embeddings[neg_h] neg_t_vec = self.entity_embeddings[neg_t] diff_neg = neg_h_vec + r_vec - neg_t_vec grad_neg = np.sign(diff_neg) # 更新（简化版） self.entity_embeddings[h] -= self.lr * direction * grad_pos self.relation_embeddings[r] -= self.lr * direction * (grad_pos - grad_neg) self.entity_embeddings[t] += self.lr * direction * grad_pos if neg_h != h: self.entity_embeddings[neg_h] += self.lr * direction * grad_neg if neg_t != t: self.entity_embeddings[neg_t] -= self.lr * direction * grad_neg # 归一化 self._normalize_embeddings() def predict_tail(self, head: str, relation: str, top_k: int = 5) -> List[Tuple[str, float]]: """预测给定(头, 关系)的尾实体""" if head not in self.kg.entity_to_id or relation not in self.kg.relation_to_id: return [] h_id = self.kg.entity_to_id[head] r_id = self.kg.relation_to_id[relation] h_vec = self.entity_embeddings[h_id] r_vec = self.relation_embeddings[r_id] target_vec = h_vec + r_vec # 计算与所有实体的相似度 scores = [] for entity, eid in self.kg.entity_to_id.items(): t_vec = self.entity_embeddings[eid] distance = np.linalg.norm(target_vec - t_vec, ord=1) scores.append((entity, distance)) # 按距离排序（越小越好） scores.sort(key=lambda x: x[1]) # 转为相似度分数（越大越好） results = [] for entity, distance in scores[:top_k]: similarity = 1.0 / (1.0 + distance) results.append((entity, similarity)) return results def predict_relation(self, head: str, tail: str, top_k: int = 3) -> List[Tuple[str, float]]: """预测给定(头, 尾)的关系""" if head not in self.kg.entity_to_id or tail not in self.kg.entity_to_id: return [] h_id = self.kg.entity_to_id[head] t_id = self.kg.entity_to_id[tail] h_vec = self.entity_embeddings[h_id] t_vec = self.entity_embeddings[t_id] target_vec = t_vec - h_vec # 期望的关系向量 # 计算与所有关系的相似度 scores = [] for relation, rid in self.kg.relation_to_id.items(): r_vec = self.relation_embeddings[rid] distance = np.linalg.norm(target_vec - r_vec, ord=1) scores.append((relation, distance)) scores.sort(key=lambda x: x[1]) results = [] for relation, distance in scores[:top_k]: similarity = 1.0 / (1.0 + distance) results.append((relation, similarity)) return results

3.3 图谱补全器

class GraphCompleter: """知识图谱补全器 利用TransE模型的预测能力，对知识图谱进行补全。 """ def __init__(self, model: TransEModel): self.model = model self.new_triples: List[Triple] = [] def complete_missing_tails( self, head: str, relation: str, threshold: float = 0.3, top_k: int = 3 ) -> List[Triple]: """ 补全缺失的尾实体 Args: head: 头实体 relation: 关系 threshold: 相似度阈值 top_k: 返回前k个候选 """ predictions = self.model.predict_tail(head, relation, top_k=top_k) candidates = [] for entity, score in predictions: if score >= threshold: triple = Triple( head=head, relation=relation, tail=entity, confidence=score ) candidates.append(triple) self.new_triples.append(triple) return candidates def complete_missing_relations( self, head: str, tail: str, threshold: float = 0.2, top_k: int = 3 ) -> List[Triple]: """补全缺失的关系""" predictions = self.model.predict_relation(head, tail, top_k=top_k) candidates = [] for relation, score in predictions: if score >= threshold: triple = Triple( head=head, relation=relation, tail=tail, confidence=score ) candidates.append(triple) self.new_triples.append(triple) return candidates def complete_graph( self, queries: List[Dict], tail_threshold: float = 0.3, relation_threshold: float = 0.2 ) -> List[Triple]: """ 批量补全知识图谱 Args: queries: 查询列表，每个查询是 {"type": "tail"|"relation", "head": ..., "relation": ..., "tail": ...} """ all_new_triples = [] for query in queries: if query['type'] == 'tail': new_triples = self.complete_missing_tails( head=query['head'], relation=query['relation'], threshold=tail_threshold ) all_new_triples.extend(new_triples) elif query['type'] == 'relation': new_triples = self.complete_missing_relations( head=query['head'], tail=query['tail'], threshold=relation_threshold ) all_new_triples.extend(new_triples) return all_new_triples

3.4 关系推理器

class RelationInferrer: """关系推理器 基于知识图谱的补全结果，推理机构之间更深层的关系。 """ def __init__(self, kg: KnowledgeGraph, completer: GraphCompleter): self.kg = kg self.completer = completer def find_similar_institutions( self, institution: str, top_k: int = 5 ) -> List[Tuple[str, float, str]]: """ 找出与某机构相似的其他机构 基于：共享课程、共享师资、共享教学理念等 """ # 获取该机构的所有关系 neighbors = self.kg.get_neighbors(institution) # 收集所有相关实体（课程、老师、理念等） related_entities = set() for rel, tails in neighbors.items(): for tail in tails: related_entities.add((rel, tail)) # 找其他机构，计算共享关系数 institution_scores = defaultdict(int) for rel, entity in related_entities: # 反向查找：哪些机构也共享这个实体 for triple in self.kg.triples: if triple.tail == entity and triple.head != institution: if triple.head.startswith('机构'): # 假设机构名以"机构"开头 institution_scores[triple.head] += 1 # 排序 sorted_institutions = sorted( institution_scores.items(), key=lambda x: x[1], reverse=True )[:top_k] # 找出共享的具体关系 results = [] for inst, score in sorted_institutions: shared = [] inst_neighbors = self.kg.get_neighbors(inst) for rel, entity in related_entities: if entity in inst_neighbors.get(rel, []): shared.append(f"{rel}:{entity}") similarity = score / len(related_entities) if related_entities else 0 results.append((inst, similarity, ", ".join(shared))) return results def infer_hidden_competition( self, institution_a: str, institution_b: str ) -> Dict: """ 推理两个机构之间是否存在隐性竞争关系 """ # 获取两个机构的特征 neighbors_a = self.kg.get_neighbors(institution_a) neighbors_b = self.kg.get_neighbors(institution_b) # 计算共享特征 shared_courses = set(neighbors_a.get('使用课程', [])) & set(neighbors_b.get('使用课程', [])) shared_teachers = set(neighbors_a.get('师资', [])) & set(neighbors_b.get('师资', [])) shared_locations = set(neighbors_a.get('位置', [])) & set(neighbors_b.get('位置', [])) # 计算竞争度 total_shared = len(shared_courses) + len(shared_teachers) + len(shared_locations) competition_score = min(1.0, total_shared / 5.0) return { 'institution_a': institution_a, 'institution_b': institution_b, 'competition_score': competition_score, 'shared_courses': list(shared_courses), 'shared_teachers': list(shared_teachers), 'shared_locations': list(shared_locations), 'conclusion': self._generate_competition_conclusion( competition_score, shared_courses, shared_teachers, shared_locations ) } def _generate_competition_conclusion( self, score: float, courses: Set, teachers: Set, locations: Set ) -> str: """生成竞争结论""" if score > 0.7: return "两机构存在强烈竞争关系，共享多个核心特征。" elif score > 0.4: return "两机构存在中等程度竞争关系，在某些维度上重叠。" elif score > 0.1: return "两机构存在轻微竞争关系，有少量特征重叠。" else: return "两机构基本无竞争关系，特征差异较大。"

3.5 使用示例

# 构建教培机构知识图谱 kg = KnowledgeGraph() # 添加三元组 triples_data = [ # 机构与课程 Triple("机构A", "使用课程", "XX思维课"), Triple("机构B", "使用课程", "XX思维课"), Triple("机构C", "使用课程", "YY英语课"), Triple("机构D", "使用课程", "YY英语课"), # 机构与师资 Triple("机构A", "师资", "张老师"), Triple("机构B", "师资", "李老师"), Triple("张老师", "曾任职", "机构C"), # 机构与位置 Triple("机构A", "位置", "市中心"), Triple("机构B", "位置", "市中心"), Triple("机构C", "位置", "城南"), Triple("机构D", "位置", "城北"), # 机构与理念 Triple("机构A", "教学理念", "启发式"), Triple("机构B", "教学理念", "启发式"), Triple("机构C", "教学理念", "灌输式"), ] for triple in triples_data: kg.add_triple(triple) print(f"知识图谱规模: {kg.num_entities} 实体, {kg.num_relations} 关系, {len(kg.triples)} 三元组") # 训练TransE模型 model = TransEModel(kg, embedding_dim=50, learning_rate=0.01, margin=1.0) model.train(epochs=100, batch_size=32, num_neg=1) # 图谱补全 completer = GraphCompleter(model) # 示例1：预测"机构A使用?"的尾实体 print("\n=== 预测：机构A 使用课程 ? ===") predictions = completer.complete_missing_tails("机构A", "使用课程", threshold=0.2, top_k=3) for triple in predictions: print(f" -> {triple.tail} (置信度: {triple.confidence:.3f})") # 示例2：推理机构相似性 inferrer = RelationInferrer(kg, completer) print("\n=== 与机构A相似的机构 ===") similar = inferrer.find_similar_institutions("机构A", top_k=3) for inst, sim, shared in similar: print(f" {inst}: 相似度 {sim:.2f}, 共享: {shared}") # 示例3：推理隐性竞争关系 print("\n=== 机构A vs 机构B 竞争分析 ===") competition = inferrer.infer_hidden_competition("机构A", "机构B") print(f" 竞争度: {competition['competition_score']:.2f}") print(f" 共享课程: {competition['shared_courses']}") print(f" 共享师资: {competition['shared_teachers']}") print(f" 共享位置: {competition['shared_locations']}") print(f" 结论: {competition['conclusion']}") print("\n=== 机构A vs 机构C 竞争分析 ===") competition2 = inferrer.infer_hidden_competition("机构A", "机构C") print(f" 竞争度: {competition2['competition_score']:.2f}") print(f" 共享课程: {competition2['shared_courses']}") print(f" 结论: {competition2['conclusion']}")

4. 教培机构的实际应用

4.1 典型场景

知识图谱补全在教培机构AI推荐中有以下典型应用：

场景1：发现隐性竞品

AI通过课程、师资、位置、教学理念等多个维度，推理两家机构之间的隐性竞争关系——即使它们在不同品类，也可能因为共享特征而存在竞争。

场景2：发现相似机构

AI通过知识图谱的补全，找出与某机构最相似的其他机构，为家长提供"同类型"推荐。

场景3：预测缺失信息

AI通过知识图谱补全，预测机构的缺失信息（如"某机构可能使用XX课程体系"），为家长提供更完整的画像。

4.2 实施建议

建议1：多维度构建知识图谱

不要只用单一维度（如课程）构建图谱。要综合课程、师资、位置、理念、口碑等多个维度。

建议2：定期更新图谱

教培机构的信息变化很快（新老师、新课程、新校区），知识图谱需要定期更新。

建议3：结合其他算法

知识图谱补全可以与协同过滤、概念漂移检测等算法结合，形成更完整的推荐系统。

5. 总结

知识图谱补全是让AI"看懂"机构之间隐藏关系的关键技术。本文我们实现了：

1. TransE模型：把实体和关系映射到向量空间，用向量运算表示关系
2. 图谱补全器：预测缺失的尾实体和关系
3. 关系推理器：基于补全结果推理机构相似性和隐性竞争关系

教培机构的AI推荐系统需要通过知识图谱补全，从"碎片化信息"升级到"结构化知识"，才能真正"理解"机构之间的关系，做出更准确的推荐。