How to Implement Conflict Detection and Resolution Using semantica.conflicts

The ConflictDetector class in semantica.conflicts provides built-in methods to scan knowledge graphs for contradictory information and resolve them using configurable strategies like keep-first, keep-last, or custom merge functions.

The semantica-agi/semantica repository includes a robust conflict management subsystem accessible through the semantica.conflicts module. This component automatically identifies contradictory entities within knowledge graphs and applies resolution policies to ensure data consistency before downstream processing.

Core Architecture of the Conflict Detection System

The ConflictDetector Class

The primary entry point for conflict detection and resolution is the ConflictDetector class defined in [semantica/conflicts/conflict_detector.py](https://github.com/semantica-agi/semantica/blob/main/semantica/conflicts/conflict_detector.py). This class implements three critical methods:

• detect_conflicts(entities) – Scans the entire graph for structural or property contradictions
• detect_value_conflicts(entities, field) – Focuses on specific attributes (e.g., name) to find disagreeing values
• resolve_conflicts(conflicts, strategy=...) – Applies resolution policies and returns a cleaned graph with metadata

The detector uses semantic hashing to compare entity signatures and maintains a property-merge registry to determine winning values during reconciliation.

Integration Points in the Pipeline

According to the source code, the conflict detector integrates at three key locations:

1. GraphBuilder – [semantica/kg/graph_builder.py](https://github.com/semantica-agi/semantica/blob/main/semantica/kg/graph_builder.py) instantiates ConflictDetector when resolve_conflicts=True
2. CLI Interface – [semantica/cli/commands.py](https://github.com/semantica-agi/semantica/blob/main/semantica/cli/commands.py) provides the semantica validate conflicts command for JSON diagnostics
3. Test Coverage – End-to-end validation in [tests/kg/test_kg.py](https://github.com/semantica-agi/semantica/blob/main/tests/kg/test_kg.py) and [tests/deduplication/test_cluster_merge_coverage.py](https://github.com/semantica-agi/semantica/blob/main/tests/deduplication/test_cluster_merge_coverage.py)

Detecting Conflicts in Knowledge Graphs

Using detect_conflicts()

To scan a knowledge graph for contradictions, instantiate the detector and pass your entity collection:

from semantica.conflicts import ConflictDetector

detector = ConflictDetector()
conflicts = detector.detect_conflicts(entities)

The entities parameter expects a list or dictionary of graph nodes. The method returns a list of Conflict objects containing entity IDs, conflict types, and offending values.

Targeted Field Analysis with detect_value_conflicts()

For targeted validation of specific attributes, use the field-specific method:

conflicts = detector.detect_value_conflicts(
    all_entities,  # List of dicts representing KG nodes

    field="name"
)

for c in conflicts:
    print(f"Entity {c.entity_id} conflict: {c.conflict_type}")

This approach is demonstrated in [tests/ingest/test_notebook_06.py](https://github.com/semantica-agi/semantica/blob/main/tests/ingest/test_notebook_06.py), which shows minimal mock-based detection scenarios.

Resolving Conflicts Automatically

Built-in Resolution Strategies

The resolve_conflicts method accepts a strategy parameter supporting multiple built-in policies:

• keep_first – Preserves the earliest encountered value (default behavior)
• keep_last – Retains the most recent value
• merge_strings – Concatenates differing string values

resolved = detector.resolve_conflicts(
    conflicts,
    strategy="keep_first"
)

print(f"Fixed {resolved['resolved_count']} conflicts")
clean_graph = resolved["graph"]

Implementing Custom Resolution Logic

You can inject domain-specific logic by passing a callable that receives two competing values and returns the chosen winner:

def longer_string(a, b):
    return a if len(a) > len(b) else b

resolved = detector.resolve_conflicts(
    conflicts,
    strategy=longer_string
)

This pattern is validated in [tests/deduplication/test_cluster_merge_coverage.py](https://github.com/semantica-agi/semantica/blob/main/tests/deduplication/test_cluster_merge_coverage.py), which demonstrates custom conflict_resolution callables for complex merge scenarios.

Integrating with GraphBuilder

Automatic Resolution Mode

The high-level GraphBuilder API provides seamless conflict handling through a boolean flag:

from semantica.kg.graph_builder import GraphBuilder

builder = GraphBuilder(resolve_conflicts=True)
kg = builder.build(source_triples)  # Returns pre-cleaned graph

When enabled, the builder automatically instantiates ConflictDetector, runs detection after graph assembly, and applies the default resolution strategy before returning the final knowledge graph.

Manual Conflict Handling Workflow

For pipelines requiring inspection or custom logic before resolution:

builder = GraphBuilder(resolve_conflicts=False)
raw_graph = builder.build(source_triples)

detector = ConflictDetector()
conflicts = detector.detect_conflicts(raw_graph)

# Inspect or log conflicts here before resolving

clean_graph = detector.resolve_conflicts(conflicts)["graph"]

The unit test test_build_with_conflict_resolution in [tests/kg/test_kg.py](https://github.com/semantica-agi/semantica/blob/main/tests/kg/test_kg.py) validates both automatic and manual integration patterns.

Practical Implementation Examples

Detecting name collisions across entities:

from semantica.conflicts import ConflictDetector

detector = ConflictDetector()
name_conflicts = detector.detect_value_conflicts(entity_list, field="name")

for conflict in name_conflicts:
    print(f"ID {conflict.entity_id}: {conflict.conflict_type}")

CLI-based validation:

semantica validate conflicts --json

This command forwards the in-memory graph to the detector and outputs a JSON summary of all contradictions, as implemented in [semantica/cli/commands.py](https://github.com/semantica-agi/semantica/blob/main/semantica/cli/commands.py).

Summary

• Primary Class: Import ConflictDetector from semantica.conflicts to access detection and resolution capabilities
• Detection Methods: Use detect_conflicts() for full-graph scanning or detect_value_conflicts() for field-specific analysis
• Resolution Strategies: Choose from built-in policies (keep_first, keep_last, merge_strings) or supply custom lambda functions
• Pipeline Integration: Enable automatic handling via GraphBuilder(resolve_conflicts=True) or manually orchestrate the detector workflow
• Source Locations: Core logic resides in semantica/conflicts/conflict_detector.py with integration points in semantica/kg/graph_builder.py

Frequently Asked Questions

How do I import and instantiate the ConflictDetector?

Import the class directly from the conflicts module using from semantica.conflicts import ConflictDetector, then instantiate with detector = ConflictDetector(). The constructor accepts optional configuration parameters for customizing detection heuristics and merge policies.

What resolution strategies are available by default?

The detector ships with three built-in strategies: keep_first (preserves the earliest value), keep_last (retains the most recent value), and merge_strings (concatenates differing strings). You can also pass any callable that accepts two arguments and returns the preferred value for custom domain logic.

Can I use conflict detection without the GraphBuilder?

Yes. While GraphBuilder provides convenient automatic integration, you can use ConflictDetector standalone by manually passing entity collections to detect_conflicts() or detect_value_conflicts(). This is useful for validating data before it enters the knowledge graph construction phase.

Where does the CLI store conflict reports?

The semantica validate conflicts command outputs diagnostics directly to stdout in JSON format. The implementation in semantica/cli/commands.py processes the in-memory graph through ConflictDetector and prints the summary, allowing you to pipe results to files or downstream processors as needed.

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