第 4 章 · 01 冲突检测与消解

第 4 章 · 01 冲突检测与消解

本节摘要：装配管线第④段的前半。conflicts 模块（10 个 Python 文件、4951 行）回答的问题是「张三的年龄，年报里是 35、新闻稿里是 53，图谱信哪个」。本节先用 ConflictType/Conflict 界定五类冲突，再精读 conflict_detector.py（1423 行）的通用检测算法——实体对齐（按 ID 分组）→属性比对（收集各来源的值）→冲突判定（去重后多于一个值即冲突），以及严重度与置信度公式；然后过 ConflictResolver 的七种消解策略（投票、来源可信度加权、时间新近等）与属性级路由；最后用 SourceTracker 把每个被采纳的值钉回来源文档与页码，供审计回放。

内容来源：semantica/conflicts/（conflict_detector.py 1423 行、conflict_resolver.py 574 行、source_tracker.py 684 行、conflict_analyzer.py 615 行、investigation_guide.py 482 行、methods.py、registry.py、config.py、conflicts_provenance.py、__init__.py）

⚠️ 注意：ConflictDetector 自带的 resolve_conflicts()（第 1346—1418 行）只是 auto_resolve=True 开关下的统计骨架——源码注释坦承 "in a real system we would update the entity"，它并不把值写回实体；真正产出采纳值的是 conflict_resolver.py 的 ConflictResolver，把 ResolutionResult.resolved_value 回写图谱由调用方负责。另外 _resolve_by_recency 依赖 sources[i]["metadata"]["timestamp"]，来源不带时间戳时会静默退化为取第一个值，生产上要留意。

学习目标

1. 区分五类冲突（value/type/relationship/temporal/logical），并能各举一个实例。
2. 掌握「实体对齐→属性比对→冲突判定」三步算法：按实体 ID 分组、组内至少两个来源才比对、set(str(v)) 去重后多于一个值即冲突。
3. 读懂 _calculate_severity 与 _calculate_conflict_confidence：critical 字段、数值跨度、来源置信度与值多样性各扮演什么角色。
4. 会选七种 ResolutionStrategy 消解策略，并会用 set_resolution_rule 做属性级路由。
5. 理解 SourceTracker 的「实体→属性→值→来源」四级溯源与来源可信度机制，说清冲突治理为何卡在抽取之后、正式建图之前。

一、冲突长什么样：五类与一个数据结构

多源数据汇入同一张图，矛盾不是意外而是必然：同一实体被两份文档各抽取一次，属性值就可能不一致。conflict_detector.py 第 68—92 行用两个类型把冲突世界固定下来：

class ConflictType(str, Enum):
    VALUE_CONFLICT = "value_conflict"                # 数值/文本冲突：年龄 35 vs 53
    TYPE_CONFLICT = "type_conflict"                  # 类型冲突：同一 ID 既是 Person 又是 Organization
    RELATIONSHIP_CONFLICT = "relationship_conflict"  # 关系冲突：同一关系的类型/属性两说
    TEMPORAL_CONFLICT = "temporal_conflict"          # 时间冲突：founded_year 两说
    LOGICAL_CONFLICT = "logical_conflict"            # 逻辑冲突：互斥类型并存

@dataclass
class Conflict:
    conflict_id: str                                 # 如 "entity_1_age_conflict"
    conflict_type: ConflictType
    entity_id: Optional[str] = None
    property_name: Optional[str] = None
    relationship_id: Optional[str] = None
    conflicting_values: List[Any] = field(default_factory=list)  # [35, 53, 35]
    sources: List[Dict[str, Any]] = field(default_factory=list)  # 与值按序对应
    confidence: float = 1.0                          # 「这确实是个冲突」的置信度
    severity: str = "medium"  # low, medium, high, critical
    recommended_action: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
​

关键是 conflicting_values 与 sources 按序对应：第 i 个值来自第 i 条来源记录。这让冲突天然可审计——不是一句「age 冲突」，而是「annual_report.pdf 第 3 页说 35（置信度 0.9），news.docx 说 53（置信度 0.6）」。

二、检测算法：实体对齐→属性比对→冲突判定

所有 detect_* 方法共享同一骨架。以最核心的 detect_value_conflicts（第 136—322 行）为例，剥掉进度跟踪噪音后只剩三步：

# ① 实体对齐：同一实体（同 ID）的多来源记录聚成一组
entity_groups: Dict[str, List[Dict[str, Any]]] = {}
for entity in entities:
    entity_id = entity.get("id") or entity.get("entity_id")
    if not entity_id:
        continue                                     # 无 ID 无法对齐，跳过
    if entity_type and entity.get("type") != entity_type:
        continue
    entity_groups[entity_id].append(entity)

# ② 属性比对：组内收集该属性的值，顺手登记来源
for entity_id, entity_list in entity_groups.items():
    if len(entity_list) < 2:
        continue  # Need at least 2 sources to have conflict
    values, sources = [], []
    for entity in entity_list:
        if property_name in entity:
            values.append(entity[property_name])
            if self.track_provenance:                # 来源登记，细节见第五节
                self.source_tracker.track_property_source(
                    entity_id, property_name, entity[property_name],
                    SourceReference(document=entity.get("source", "unknown"),
                                    page=entity.get("page"),
                                    confidence=entity.get("confidence", 1.0), ...))

    # ③ 冲突判定：值的字符串化集合多于一个元素，即冲突
    unique_values = list(set(str(v) for v in values if v is not None))
    if len(unique_values) > 1:
        conflict = Conflict(
            conflict_id=f"{entity_id}_{property_name}_conflict",
            conflict_type=ConflictType.VALUE_CONFLICT,
            conflicting_values=values, sources=sources,
            confidence=self._calculate_conflict_confidence(values, sources),
            severity=self._calculate_severity(property_name, values),
            recommended_action=self._recommend_action(property_name, values))
        self.detected_conflicts[conflict.conflict_id] = conflict
​

三个细节值得划线。其一，「对齐」用 ID 精确相等——跨来源 ID 本身不一致的场景（zhangsan vs 张三）要先靠第 4.2 节的去重合并统一 ID，这就是「先去重、后查冲突」的管线次序。其二，判定用 str(v) 字符串化后比较，35 与 "35" 视为相同——宽松但实用。其三，来源登记发生在检测时，Conflict.sources 只是快照，完整溯源留在 SourceTracker 里。

其余四类检测只是换了比对对象：detect_type_conflicts（第 655 行起）比对 type/entity_type；detect_relationship_conflicts（第 339 行起）按 rel_id = f"{source_id}_{target_id}_{type}" 分组，比对 type/properties/confidence；detect_temporal_conflicts（第 818 行起）只扫时间属性白名单 ["founded", "founded_year", "established", "created", "timestamp", "date", "start_date", "end_date"]；detect_logical_conflicts（第 1025—1203 行）不比值而是查互斥类型表（第 1047—1053 行的 incompatible_types：Person 与 Organization/Company/Institution 互斥、Location 与前两者互斥），两两比对组内类型，命中即产出 severity="critical"、confidence=1.0 的冲突——逻辑互斥没有商量余地。总入口 detect_conflicts（第 1205—1344 行）按 method 分发到上述方法；method="all" 时依次跑值冲突（配置了 conflict_fields 就逐字段，否则实体级全字段扫描）、类型、时间、逻辑四轮，methods.detect_conflicts 是它的注册表包装。

三、打分：严重度、置信度与建议动作

检测出来只是第一步，还要回答「先处理哪个」。三个小函数（第 558—612 行）给出工程化答案：

def _calculate_conflict_confidence(self, values, sources) -> float:
    if not sources:
        return 0.5
    avg_confidence = sum(s.get("confidence", 0.5) for s in sources) / len(sources)
    value_diversity = (len(set(str(v) for v in values)) / len(values)
                       if values else 0)
    return min(1.0, avg_confidence * (1 + value_diversity))   # 封顶 1.0

def _calculate_severity(self, property_name, values) -> str:
    critical_fields = ["id", "name", "type", "founded_year", "revenue"]
    if property_name.lower() in critical_fields:
        return "critical"
    try:
        numeric_values = [float(v) for v in values if v is not None]
        if numeric_values:
            value_range = max(numeric_values) - min(numeric_values)
            if value_range > 1000:                # 数值冲突跨度大
                return "high"
    except (ValueError, TypeError):
        pass
    return "medium"
​

读法：置信度回答「这真是冲突吗」——来源平均置信度越高、值越五花八门，越可能是真冲突而非抽取噪声；严重度回答「多要命」——落在 critical_fields（id/name/type/founded_year/revenue）直接判 critical，数值跨度超过 1000（比如营收两说差 3000 万）判 high，其余 medium。_recommend_action 按值个数给人话建议：恰好两个值时「比对原始文档，取更新或更权威来源」，更多时「建议人工复核」。get_conflict_report()（第 614—653 行）把 detected_conflicts 汇成 total/by_type/by_severity/conflicts 四键报告，是治理看板的直接数据源。

四、消解：七种策略与属性级路由

冲突交给 conflict_resolver.py 的 ConflictResolver。resolve_conflict() 调度前先做属性级路由（第 214—219 行）：调用时不指定策略就走默认（voting），但若曾用 set_resolution_rule("entity_1", "revenue", "credibility_weighted") 登记过规则，引擎会按 f"{entity_id}.{property_name}" 查 resolution_rules，该实体该属性自动改走指定策略——「关键财务字段必须按来源权威度消解」这类领域约定由此落成配置。七种策略与两个代表性实现（第 70—79、355—435 行）：

class ResolutionStrategy(str, Enum):
    VOTING = "voting"                    # 多数投票：最高频值胜出
    CREDIBILITY_WEIGHTED = "credibility_weighted"  # 来源可信度加权
    MOST_RECENT = "most_recent"          # 时间新近：最新时间戳的值胜出
    FIRST_SEEN = "first_seen"            # 先到先得
    HIGHEST_CONFIDENCE = "highest_confidence"      # 来源置信度最高者胜出
    MANUAL_REVIEW = "manual_review"      # 旗标人工复核
    EXPERT_REVIEW = "expert_review"      # 旗标专家复核

def _resolve_by_voting(self, conflict: Conflict) -> ResolutionResult:
    value_counts = Counter(conflict.conflicting_values)
    most_common_value, count = value_counts.most_common(1)[0]
    total_votes = len(conflict.conflicting_values)
    confidence = count / total_votes if total_votes > 0 else 0.0   # 得票率即置信度
    return ResolutionResult(..., resolved_value=most_common_value,
        resolution_notes=f"Resolved by voting: {count}/{total_votes} votes...")

def _resolve_by_credibility(self, conflict: Conflict) -> ResolutionResult:
    value_weights: Dict[Any, float] = {}
    for i, value in enumerate(conflict.conflicting_values):
        source = conflict.sources[i] if i < len(conflict.sources) else {}
        document = source.get("document", "unknown")
        credibility = self.source_tracker.get_source_credibility(document)  # 默认 0.5
        weight = source.get("confidence", 0.5) * credibility    # 抽取置信度 × 来源可信度
        value_weights[value] = value_weights.get(value, 0.0) + weight
    resolved_value = max(value_weights.items(), key=lambda x: x[1])[0]
    confidence = value_weights[resolved_value] / sum(value_weights.values())
​

投票是「民主」：三个来源两个说 35，35 胜出，得票率 2/3 直接充当结果置信度；可信度加权是「加权民主」：每个值的票重＝该来源这次抽取的置信度乘该文档的历史可信度——同一票，官方年报比小报重。_resolve_by_recency（第 437—468 行）翻每个来源的 metadata.timestamp 取最大者（无时间戳则退化，见开头注意）；MANUAL_REVIEW/EXPERT_REVIEW 返回 resolved=False 并打 requires_manual_review 标记，把决定权交还给人——能识别「不该自动决定」同样是策略。每次消解都追加进 resolution_history，result.metadata 带全冲突上下文，构成审计链。

五、来源追踪：让每个值都能「出示证件」

消解的票重来自 SourceTracker（source_tracker.py）。核心结构（第 74—95、125—135 行）：

@dataclass
class SourceReference:
    document: str                  # 来源文档标识
    page: Optional[int] = None     # 页码
    section: Optional[str] = None  # 章节
    line: Optional[int] = None
    timestamp: Optional[datetime] = None
    confidence: float = 1.0        # 该来源此次抽取的置信度
    metadata: Dict[str, Any] = field(default_factory=dict)

# SourceTracker 内部：实体 → 属性 → PropertySource(值 + 来源列表)
self.entity_sources: Dict[str, Dict[str, PropertySource]] = defaultdict(dict)
self.source_credibility: Dict[str, float] = {}    # 文档 → 可信度分
​

检测阶段每次 track_property_source（第 186—239 行）都以 (document, page, section) 三元组去重后把 SourceReference 追加进对应属性的来源列表。可信度两头开放：set_source_credibility("annual_report.pdf", 0.9) 手工设定（越界抛 ValidationError），get_source_credibility 查不到返回中性默认 0.5——未登记的来源不奖不罚。审计侧两个出口：generate_source_report()（第 409 行起）输出每实体每属性的全部来源及可信度；get_traceability_chain(entity_id, property_name)（第 470—526 行）输出线性溯源链，每环含 document/page/section/confidence/timestamp——监管问「这个 age=35 哪来的」，一条链到底。至于消解结果回写：ResolutionResult.resolved_value 是「该写回去的值」，回写动作由调用方执行（建图前统一替换实体属性，或经第 4.2 节实体合并落图）；ConflictDetector.resolve_conflicts 的 auto_resolve 仅作统计占位，见本节开头的注意。

💡 装配要点：本段心智模型是「检测三步＋消解七策＋溯源四级」。检测端五类冲突共享「按 ID 对齐分组→组内≥2 来源→值集合基数>1 即冲突」骨架，严重度看字段敏感级与数值跨度；消解端默认投票，关键属性用 set_resolution_rule 路由到可信度加权或时间新近，拿不准的旗标人工；溯源端 SourceTracker 从检测时逐值登记来源，可信度分参与票重。接进管线的方式：第 3 章抽取产物（或第 4.2 节去重后的实体表）喂给 detect_conflicts(entities, method="all")，冲突过 ConflictResolver 消解，resolved_value 回写后再流入第 5 章建图推理——脏图推不出干净结论。

本节要点回顾

• 五类冲突：value（年龄两说）、type（既是 Person 又是 Organization）、relationship、temporal（founded_year 两说）、logical（互斥类型并存，critical＋confidence 1.0）。
• 检测三步：按 id/entity_id 分组对齐、组内至少两个来源、set(str(v)) 去重后多于一个值即冲突；detect_conflicts(method=...) 是分发总入口。
• 打分公式：置信度＝来源平均置信度×(1＋值多样性) 封顶 1.0；严重度：critical 字段直判 critical、数值跨度>1000 判 high、默认 medium。
• 七种消解策略：投票（得票率即置信度）、可信度加权（抽取置信度×来源可信度作票重）、most_recent/first_seen/highest_confidence、manual/expert_review（resolved=False 旗标）；set_resolution_rule 按 实体.属性 路由；来源追踪：SourceReference（document/page/section/confidence/timestamp）、entity_sources 四级嵌套、可信度默认 0.5、get_traceability_chain 出审计链。

下一节：02 去重与实体合并——阻塞分桶把 O(n²) 比对压到桶内、语义相似度做精判、union-find 聚组、五种合并策略与 merged_from 双来源溯源。