How to Implement Conflict Detection Across Overlapping Knowledge Sources in Semantica

Use the ConflictDetector class from semantica.conflicts to scan entities from multiple sources, group them by ID, and identify value, type, temporal, logical, and relationship conflicts while tracking full provenance via SourceTracker.

Implementing conflict detection across overlapping knowledge sources is essential when building knowledge graphs from heterogeneous data. The Semantica framework provides a complete pipeline for identifying, reporting, and resolving conflicts between sources that describe the same entities. This article walks through the core components, detection flow, and practical implementation using the actual source code from the semantica-agi/semantica repository.

Core Components of the Conflict Detection Pipeline

The Semantica conflict detection system consists of three primary classes working together:

Component	Role	Key File
ConflictDetector	Scans entities and discovers conflicts across multiple dimensions	semantica/conflicts/conflict_detector.py
SourceTracker	Records provenance (document, confidence, timestamps) for traceability	semantica/conflicts/source_tracker.py
ConflictResolver	Applies resolution strategies (voting, weighted, recency) to conflicts	semantica/conflicts/conflict_resolver.py

Data Model Requirements

Entities must be plain Python dictionaries with a unique identifier. Source metadata enables conflict detection across overlapping knowledge sources by preserving where each value originated:

{
    "id": "company_123",
    "name": "Acme Corp",
    "revenue": 1500000,
    "source": "financial_report_q1.pdf",
    "confidence": 0.92,
    "metadata": {"timestamp": "2024-03-01"},
}

The SourceTracker captures this provenance through track_property_source (lines 48-61 in source_tracker.py), storing document references, page numbers, confidence scores, and timestamps for every property value.

Step-by-Step Conflict Detection Flow

Step 1: Initialize Source Tracking

Create a SourceTracker instance to maintain provenance chains:

from semantica.conflicts import SourceTracker

tracker = SourceTracker()

Step 2: Create the Conflict Detector

Pass the tracker to ConflictDetector for integrated provenance capture:

from semantica.conflicts import ConflictDetector

detector = ConflictDetector(source_tracker=tracker)

Step 3: Run Detection on Combined Sources

Feed all entities from overlapping sources into detect_conflicts:

conflicts = detector.detect_conflicts(
    entities=all_entities,  # List[Dict] from all sources

    method="all",           # Detect every conflict type

)

Step 4: Internal Processing Pipeline

The ConflictDetector executes this sequence (see conflict_detector.py):

1. Group by entity ID — builds entity_groups dictionary (lines 71-79)
2. Collect property values and sources — creates SourceReference objects for each value (lines 44-61 in detect_value_conflicts)
3. Identify conflicts — triggered when unique_values > 1 (lines 71-79)
4. Calculate confidence and severity — via _calculate_conflict_confidence (lines 558-590) and _calculate_severity (lines 775-802)
5. Store Conflict objects — appended to self.detected_conflicts (lines 86-89)

Step 5: Generate Reports

Obtain a structured summary with counts, types, and detailed conflict listings:

report = detector.get_conflict_report()

Types of Conflicts Detected

Conflict Type	Description	Detection Method Location
Value conflict	Same property holds different values across sources	detect_value_conflicts (lines 36-117)
Type conflict	Entity classified with different types	detect_type_conflicts (lines 555-610)
Temporal conflict	Dates or timestamps disagree	detect_temporal_conflicts (lines 818-887)
Logical conflict	Incompatible type combinations (e.g., Person vs. Organization)	detect_logical_conflicts (lines 1002-1074)
Relationship conflict	Relationship properties differ	detect_relationship_conflicts (lines 361-423)
Entity-wide conflict	Value detection run across all entity properties	detect_entity_conflicts (lines 483-545)

Practical Implementation Examples

Example 1: Detect Value Conflicts Across Two HR Systems

from semantica.conflicts import ConflictDetector, SourceTracker

# Source A: Internal HR system

source_a = [
    {"id": "person_1", "name": "Alice", "age": 30,
     "source": "HR_system_A.csv", "confidence": 0.9},
    {"id": "person_2", "name": "Bob", "age": 45,
     "source": "HR_system_A.csv", "confidence": 0.85},
]

# Source B: Contractor database

source_b = [
    {"id": "person_1", "name": "Alicia", "age": 30,
     "source": "HR_system_B.xlsx", "confidence": 0.95},
    {"id": "person_2", "name": "Bob", "age": 46,
     "source": "HR_system_B.xlsx", "confidence": 0.88},
]

# Combine and detect

all_entities = source_a + source_b
tracker = SourceTracker()
detector = ConflictDetector(source_tracker=tracker)

conflicts = detector.detect_conflicts(all_entities, method="value")
print(f"Detected {len(conflicts)} conflicts")
for c in conflicts:
    print(f"- {c.entity_id}.{c.property_name}: {c.conflicting_values}")

# Generate report

report = detector.get_conflict_report()
print(report["by_type"])

Output:

Detected 2 conflicts
- person_1.name: ['Alice', 'Alicia']
- person_2.age: [45, 46]

Example 2: Automatic Conflict Resolution

from semantica.conflicts import ConflictResolver, ConflictDetector, SourceTracker

# Assume previous detection produced `conflicts`

tracker = SourceTracker()
detector = ConflictDetector(source_tracker=tracker)
conflicts = detector.detect_conflicts(all_entities, method="value")

resolver = ConflictResolver(default_strategy="voting")
resolution_results = resolver.resolve_conflicts(conflicts)

for r in resolution_results:
    print(f"{r.conflict_id} → {r.resolved_value} "
          f"({r.resolution_strategy}, confidence={r.confidence:.2f})")

Alternative strategies: credibility_weighted, most_recent, highest_confidence — see ResolutionStrategy enum in conflict_resolver.py.

Example 3: Traceability and Provenance Querying

conflict = conflicts[0]  # e.g., person_1 name conflict

# Retrieve all sources for this specific property

prop_source = tracker.get_property_sources(
    conflict.entity_id,
    conflict.property_name
)

print("Conflicting sources:")
for src in prop_source.sources:
    print(f"- {src.document} (page {src.page}, "
          f"confidence={src.confidence}, time={src.timestamp})")

# Generate full traceability chain

chain = tracker.generate_traceability_chain(
    entity_id=conflict.entity_id,
    property_name=conflict.property_name
)

The generate_traceability_chain method (lines 437-528 in source_tracker.py) produces an auditable path showing exactly which documents contributed each value.

Provenance Architecture: How Source Tracking Works

The SourceTracker maintains a mapping structure that enables conflict detection across overlapping knowledge sources to remain fully auditable:

• track_property_source (lines 48-61): Called automatically during detection to record origin metadata
• get_property_sources: Retrieves all contributing sources for any entity-property pair
• generate_traceability_chain (lines 437-528): Builds human-readable provenance chains

This architecture ensures that every detected conflict carries complete lineage information, supporting regulatory compliance and debugging.

Performance and Scalability Considerations

The ConflictDetector groups entities using dictionary lookups (O(1)) before performing pairwise value comparisons. For large datasets:

• Filter by entity type before detection using entity_type parameter
• Use specific detection methods (method="value" instead of "all") to reduce computation
• Leverage confidence thresholds in SourceTracker to exclude low-credibility sources early

Key Source Files Reference

File	Purpose	Key Lines
semantica/conflicts/conflict_detector.py	Core detection logic	36-117 (value), 555-610 (type), 818-887 (temporal)
semantica/conflicts/source_tracker.py	Provenance and traceability	48-61 (tracking), 437-528 (chains)
semantica/conflicts/conflict_resolver.py	Resolution strategies	Full ResolutionStrategy enum
tests/conflicts/test_conflicts.py	Usage patterns and validations	Complete test suite
tests/kg/test_kg.py	Knowledge graph integration	GraphBuilder integration examples

Summary

• Use ConflictDetector with SourceTracker to implement conflict detection across overlapping knowledge sources with full provenance preservation
• Initialize the pipeline in three steps: create tracker, instantiate detector, feed combined entity list
• Choose detection scope via method parameter: "value", "type", "temporal", "logical", "relationship", or "all"
• Access six conflict types through dedicated methods in conflict_detector.py, each with documented line ranges
• Generate traceability chains using SourceTracker.generate_traceability_chain for audit requirements
• Apply automatic resolution through ConflictResolver with pluggable strategies (voting, credibility-weighted, recency)
• Reference test files in tests/conflicts/ for validated implementation patterns

Frequently Asked Questions

How does the ConflictDetector group entities from different sources?

The detector groups entities by their id or entity_id field using a dictionary structure (entity_groups) at lines 71-79 of conflict_detector.py. All entities sharing the same identifier are collected into a list, then their properties are compared pairwise to identify discrepancies.

What information does a Conflict object contain?

Each Conflict object stores: the entity ID, property name, conflicting values as a list, conflict type classification, severity score, confidence calculation, and a list of source dictionaries documenting where each value originated. This complete provenance enables both manual review and automated resolution.

Can I detect conflicts without using SourceTracker?

Yes, SourceTracker is optional—ConflictDetector will create a default tracker if none is provided. However, omitting explicit source tracking loses provenance information, making traceability chains unavailable and reducing resolution strategy options for credibility-weighted decisions.

How do I choose between resolution strategies?

Select based on your data characteristics: voting works well with many equivalent sources; credibility_weighted prioritizes high-confidence sources; most_recent resolves temporal data conflicts; highest_confidence selects single best source. Pass your choice to ConflictResolver(default_strategy="...") or per-call via strategy parameter.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}