Semantica 中的去重 · Semantica Cookbook

欢迎来到去重演练！在任何知识图谱中，数据往往来自多个来源，导致出现重复实体（例如 "Apple Inc." 与 "Apple Inc"）。

Semantica 提供了健壮的去重模块来帮助你： 1. 计算相似度：使用字符串、属性和嵌入来比较实体。 2. 检测重复：找出代表同一现实世界对象的一对或一组实体。 3. 聚类实体：将相似实体分组。 4. 合并实体：将重复项合并为单个规范化实体，同时消解冲突。

本 notebook 将通过清晰的示例引导你完成每一步。

# 安装 Semantica
!pip install -q semantica

11. 准备示例数据

让我们创建一个带有一些刻意重复项的数据集。我们将模拟来自不同来源（例如 CRM 和公共数据库）的数据。

我们的实体： - Apple：诸如 "Apple Inc."、"Apple Inc" 和仅 "Apple" 的变体。 - Microsoft：诸如 "Microsoft Corp" 和 "Microsoft" 的变体。 - Google：一个用于对照的唯一实体。

entities = [
    # Apple 变体
    {
        "id": "e1",
        "name": "Apple Inc.",
        "type": "Company",
        "properties": {"industry": "Technology", "hq": "Cupertino", "founded": 1976},
        "relationships": [{"predicate": "founded_by", "object": "Steve Jobs"}]
    },
    {
        "id": "e2",
        "name": "Apple Inc",
        "type": "Company",
        "properties": {"industry": "Tech", "hq": "Cupertino, CA"}, # 属性略有不同
        "relationships": []
    },
    {
        "id": "e3",
        "name": "Apple",
        "type": "Company",
        "properties": {"industry": "Consumer Electronics"}, 
        "relationships": [{"predicate": "ceo", "object": "Tim Cook"}]
    },

    # Microsoft 变体
    {
        "id": "e4",
        "name": "Microsoft Corp",
        "type": "Company",
        "properties": {"industry": "Software", "hq": "Redmond"}
    },
    {
        "id": "e5",
        "name": "Microsoft",
        "type": "Company",
        "properties": {"industry": "Tech", "hq": "Redmond, WA"}
    },

    # 唯一实体
    {
        "id": "e6",
        "name": "Google LLC",
        "type": "Company",
        "properties": {"industry": "Internet"}
    }
]

print(f"Created {len(entities)} sample entities.")

22. 相似度计算

SimilarityCalculator 是核心引擎。它比较两个实体并返回一个介于 0 和 1 之间的分数。它考察： - 字符串相似度：名称和文本字段。 - 属性相似度：键值对的重叠。 - 关系相似度：与其他实体的连接。 - 嵌入：语义向量相似度（如果可用）。

你可以自定义每个因素的权重。

from semantica.deduplication import SimilarityCalculator, SimilarityResult
# 使用自定义权重初始化计算器
calculator = SimilarityCalculator(
    string_weight=0.5,      # 名称权重高
    property_weight=0.3,    # 属性权重中等
    relationship_weight=0.2 # 关系权重较低
)

# 比较 "Apple Inc." (e1) 与 "Apple Inc" (e2)
score_e1_e2 = calculator.calculate_similarity(entities[0], entities[1])

print(f"Similarity between '{entities[0]['name']}' and '{entities[1]['name']}':")
print(f"  Total Score: {score_e1_e2.score:.4f}")
print(f"  Breakdown:   {score_e1_e2.components}")

# 比较 "Apple Inc." (e1) 与 "Microsoft" (e5)
score_e1_e5 = calculator.calculate_similarity(entities[0], entities[4])

print(f"\nSimilarity between '{entities[0]['name']}' and '{entities[4]['name']}':")
print(f"  Total Score: {score_e1_e5.score:.4f}")

33. 重复检测

DuplicateDetector 使用相似度计算器扫描你的数据集以查找重复项。它可以找到： - 对：简单的 A 匹配 B。 - 组：A 匹配 B，且 B 匹配 C。

它使用 similarity_threshold 来决定什么算作匹配。

# 导入用于重复检测的特定类
from semantica.deduplication import DuplicateDetector, DuplicateCandidate, DuplicateGroup
from semantica.deduplication import DeduplicationConfig

detector = DuplicateDetector(
    similarity_threshold=0.7, 
    confidence_threshold=0.6  
)

# 检测对
candidates = detector.detect_duplicates(entities)

print(f"Found {len(candidates)} duplicate pairs:")
for c in candidates:
    print(f"  - {c.entity1['name']} <==> {c.entity2['name']} (Score: {c.similarity_score:.2f})")

增量检测

如果你已有数据库并摄取新数据，你不想重新比较所有内容。使用 incremental_detect。

existing_db = entities[:3] # Apple 实体
new_data = [entities[4]]   # Microsoft
# 检查新数据是否与现有数据库中的任何内容匹配
inc_candidates = detector.incremental_detect(new_data, existing_db)

print(f"New matches found: {len(inc_candidates)}")
# 预期：0，因为 Microsoft 不是 Apple。

44. 聚类

有时仅靠"对"是不够的。ClusterBuilder 将相关实体分组为簇。这对于理解重复实体的完整范围很有用。

# 导入用于聚类的特定类
from semantica.deduplication import ClusterBuilder, Cluster, ClusterResult
cluster_builder = ClusterBuilder(threshold=0.7)
result = cluster_builder.build_clusters(entities)

print(f"Found {len(result.clusters)} clusters:")
for i, cluster in enumerate(result.clusters):
    names = [e['name'] for e in cluster.entities]
    print(f"  Cluster {i+1}: {names}")

cluster_builder = ClusterBuilder(threshold=0.7)
result = cluster_builder.build_clusters(entities)

print(f"Found {len(result.clusters)} clusters:")
for i, cluster in enumerate(result.clusters):
    names = [e['name'] for e in cluster.entities]
    print(f"  Cluster {i+1}: {names}")

55. 实体合并

一旦找到重复项，EntityMerger 就会合并它们。你需要选择一个合并策略：

KEEP_FIRST / KEEP_LAST：基于顺序。
KEEP_MOST_COMPLETE：保留数据最多（属性 + 关系）的实体。
KEEP_HIGHEST_CONFIDENCE：使用内部置信度分数。
MERGE_ALL：合并所有内容（数组被拼接，冲突通过投票消解）。

# 导入用于实体合并的特定类
from semantica.deduplication import EntityMerger, MergeStrategy, MergeStrategyManager, MergeOperation, MergeResult

merger = EntityMerger()

# 我们将使用 'KEEP_MOST_COMPLETE' 策略
# 这确保我们不会丢失来自更丰富实体的有价值信息
merge_ops = merger.merge_duplicates(
    entities, 
    strategy=MergeStrategy.KEEP_MOST_COMPLETE
)

print(f"Performed {len(merge_ops)} merge operations.")

print("\n--- Merged Results ---")
for op in merge_ops:
    final_ent = op.merged_entity
    original_count = len(op.source_entities)
    print(f"Merged {original_count} entities into: '{final_ent['name']}'")
    print(f"  - Final Properties: {final_ent['properties']}")
    print(f"  - Final Relationships: {len(final_ent.get('relationships', []))}")

66. 完整工作流

让我们将其封装为一个干净的函数，接收脏数据并返回干净数据。

def deduplicate_dataset(raw_entities):
    print("1. Detecting duplicates...")
    # 步骤 1：检测
    detector = DuplicateDetector(similarity_threshold=0.75)
    # 如果我们只想合并，可以跳过显式的检测调用，
    # 因为 EntityMerger 内部会调用检测，但手动执行可以进行检查。

    print("2. Merging entities...")
    # 步骤 2：合并
    merger = EntityMerger()
    ops = merger.merge_duplicates(raw_entities, strategy=MergeStrategy.KEEP_MOST_COMPLETE)

    # 收集所有最终 ID 以查看剩余内容
    merged_entities = [op.merged_entity for op in ops]

    # 找出未参与任何合并的实体（单例）
    merged_ids = set()
    for op in ops:
        for source in op.source_entities:
            merged_ids.add(source['id'])

    singletons = [e for e in raw_entities if e['id'] not in merged_ids]

    final_dataset = merged_entities + singletons
    return final_dataset

# 运行工作流
clean_data = deduplicate_dataset(entities)

print(f"\nOriginal Size: {len(entities)}")
print(f"Cleaned Size:  {len(clean_data)}")
print("\nFinal Entity Names:")
for e in clean_data:
    print(f"  - {e['name']}")

7小结

你已经学习了如何： 1. 导入必要的去重类。 2. 计算实体之间的相似度。 3. 使用可配置阈值检测重复。 4. 聚类相似实体。 5. 合并重复项，得到干净、规范化的数据集。

该模块对于维护高质量的知识图谱至关重要，尤其是在从多个可能混乱的来源摄取数据时。