How to Implement Conflict Resolution Strategies for Contradictory Facts in Semantica

Semantica provides a built-in conflict-resolution pipeline with ConflictDetector, ConflictResolver, and configurable per-property strategies that resolve contradictory facts during knowledge graph construction.

Conflict resolution for contradictory facts is essential when building knowledge graphs from heterogeneous data sources. Semantica (available at github.com/semantica-agi/semantica) ships with a complete conflict-resolution framework that detects, reports, and resolves contradictions through pluggable strategies. This guide shows how to implement conflict resolution strategies using the actual source components found in semantica/conflicts/ and semantica/kg/.

Core Conflict-Resolution Components

Semantica's architecture separates detection from resolution, enabling flexible integration into any ingestion workflow.

Component	File Location	Purpose
ConflictDetector	[semantica/conflicts/conflict_detector.py](https://github.com/semantica-agi/semantica/blob/main/semantica/conflicts/conflict_detector.py)	Scans entities/triples for duplicate IDs, value mismatches, and schema violations
ConflictResolver	[semantica/conflicts/conflict_resolver.py](https://github.com/semantica-agi/semantica/blob/main/semantica/conflicts/conflict_resolver.py)	Applies configurable resolution strategies and returns resolved graphs
GraphBuilder	[semantica/kg/graph_builder.py](https://github.com/semantica-agi/semantica/blob/main/semantica/kg/graph_builder.py)	Orchestrates full pipeline with resolve_conflicts=True
SourceTracker	[semantica/conflicts/conflicts_provenance.py](https://github.com/semantica-agi/semantica/blob/main/semantica/conflicts/conflicts_provenance.py)	Records audit trail of all resolution decisions

Built-in Resolution Strategies

The ConflictResolver supports multiple conflict resolution strategies out of the box:

• keep_first — Retains the first observed value (chronological or ingestion order)
• keep_last — Retains the most recently observed value
• longer_string — Keeps the longer textual value (useful for descriptions and abstracts)
• merge_all — Aggregates all distinct values into a set or list
• custom_callable — Any user-defined function fn(a, b) -> resolved_value

Strategies can be assigned globally via default_strategy or per-predicate via property_strategies.

Detecting Conflicts in Your Data

Before resolution, the ConflictDetector identifies three primary conflict types:

1. Duplicate identifiers — Multiple entities share the same @id
2. Value mismatch — Same entity, same predicate, different values
3. Schema violation — Property values incompatible with declared datatypes

The detector emits typed conflict objects (Conflict, ValueConflict) that the resolver consumes.

Implementing Automatic Conflict Resolution

Basic Pipeline with Default Strategies

Enable automatic conflict resolution by passing resolve_conflicts=True to GraphBuilder:

from semantica.kg.graph_builder import GraphBuilder

# Default detector and resolver are instantiated automatically

builder = GraphBuilder(resolve_conflicts=True)

raw_entities = [
    {"@id": "E1", "schema:name": "Apple", "schema:price": 1.00},
    {"@id": "E1", "schema:name": "Apple Inc.", "schema:price": 0.95},
]

kg = builder.build(raw_entities)
print(kg.get_entity("E1"))

The default resolver uses keep_first for all properties unless overridden.

Custom Per-Property Strategies

Implement domain-specific conflict resolution by mapping predicates to strategies:

from semantica.conflicts.conflict_resolver import ConflictResolver
from semantica.kg.graph_builder import GraphBuilder

def keep_longest(a, b):
    """Custom callable: prefer longer string values."""
    return a if len(str(a)) >= len(str(b)) else b

def merge_tag_lists(a, b):
    """Custom callable: deduplicated union of two lists."""
    return list(set(a + b))

resolver = ConflictResolver(
    default_strategy="keep_last",
    property_strategies={
        "schema:name": keep_longest,
        "schema:description": "keep_last",
        "schema:tags": merge_tag_lists,
        "schema:rating": lambda a, b: max(a, b),
    },
)

builder = GraphBuilder(resolve_conflicts=True, conflict_resolver=resolver)

entities = [
    {
        "@id": "E2",
        "schema:name": "Short",
        "schema:tags": ["alpha"],
        "schema:rating": 3
    },
    {
        "@id": "E2",
        "schema:name": "A Much Longer Title",
        "schema:tags": ["beta"],
        "schema:rating": 4
    },
]

kg = builder.build(entities)
entity = kg.get_entity("E2")

# name: "A Much Longer Title" (keep_longest)

# tags: ["alpha", "beta"] (merged)

# rating: 4 (max)

Provenance Tracking and Auditability

Every resolution decision is traceable through SourceTrackerWithUnifiedBackend in conflicts_provenance.py:

from semantica.conflicts.conflicts_provenance import SourceTrackerWithUnifiedBackend

tracker = SourceTrackerWithUnifiedBackend(agent_id="conflict_resolver")

# Records are created automatically during resolution

records = tracker.get_records_for_entity("E2")

for r in records:
    print(f"{r.predicate}: {r.chosen_value} (strategy: {r.strategy})")

The provenance log enables:

• Compliance auditing (GDPR, data lineage requirements)
• Debugging unexpected resolution outcomes
• Reproducibility of knowledge graph construction

Fine-Grained Control with PropertyMergeRule

For advanced scenarios, PropertyMergeRule objects in semantica/deduplication/merge_rules.py define merge behavior at the predicate level. These rules integrate with the conflict resolver to handle complex data types, cardinality constraints, and domain-specific merge logic.

Complete Example: Multi-Source Ingestion

from semantica.conflicts.conflict_resolver import ConflictResolver
from semantica.conflicts.conflicts_provenance import SourceTrackerWithUnifiedBackend
from semantica.kg.graph_builder import GraphBuilder

# Domain-specific resolution policy

resolver = ConflictResolver(
    default_strategy="keep_first",
    property_strategies={
        "schema:name": "longer_string",
        "schema:dateModified": "keep_last",
        "schema:author": "merge_all",
    },
)

builder = GraphBuilder(
    resolve_conflicts=True,
    conflict_resolver=resolver,
)

# Simulating data from three sources with contradictions

source_a = {"@id": "Book1", "schema:name": "Deep Learning", "schema:pages": 780}
source_b = {"@id": "Book1", "schema:name": "Deep Learning: A Comprehensive Guide", "schema:dateModified": "2024-01-15"}
source_c = {"@id": "Book1", "schema:name": "DL Handbook", "schema:author": ["A. Author", "B. Coauthor"]}

kg = builder.build([source_a, source_b, source_c])

Summary

• Enable resolution — Pass resolve_conflicts=True to GraphBuilder for automatic pipeline integration
• Choose strategies — Use built-in strategies (keep_first, keep_last, longer_string, merge_all) or supply custom callables
• Customize per property — Map predicate IRIs to specific strategies via property_strategies in ConflictResolver
• Maintain provenance — Audit all decisions through SourceTrackerWithUnifiedBackend for compliance and debugging
• Reference implementation — See notebook 17 Conflict Detection and Resolution (link) for step-by-step examples

Frequently Asked Questions

What types of conflicts can Semantica detect?

Semantica's ConflictDetector identifies duplicate identifiers (same @id on multiple entities), value mismatches (conflicting values for the same predicate), and schema violations (datatype incompatibilities). These are emitted as typed conflict objects consumed by the resolver.

Can I use different strategies for different properties?

Yes. The ConflictResolver accepts a property_strategies dictionary that maps predicate IRIs to either strategy keywords or custom callables. Unmapped properties fall back to default_strategy.

Is the resolution process auditable?

Yes. Every resolution writes a provenance record via SourceTrackerWithUnifiedBackend in conflicts_provenance.py. Each record captures the entity ID, predicate, chosen value, rejected value, and strategy applied, enabling full audit trails for compliance.

How do I implement a completely custom resolution algorithm?

Pass any callable with signature fn(a, b) -> resolved_value to property_strategies or use as default_strategy. The callable receives the conflicting values and returns the resolved result. For multi-value aggregation, your callable can return lists, sets, or domain-specific structures.

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