冲突检测与冲突消解 · Semantica Cookbook

1概览

在现代数据流水线中，尤其是构建知识图谱的流水线，数据往往从多个异构来源（例如内部数据库、第三方 API、网页抓取）摄取。差异不可避免。

Semantica 冲突消解模块（semantica.conflicts）为管理这些数据不一致提供了健壮的框架。它旨在确保你的下游应用只消费高质量、已协调一致的数据。

什么算作"冲突"？

当同一实体的多条记录在某个字段上不一致时，就会发生冲突。
Semantica 通常假设每条记录是一个字典，包含：
id（或 entity_id）：所描述实体的稳定标识符
一个或多个属性（例如 name、birth_date、department）
source：值的来源（db、scrape、api、file）
可选的 timestamp：观察到该值的时间
该模块是来源感知的：它可以记录哪些来源贡献了哪些值，然后使用投票或可信度等策略进行消解。

实用 API 说明（避免常见的不匹配）

当你只想检查一个字段时，使用 ConflictDetector.detect_value_conflicts(entities, property_name=...)。
当你想要更广泛的扫描（值/类型/时序等）时，使用 ConflictDetector.detect_conflicts(entities)。
如果你从知识图谱字典（通过 GraphBuilder 构建）开始，请通过 kg.get("entities", []) 提取实体。

关键能力

多维冲突检测
- 值冲突：同一属性的不同值（例如 "Google" 与 "Google Inc."）。
- 类型冲突：数据类型不匹配（例如字符串与整数）。
- 时序冲突：时间顺序不一致（例如 start_date 晚于 end_date）。
溯源与来源追踪
- 细粒度追踪：将每个属性值追溯到其具体的来源文档、页面或 API 调用。
- 可信度评分：为来源分配信任分数（例如内部 HR 数据库为 0.95，网页抓取为 0.60）。
自动化消解策略
- 投票：少数服从多数（适用于多个等权重的来源）。
- 可信度加权：来自更高信任来源的值覆盖其他值。
- 时效性：最新的数据点胜出。
- 专家评审：标记复杂冲突以供人工介入。
调查与审计
- 调查指南：自动生成分步指南，供人类分析师解决棘手的冲突。
- 审计轨迹：记录每个冲突是如何被消解的，以满足合规要求。

2安装

确保你的环境中已安装 Semantica：

# 安装 semantica
pip install semantica

如果你在 Semantica 仓库内运行此 notebook，建议使用可编辑安装（这样代码的改动会立即生效）：
pip install -e .
如果你使用托管的 notebook 环境，%pip install semantica 通常比 !pip install ... 更可靠，因为它会安装到当前内核中。

# 安装 semantica 包
!pip install -q semantica

# 导入 json 与 datetime 工具，用于处理记录数据和时间戳
import json
from datetime import datetime

3步骤 1：模拟多来源数据

为了演示该框架，我们将模拟一个涉及员工数据的真实场景。

场景： 我们从三个不同的来源收到了员工 001 的记录：

HR 数据库：高度可信的内部来源。
LinkedIn 抓取：不太可靠的外部来源。
公共目录：过时的公共 API。

记录结构（Semantica 所期望的）：

每条记录是一个描述同一实体（id：emp_001）的字典。
每条记录包含一个 source 键，以便将冲突归因到具体来源。
timestamp 让你可以应用基于时间的消解策略（例如最新者胜出）。

冲突： * birth_date：公共目录列出了不同的年份。 * department：LinkedIn 使用了更具体的名称（"Software Engineering"），而 HR 数据库中是通用的 "Engineering"。

# 1. 定义来源元数据
sources_metadata = {
    "hr_db": {"credibility": 0.95, "type": "internal_database"},
    "linkedin_scrape": {"credibility": 0.60, "type": "web_scrape"},
    "public_dir": {"credibility": 0.40, "type": "public_api"}
}

# 2. 定义来自这些来源的实体记录
entity_records = [
    {
        "id": "emp_001",
        "name": "John Doe",
        "birth_date": "1980-05-15",
        "department": "Engineering",
        "source": "hr_db",
        "timestamp": "2023-01-01T10:00:00"
    },
    {
        "id": "emp_001",
        "name": "Jonathan Doe",
        "birth_date": "1980-05-15",
        "department": "Software Engineering",
        "source": "linkedin_scrape",
        "timestamp": "2023-06-15T14:30:00"
    },
    {
        "id": "emp_001",
        "name": "John Doe",
        "birth_date": "1982-05-15",  # 冲突：年份不同
        "department": "Engineering",
        "source": "public_dir",
        "timestamp": "2022-12-01T09:00:00"
    }
]

print(f"Loaded {len(entity_records)} records for Employee 001")

4步骤 2：注册并追踪来源

在能够基于信任有效消解冲突之前，我们必须使用 SourceTracker 注册我们的来源。

SourceTracker 充当以下内容的中央注册表：

可信度分数：你对来源的信任程度。
元数据：有用的上下文（例如来源类型、记录系统与抓取）。

如何看待可信度分数：

将分数用作相对排序（确切的小数不如排名重要）。
从简单开始：internal_db > vendor_api > web_scrape。
之后使用分析重新审视分数（例如"哪些来源经常出错？"）。

我们遍历模拟的来源并注册它们。

# 使用 SourceTracker 注册来源及其可信度分数
from semantica.conflicts import SourceTracker

# 创建来源追踪器实例
source_tracker = SourceTracker()

print("Registering sources...")
# 遍历所有来源并注册其类型与可信度分数
for source_id, metadata in sources_metadata.items():
    source_tracker.register_source(
        source_id=source_id,
        source_type=metadata["type"],
        credibility_score=metadata["credibility"]
    )
    print(f"  - Registered '{source_id}' with credibility {metadata['credibility']}")

5步骤 3：检测冲突

我们使用 ConflictDetector 扫描记录以发现差异。

该检测器很灵活，可以配置为检查：

特定属性：只检查关键字段，如 birth_date。
实体级扫描：扫描某个实体的多个（或全部）属性。

你会得到什么：

一个 Conflict 对象列表。
你通常会检查的有用字段：
conflict_type（例如 value_conflict）
entity_id、property_name
conflicting_values 和 sources
severity 和 confidence

这里，我们显式检查 birth_date 和 department。

from semantica.conflicts import ConflictDetector

# 使用我们已填充的来源追踪器初始化检测器
detector = ConflictDetector(source_tracker=source_tracker)

conflicts = []

# 1. 检查 birth_date
dob_conflicts = detector.detect_value_conflicts(entity_records, "birth_date")
conflicts.extend(dob_conflicts)

# 2. 检查 department
dept_conflicts = detector.detect_value_conflicts(entity_records, "department")
conflicts.extend(dept_conflicts)

print(f"Detected {len(conflicts)} conflicts:")
for conflict in conflicts:
    print(f"- {conflict.conflict_type.value}: {conflict.property_name} for {conflict.entity_id}")
    print(f"  Values: {conflict.conflicting_values}")
    print(f"  Severity: {conflict.severity}")
    print("--- ")

6步骤 4：分析冲突模式

在处理大型数据集时，单个冲突不如系统性模式重要。ConflictAnalyzer 帮助回答如下问题：

"是否某个特定来源导致了大多数冲突？"
"冲突是否集中在某个特定的实体类型或属性上？"
"冲突严重程度和冲突类型的分布如何？"

如何在流水线中使用：

运行分析以识别有噪声的来源。
使用结果调整可信度分数（步骤 2）或优化摄取/清洗规则。
随时间追踪趋势，以捕捉上游系统的回归。

# 分析冲突模式，识别有噪声的来源与分布
from semantica.conflicts import ConflictAnalyzer

# 创建分析器并分析已检测到的冲突
analyzer = ConflictAnalyzer()
analysis = analyzer.analyze_conflicts(conflicts)

# 输出按类型和严重程度的统计
print("Conflict Analysis Summary:")
print(f"Total Conflicts: {analysis['total_conflicts']}")
print(f"By Type: {analysis.get('by_type', {}).get('counts')}")
print(f"By Severity: {analysis.get('by_severity', {}).get('counts')}")

7步骤 5：消解冲突

这是决定信任哪个值的关键步骤。Semantica 提供灵活的消解策略。

策略 A：投票（少数服从多数）

该策略选择出现最频繁的值。它很简单，但将所有来源视为平等。

当你拥有许多质量相近的独立来源时最适用。
如果你有一个应始终占主导地位的单一记录系统，则不太适用。

策略 B：可信度加权

该策略根据每个值来源的 credibility 计算加权分数。

示例： * hr_db（0.95）说 "1980-05-15" * public_dir（0.40）说 "1982-05-15"

即使多个低质量来源就错误日期达成一致，高可信度来源也很可能胜出。

消解器返回什么：

一个消解结果列表，其中每一项通常包含：
是否已消解
选定的值（resolved_value）
一个置信度分数
元数据（如属性名）以支持审计轨迹

运行下一个单元格，比较投票与可信度加权的结果。

from semantica.conflicts import ConflictResolver

resolver = ConflictResolver()

# 关键：将来源追踪器关联到消解器。
# 这使消解器能够查找我们在步骤 2 中注册的可信度分数。
resolver.set_source_tracker(source_tracker)

print("--- Resolution: Voting ---")
voting_results = resolver.resolve_conflicts(conflicts, strategy="voting")
for res in voting_results:
    print(f"Property: {res.metadata.get('property_name'):<15} | Resolved Value: {res.resolved_value}")

print("\n--- Resolution: Credibility Weighted ---")
# 注意 HR 数据库的值如何因更高的可信度而被优先选择
credibility_results = resolver.resolve_conflicts(conflicts, strategy="credibility_weighted")
for res in credibility_results:
    print(f"Property: {res.metadata.get('property_name'):<15} | Resolved Value: {res.resolved_value} (Confidence: {res.confidence:.2f})")

8步骤 6：生成调查指南

并非所有冲突都能自动消解。高风险或低置信度的消解需要人工评审。

InvestigationGuideGenerator 为分析师生成结构化的"行动计划"，详细说明：

什么处于冲突中（实体 + 字段 + 相互竞争的值）。
谁参与其中（哪些来源产生了哪些值）。
如何验证正确的数据（分析师可以遵循的可操作步骤）。

何时生成指南：

低置信度的消解。
关键字段上的冲突（身份、法定名称、合规属性）。
任何你想在写回图谱之前设置人工介入检查点的时刻。

from semantica.conflicts import InvestigationGuideGenerator

guide_generator = InvestigationGuideGenerator()

# 为第一个冲突（birth_date）生成指南
guide = guide_generator.generate_guide(conflicts[0])

print(f"=== {guide.title} ===")
print(f"Summary: {guide.conflict_summary}\n")

print("Investigation Steps:")
for i, step in enumerate(guide.investigation_steps, 1):
    print(f"{i}. {step.description}")
    print(f"   Action: {step.action}")

print("\nRecommended Actions:")
for action in guide.recommended_actions:
    print(f"[ ] {action}")

9结论

你已经使用 Semantica 成功构建了一个冲突消解流水线！

我们取得的成果回顾：

模拟了带有真实分歧的多来源实体记录。
注册了带可信度分数的来源，以建立信任层级。
检测了特定属性的值冲突。
分析了冲突，以了解按类型和严重程度的分布。
使用投票和可信度加权策略消解了冲突。
生成了调查指南以支持人工评审。

在真实项目中的建议下一步：

与你的摄取层集成，使每个抽取出的值都包含 source 和（理想情况下）timestamp。
使用 ConflictDetector.detect_conflicts(...) 将检测扩展到值冲突之外。
存储消解结果和指南输出，为下游消费者构建审计轨迹。