socratic-conflict 0.1.1

Conflict detection and resolution system for multi-agent workflows with consensus building and decision versioning.

Socratic Conflict - Conflict Detection & Resolution System

A comprehensive system for detecting, analyzing, and resolving conflicts between multiple agents in collaborative AI systems.

Features

Conflict Detection

• Automatic Detection: Identifies data conflicts, decision conflicts, and workflow conflicts
• Severity Assessment: Calculates severity based on number of conflicting agents and disagreement magnitude
• Context Tracking: Maintains context information for workflow-specific conflicts
• History Management: Complete conflict history with retrieval and filtering

Resolution Strategies

Choose from 5 configurable resolution approaches:

1. VotingStrategy - Simple majority voting among proposals
2. ConsensusStrategy - Selects highest confidence proposal
3. WeightedStrategy - Scores proposals by agent weight × confidence
4. PriorityStrategy - Uses predefined priority rules to select winners
5. HybridStrategy - Combines multiple strategies for sophisticated conflict resolution

Consensus Algorithms

5 different algorithms for reaching agreements:

1. MajorityConsensus - >50% support required
2. UnanimousConsensus - 100% agreement required
3. SupermajorityConsensus - Configurable threshold (default 2/3)
4. RankedChoiceConsensus - Confidence-based ranking and selection
5. QuorumConsensus - Minimum participation requirement + majority rule

Decision Management

• Versioning: Track all decision versions for a conflict
• Reversion: Revert decisions with reason tracking
• Statistics: Comprehensive conflict statistics by type, severity, strategy

Installation

pip install socratic-conflict

Quick Start

Basic Conflict Detection

from socratic_conflict import ConflictDetector, Proposal, Conflict

# Create detector
detector = ConflictDetector()

# Detect data conflict
conflict = detector.detect_data_conflict(
    field_name="priority",
    values={"agent1": "high", "agent2": "low"},
    agents=["agent1", "agent2"]
)

print(f"Conflict detected: {conflict.conflict_type}")
print(f"Severity: {conflict.severity}")

Resolve with Voting

from socratic_conflict import VotingStrategy

strategy = VotingStrategy()
resolution = strategy.resolve(conflict)

print(f"Recommended proposal: {resolution.recommended_proposal_id}")
print(f"Confidence: {resolution.confidence:.2%}")

Use Consensus Algorithm

from socratic_conflict.consensus import SupermajorityConsensus

algorithm = SupermajorityConsensus(threshold=0.75)  # 75% required
proposal_id, confidence = algorithm.reach_consensus(conflict)

print(f"Consensus reached on: {proposal_id}")
print(f"Support: {confidence:.2%}")

Track Decision History

from socratic_conflict.history import HistoryTracker

tracker = HistoryTracker()

# Add conflicts and track decisions
tracker.add_conflict(conflict)
tracker.add_resolution(resolution)

# Get full history
history = tracker.get_conflict_history(conflict.conflict_id)
print(f"Resolution strategy used: {history['resolutions'][0].strategy}")

# Get statistics
stats = tracker.get_statistics()
print(f"Total conflicts: {stats['total_conflicts']}")
print(f"Resolution rate: {stats['resolution_rate']:.1%}")

Architecture

socratic_conflict/
├── core/              # Data models
│   └── conflict.py    # Proposal, Conflict, Resolution, ConflictDecision
├── detection/         # Detection engine
│   └── detector.py    # Conflict detection and severity calculation
├── resolution/        # Resolution strategies
│   └── strategies.py  # 5 configurable resolution strategies
├── consensus/         # Consensus algorithms
│   └── algorithms.py  # 5 consensus algorithms
└── history/           # History tracking
    └── tracker.py     # Versioning and retrieval

Core Models

Proposal

@dataclass
class Proposal:
    title: str                          # Human-readable proposal title
    source_agent: str                   # Agent that proposed it
    confidence: float = 0.0             # Confidence score (0.0-1.0)
    description: str = ""               # Detailed description
    rationale: str = ""                 # Why this proposal

Conflict

@dataclass
class Conflict:
    conflict_type: str                  # "data", "decision", "workflow"
    severity: str = "medium"            # "low", "medium", "high", "critical"
    proposals: List[Proposal]           # Competing proposals
    related_agents: List[str]           # Agents involved
    context: Dict[str, Any]             # Additional context

Resolution

@dataclass
class Resolution:
    conflict_id: str                    # Associated conflict
    strategy: str                       # Strategy used for resolution
    recommended_proposal_id: str        # Winning proposal
    confidence: float                   # Resolution confidence (0.0-1.0)

Resolution Strategies in Detail

VotingStrategy

Simple majority voting - each agent gets one vote, proposal with most votes wins.

strategy = VotingStrategy()
# Weights: equal vote per agent regardless of confidence

ConsensusStrategy

Selects the proposal with highest confidence score from source agent.

strategy = ConsensusStrategy()
# Selection: argmax(proposal.confidence)

WeightedStrategy

Combines agent weight with proposal confidence.

strategy = WeightedStrategy(weights={
    "expert_agent": 0.9,      # High weight for experts
    "junior_agent": 0.3       # Lower weight for juniors
})
# Score: agent_weight × proposal.confidence

PriorityStrategy

Uses predefined priority rules - highest priority agent's proposal wins.

strategy = PriorityStrategy(priority_rules={
    "admin": 10,
    "manager": 5,
    "user": 1
})

HybridStrategy

Combines multiple strategies and selects best result.

strategy = HybridStrategy(strategies=[
    VotingStrategy(),
    ConsensusStrategy(),
    WeightedStrategy(weights={"expert": 0.9})
])
# Runs all strategies, returns highest confidence result

Consensus Algorithms in Detail

MajorityConsensus

Classic majority rule - >50% support.

algorithm = MajorityConsensus()
# Threshold: 0.5 (>50%)

UnanimousConsensus

All agents must agree.

algorithm = UnanimousConsensus()
# Threshold: 1.0 (100%)

SupermajorityConsensus

Configurable threshold, useful for important decisions.

algorithm = SupermajorityConsensus(threshold=0.67)  # 2/3 majority
algorithm = SupermajorityConsensus(threshold=0.75)  # 3/4 supermajority

RankedChoiceConsensus

Uses confidence scores for ranking - more nuanced than simple voting.

algorithm = RankedChoiceConsensus()
# Considers proposal confidence scores in ranking

QuorumConsensus

Requires minimum participation before majority applies.

algorithm = QuorumConsensus(quorum_fraction=0.75)  # 75% must participate
# Combines quorum requirement with >50% majority of participants

Testing

Run tests with coverage:

pytest tests/unit/ --cov=socratic_conflict --cov-report=term-missing

Current test coverage:

• Core models: 100%
• Detection: 98%
• Strategies: 90%
• Overall: 69%

Quality Gates

All code passes:

• ✅ Black formatting (100% compliant)
• ✅ Ruff linting (0 issues)
• ✅ MyPy type checking (strict mode)
• ✅ Python 3.9-3.12 compatibility

Use Cases

Multi-Agent Negotiation

# Agents propose different solutions
proposals = [
    Proposal("Algorithm A", source_agent="ml_expert", confidence=0.9),
    Proposal("Algorithm B", source_agent="data_engineer", confidence=0.7),
]

conflict = Conflict(
    conflict_type="decision",
    proposals=proposals,
    related_agents=["ml_expert", "data_engineer"]
)

# Resolve with weighted strategy favoring ML expertise
strategy = WeightedStrategy(weights={
    "ml_expert": 0.8,
    "data_engineer": 0.5
})

resolution = strategy.resolve(conflict)

Data Validation Conflicts

# Different validation agents disagree
conflict = detector.detect_data_conflict(
    field_name="email_format",
    values={
        "strict_validator": "invalid@format",
        "lenient_validator": "valid@format"
    },
    agents=["strict_validator", "lenient_validator"]
)

# Use unanimous consensus for critical data
algorithm = UnanimousConsensus()
result = algorithm.reach_consensus(conflict)  # Will fail (no consensus)

Workflow Execution Conflicts

# Different execution paths proposed
conflict = detector.detect_workflow_conflict(
    workflow_id="data_pipeline",
    conflicting_steps=[
        {"step_id": "s1", "agent": "path_planner_1", "action": "process"},
        {"step_id": "s2", "agent": "path_planner_2", "action": "skip"}
    ]
)

# Track decision history
tracker.add_conflict(conflict)
decision = ConflictDecision(
    conflict_id=conflict.conflict_id,
    chosen_proposal_id=resolution.recommended_proposal_id,
    decided_by="workflow_orchestrator"
)
tracker.add_decision(decision)

Integration with Other Socratic Packages

Socratic Conflict integrates naturally with:

• Socratic Agents: Detect/resolve conflicts between agent proposals
• Socratic Workflow: Handle conflicting execution paths
• Socratic Knowledge: Resolve knowledge base consistency conflicts

Performance

• Conflict detection: O(n) where n = number of agents
• Resolution strategies: O(n·m) where m = number of proposals
• Consensus algorithms: O(n·log n) average case
• Memory: O(n) per conflict tracked

For typical use cases (2-10 agents, 2-5 proposals):

• Detection: <1ms
• Resolution: <5ms
• Consensus: <5ms

License

MIT License - See LICENSE file for details

Contributing

Contributions welcome! Please ensure:

1. Tests pass (pytest tests/)
2. Code is formatted (black src/ tests/)
3. Linting passes (ruff check src/ tests/)
4. Type checking passes (mypy src/)

Changelog

v0.1.0 (March 16, 2026)

• ✅ Initial MVP release
• ✅ Conflict detection engine
• ✅ 5 resolution strategies
• ✅ 5 consensus algorithms
• ✅ Decision versioning and history
• ✅ 33 comprehensive unit tests
• ✅ 69% test coverage
• ✅ Full type checking (MyPy strict mode)

Authors

Created as part of the Socrates Ecosystem - a comprehensive AI development framework.

Status

✅ Production Ready - Currently used in production environments ⏳ Phase 5 - Openclaw skill and LangChain integration in development