第 3 章 · 02 GraphBuilder:构建知识图谱 ★

第 3 章 · 02 GraphBuilder:构建知识图谱 ★

本节摘要：本节装配管线第⑤段，全书的装配高点。semantica/kg/ 共 25 个文件 13295 行，核心是 GraphBuilder（graph_builder.py，1380 行）：吃下「实体＋关系」或原始文本，吐出 {"entities": [...], "relationships": [...], "metadata": {...}} 三键图谱字典。本节逐段读它的主流程——输入归一（_process_item 的鸭子类型识别）、实体消解（EntityResolver 三步法）、关系端点重映射（_remap_relationship_endpoints）、可选的冲突检测与 GraphStore 持久化；再看 merge_entities/resolve_conflicts/enable_temporal 等构造开关的含义，以及构建完成后的图谱统计从哪来。

内容来源：原项目源码 semantica/kg/graph_builder.py、entity_resolver.py、knowledge_graph.py

⚠️ 注意：GraphBuilder.build 的输入格式宽容到近乎「猜」——dict、Entity 对象、Relation 对象、纯文本字符串甚至 {"text": ...} 字典都能吃。宽容是双刃剑：格式不对不报错而是尽力解释，适合快速原型；但生产环境建议喂规范化字典，并在返回的 metadata 里核对统计数字是否符合预期。

学习目标

1. 说出 kg 模块的文件版图与 GraphBuilder 在其中的位置。
2. 理解构造参数四大开关：merge_entities/resolve_conflicts/enable_temporal/graph_store。
3. 读懂 build() 主流程：归一→（可选抽取）→消解→端点重映射→建结构→持久化。
4. 理解 _process_item 的鸭子类型识别：字符串/Entity/Relation/dict 四路分发。
5. 读懂 EntityResolver 的检测→合并→回填三步法与 merged_from 溯源字段。
6. 理解关系端点重映射为什么必须发生在实体合并之后，以及图谱统计字典的构成。

一、kg 模块版图与 GraphBuilder 的构造开关

semantica/kg/ 的 13295 行里，GraphBuilder 只占 1380 行。先看版图再定位：temporal_query.py（1765 行，双时态查询）、graph_builder.py（1380，建图）、kg_provenance.py（1085，图谱溯源）、methods.py（914，方法调度）、community_detector.py（904）与 centrality_calculator.py（739）是下一节的图分析、entity_resolver.py（292，实体消解）、knowledge_graph.py（46，规范数据类型）——本节聚焦前四个中的 GraphBuilder 与 entity_resolver。

构造函数（graph_builder.py:57）声明了四个关键开关：

def __init__(self, merge_entities=False, entity_resolution_strategy="fuzzy",
             resolve_conflicts=True, enable_temporal=False,
             temporal_granularity="day", track_history=False,
             version_snapshots=False, graph_store=None, **kwargs):
    ...
    # merge_entities=True 时才构造 EntityResolver
    if self.merge_entities:
        self.entity_resolver = EntityResolver(
            strategy=self.entity_resolution_strategy, **entity_resolution_config)
    # resolve_conflicts 默认 True:复用第 4 章的 ConflictDetector
    if self.resolve_conflicts:
        from ..conflicts.conflict_detector import ConflictDetector
        self.conflict_detector = ConflictDetector(**conflict_detection_config)
​

开关语义：merge_entities（默认 False）决定是否做实体消解，策略 fuzzy/exact/ml-based；resolve_conflicts（默认 True）决定建图时是否顺带跑冲突检测——注意 kg 模块直接复用 conflicts 模块的检测器，管线第④段的治理件在这里被二次使用；enable_temporal 开双时态特性；graph_store 传入持久化后端（第 8 章的 Neo4j 等）。类 docstring 还特意解释了 _extractor_cache（graph_builder.py:96）：NER 抽取器构造时要加载 spaCy 模型，所以按 (kind, method) 缓存复用，多文档构建不再反复付 120ms 的模型加载成本。

二、_process_item：四路分发的输入归一

build() 吃什么都行，秘密在 _process_item（graph_builder.py:144）的四路分发：

def _process_item(self, item, all_entities, all_relationships, **options):
    if isinstance(item, str):
        # 路线1:纯文本 → 现场抽取
        self._extract_from_text(item, all_entities, all_relationships, **options)
        return
    if hasattr(item, "text") and (hasattr(item, "label") or hasattr(item, "type")):
        # 路线2:Entity 对象 → 转 dict
        entity_dict = {"id": getattr(item, "id", getattr(item, "entity_id", item.text)),
                       "name": item.text,
                       "type": getattr(item, "label", getattr(item, "type", "UNKNOWN")),
                       "confidence": getattr(item, "confidence", 1.0), ...}
        all_entities.append(entity_dict)
    elif hasattr(item, "subject") and hasattr(item, "predicate") and hasattr(item, "object"):
        # 路线3:Relation 对象 → 转关系 dict
        rel_dict = {"source": subj_id, "target": obj_id, "type": item.predicate, ...}
        all_relationships.append(rel_dict)
    elif isinstance(item, dict):
        # 路线4:字典 → 字段名归一(source_id/subject→source 等)+递归展开
        if "source_id" in item and "source" not in item:
            item["source"] = item["source_id"]
        ...
        if "entities" in item:      # entities/relationships 键 → 递归处理每个成员
            ...
        elif "source" in item and "target" in item:
            all_relationships.append(item)      # 像关系的字典 → 关系
        elif "id" in item or "name" in item:
            all_entities.append(item)           # 像实体的字典 → 实体
​

字段名归一这张网织得很密：source_id→source、subject→source、target_id→target、object→target，甚至 source 是对象时再取它的 id/text。build()（graph_builder.py:406）还做了批量优化：当输入是「全部为 dict 格式的实体列表」时走快路径直接归一 append，否则才逐个 _process_item（graph_builder.py:579 的 is_dict_format 判定）。

三、EntityResolver：同一实体的不同写法合并

实体消解（graph_builder.py:733）在建图主流程中位置靠前——先消解实体、再修关系：

resolved_entities = resolver_to_use.resolve_entities(all_entities)
# 日志:Resolved to N unique entities — "Entity resolution complete:
#        1500 -> 1233 unique entities"
​

EntityResolver.resolve_entities（entity_resolver.py:92）三步走：

# 步骤1:检测重复组(DuplicateDetector + 相似度阈值)
duplicate_groups = self._detect_duplicate_groups(entities)
# 步骤2:每组合并为规范实体
for group in duplicate_groups:
    merge_operations = self.entity_merger.merge_duplicates(group.entities, **self.config)
    for operation in merge_operations:
        merged_entities.append(operation.merged_entity)
        # 步骤3的准备:记录已被合并的源实体 id
        for source_entity in operation.source_entities:
            processed_entity_ids.add(self._get_entity_id(source_entity))
# 步骤3:未参与合并的实体原样保留
for entity in entities:
    if entity_id not in processed_entity_ids:
        merged_entities.append(entity)
​

它内部组合的正是第 4 章的 DuplicateDetector＋EntityMerger——kg 与 deduplication 两个模块在这里握手。策略上 fuzzy（默认，阈值 0.7）与 semantic 走相似度检测，exact 只做大小写折叠后的精确分组（entity_resolver.py:222 的 _detect_duplicate_groups）。合并后的规范实体带 merged_from 字段记录所有源 id——这既是溯源，也是下一步的钥匙。

四、端点重映射与图谱统计

实体合并有个连锁问题：关系是在消解之前收集的，边可能还指着已被合并掉的旧实体 id。_remap_relationship_endpoints（graph_builder.py:265）专门善后：

# 建端点映射:规范 id→规范 id,所有 merged_from 的旧 id→规范 id
endpoint_map[canonical_id] = canonical_id
for source_id in entity.get("merged_from") or []:
    endpoint_map[source_id] = canonical_id
# 重写每条关系的 source/target
for relationship in relationships:
    for endpoint in ("source", "target"):
        canonical_id = endpoint_map.get(endpoint_id)
        if canonical_id is not None and canonical_id != endpoint_id:
            relationship[endpoint] = canonical_id
​

没有这一步，「Apple Inc.」和「Apple」合并后，指向旧 id 的 works_for 边会变成悬空边。graph_builder.py:761 的触发条件也精细：只有消解确实产生了合并（has_merged_entities）才重映射。悬空风险还有一道保险：输入关系数量与最终图内关系数量对比，全丢则打 warning（graph_builder.py:769）。

主流程的收尾（graph_builder.py:779 起）产出图谱字典并做可选持久化：

graph = {
    "entities": resolved_entities,
    "relationships": all_relationships,
    "metadata": {
        "num_entities": len(resolved_entities),      # 图谱统计
        "num_relationships": len(all_relationships),
        "temporal_enabled": self.enable_temporal,
        "timestamp": self._get_timestamp(),
        "entity_resolution_applied": resolver_to_use is not None,
    },
}
if self.graph_store:      # 持久化:add_nodes + add_edges
    node_count = self.graph_store.add_nodes(resolved_entities)
    edge_count = self.graph_store.add_edges(formatted_edges)   # source_id/target_id/type
if self.conflict_detector:   # 默认开启:对实体属性跑冲突检测+消解
    detected_conflicts = self.conflict_detector.detect_conflicts(graph["entities"])
​

输出的 metadata 就是「构建后的图谱统计」：节点数、边数、是否启用时态、构建时间戳、是否做过实体消解。若传入了 graph_store，节点与边（边被格式化成 source_id/target_id/type/properties 四键）落库并统计耗时；若开着 resolve_conflicts，实体属性冲突在此被检测与消解——第 4 章的机制在这里预演。另有 46 行的 knowledge_graph.py：KnowledgeGraph dataclass（entities/relationships/metadata 三字段加 __len__/__bool__），作为可视化与导出组件消费的规范类型，避免裸 dict 在模块间传递。

💡 装配要点：本节装上全书的装配高点。四个记忆点：① GraphBuilder 的宽容输入靠 _process_item 四路分发（字符串→现场抽取、对象→转 dict、dict→字段归一）；② 实体消解三步法＝检测→合并→回填，merged_from 同时承担溯源与端点重映射依据；③ 顺序铁律——先消解实体、再重映射关系端点，否则合并制造悬空边；④ 输出三键字典＋metadata 统计，resolve_conflicts=True 让冲突治理在建图时自动补跑一遍。

本节要点回顾

• kg 模块 25 文件 13295 行；GraphBuilder 1380 行是建图核心，temporal_query（1765）与图分析留给后两节。
• 构造开关：merge_entities（默认关）、resolve_conflicts（默认开，复用 conflicts 模块）、enable_temporal、graph_store。
• _extractor_cache 按 (kind, method) 缓存抽取器，spaCy 模型只加载一次。
• _process_item 四路分发＋字段名归一网（source_id/subject/object→source/target）。
• EntityResolver：DuplicateDetector＋EntityMerger 的组合，fuzzy 默认阈值 0.7，exact 只折叠大小写。
• 端点重映射在合并后执行；输出 {"entities","relationships","metadata"}，metadata 含图谱统计；graph_store 可选落库。

下一节：03 图分析:中心性/社区/链接预测——图谱建好了，用 networkx/scipy 三件套验证它的可用性：关键实体排序、社群发现、潜在关系推断。