How to Detect and Resolve Conflicting Facts from Multiple Data Sources in Semantica

Semantica provides a decoupled pipeline using ConflictDetector to identify five conflict types across entity properties and ConflictResolver to apply configurable strategies including voting, credibility-weighted scoring, and recency-based selection.

When building knowledge graphs from heterogeneous sources, the same real-world entity often arrives with contradictory attributes. Semantica's semantica-agi/semantica repository ships with a production-ready conflict detection and resolution system that handles this at scale. The architecture cleanly separates detection (finding discrepancies) from resolution (choosing canonical values), with full audit trails and progress tracking for large-scale ingestion jobs.

Conflict Detection: Finding Data Discrepancies

The detection layer is implemented in semantica/conflicts/conflict_detector.py. The ConflictDetector class scans raw entity graphs and flags mismatches across five dimensions:

Conflict Type	Trigger Condition	Detection Method
VALUE_CONFLICT	Same property holds different values across sources	detect_value_conflicts()
TYPE_CONFLICT	Incompatible type assignments for an entity	detect_type_conflicts()
RELATIONSHIP_CONFLICT	Contradictory relationship declarations	detect_relationship_conflicts()
TEMPORAL_CONFLICT	Timestamp inconsistencies or ordering violations	detect_temporal_conflicts()
LOGICAL_CONFLICT	Domain rule violations (custom predicates)	detect_logical_conflicts()

Entity Grouping and Property Comparison

All detectors follow a common pattern. First, records sharing an id or entity_id are collated into groups (lines 71–74 of conflict_detector.py):

entity_groups: Dict[str, List[Dict[str, Any]]] = {}

For each property of interest, the detector builds a set of distinct values. When multiple values exist, a Conflict object is instantiated with provenance, confidence scoring via _calculate_conflict_confidence(), and severity rating via _calculate_severity() (lines 71–88):

if len(unique_values) > 1:
    conflict = Conflict(
        conflict_id=f"{entity_id}_{property_name}_conflict",
        conflict_type=ConflictType.VALUE_CONFLICT,
        ...
    )

Conflict Reporting

After scanning completes, get_conflict_report() aggregates counts by type and severity into a serializable dictionary (lines 214–252). All operations report progress through semantica.utils.progress_tracker, enabling real-time monitoring even for million-entity datasets.

Conflict Resolution: Choosing Canonical Values

Resolution is handled by ConflictResolver in semantica/conflicts/conflict_resolver.py. The ResolutionStrategy enumeration supports five approaches:

Strategy	Selection Mechanism	Best For
Voting	Counter.most_common — most frequent value wins	High-volume, no source reliability data
Credibility-Weighted	Vote × SourceTracker.get_source_credibility() × source confidence	Known source quality tiers
Most Recent / First Seen	Timestamp comparison from source metadata	Rapidly changing entities
Highest Confidence	Max confidence field value	Sources self-report quality
Manual / Expert Review	Flags for human triage, recorded in metadata	High-stakes or ambiguous cases

Single and Batch Resolution

Resolve one conflict with resolve_conflict(conflict, strategy) or many with resolve_conflicts(conflicts, strategy). Each call:

• Creates a progress tracker entry
• Normalizes the strategy (default: voting)
• Returns a ResolutionResult containing:
  • resolved: success boolean
  • resolved_value: chosen canonical value
  • confidence: decision confidence score
  • resolution_strategy: strategy applied
  • metadata: conflict type, entity ID, property name, audit trail

The resolver maintains an internal resolution_history for full traceability.

Property-Level Strategy Overrides

Per-property rules allow fine-grained control without hardcoding:

resolver.set_resolution_rule("entity_42", "name", "credibility_weighted")
resolver.set_resolution_rule("entity_42", "founded_year", "most_recent")

Unspecified properties fall back to the global default.

End-to-End Implementation

Detecting Value Conflicts

from semantica.conflicts import ConflictDetector

entities = [
    {"id": "c123", "name": "Acme Corp", "source": "source_A"},
    {"id": "c123", "name": "Acme Corporation", "source": "source_B"},
    {"id": "c124", "name": "Globex", "source": "source_A"},
]

detector = ConflictDetector()
name_conflicts = detector.detect_value_conflicts(entities, "name")

# Inspect results

print(name_conflicts[0].conflict_id)          # → c123_name_conflict

print(name_conflicts[0].conflicting_values)  # → ["Acme Corp", "Acme Corporation"]

Source: ConflictDetector.detect_value_conflicts — lines 71–88 of conflict_detector.py

Resolving with Source Credibility

from semantica.conflicts import ConflictResolver

resolver = ConflictResolver()
resolver.set_resolution_rule("c123", "name", "credibility_weighted")
result = resolver.resolve_conflict(name_conflicts[0])

print(result.resolved_value)      # chosen based on source credibility

print(result.resolution_strategy) # → credibility_weighted

Source: ConflictResolver.resolve_conflict — lines 81–89 of conflict_resolver.py

Batch Processing with Mixed Strategies

resolver = ConflictResolver()
resolver.set_resolution_rule("c123", "founded_year", "most_recent")

# Defaults to voting for conflicts without explicit rules

resolved = resolver.resolve_conflicts(all_conflicts)

for r in resolved:
    if r.resolved:
        # Apply canonical value back to knowledge graph

        kg.update(r.metadata["entity_id"], r.metadata["property"], r.resolved_value)

Extending the System

Adding Custom Conflict Types

1. Extend the ConflictType enumeration
2. Implement detect_<custom>_conflicts() following existing patterns
3. Register in detect_conflicts() dispatcher

Adding Custom Resolution Strategies

1. Subclass ConflictResolver
2. Add resolution method returning ResolutionResult
3. Map strategy name in _normalize_strategy()

Both extension paths inherit progress tracking and logging automatically.

Core Source Files

File	Purpose
semantica/conflicts/conflict_detector.py	Detection engine, entity grouping, conflict object construction
semantica/conflicts/conflict_resolver.py	Strategy implementation, rule handling, resolution history
semantica/conflicts/source_tracker.py	Provenance tracking for credibility scoring
semantica/utils/progress_tracker.py	Shared progress infrastructure

Summary

• Detection and resolution are decoupled — persist Conflict objects and resolve later with different strategies
• Five conflict types cover values, types, relationships, timestamps, and logical rules
• Five resolution strategies range from simple voting to credibility-weighted and manual review
• Progress tracking is built into both components for large-scale operations
• Per-property rules allow granular control without code changes
• Full audit trails via resolution_history and ResolutionResult.metadata

Frequently Asked Questions

What conflict types does Semantica detect?

Semantica detects five conflict types: VALUE_CONFLICT (contradictory property values), TYPE_CONFLICT (incompatible type assignments), RELATIONSHIP_CONFLICT (conflicting relationship declarations), TEMPORAL_CONFLICT (timestamp inconsistencies), and LOGICAL_CONFLICT (domain rule violations). Each type has a dedicated detection method in ConflictDetector.

How does credibility-weighted resolution work?

Credibility-weighted resolution multiplies each vote by three factors: source frequency (Counter), source credibility from SourceTracker.get_source_credibility(), and the source-declared confidence score. This prioritizes values from historically reliable sources that also express high confidence in their data.

Can I resolve conflicts without running detection first?

Yes. The components are fully decoupled. You can construct Conflict objects manually or load serialized conflicts, then pass them to ConflictResolver. This enables scenarios like third-party conflict ingestion or reprocessing with updated strategies without re-scanning source data.

What happens when no resolution strategy is specified?

ConflictResolver defaults to voting (ResolutionStrategy.VOTING). You can override this globally or set per-property rules using set_resolution_rule(entity_id, property_name, strategy) before calling resolve_conflicts().

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