consensus_verification.py

"""
consensus_verification.py
A deterministic model for consensus emergence in multi-agent systems.
"""
import hashlib
import numpy as np
from dataclasses import dataclass
from typing import Dict, List

@dataclass(frozen=True)
class Claim:
    """Represents a proposition with a unique identifier and a clarity score."""
    identifier: str  # Hash of the canonical statement
    clarity: float   # 0.0 (ambiguous) to 1.0 (clear)

class ConsensusEngine:
    """
    Simulates a network of agents converging on belief states through
    iterative local updates.
    """
    def __init__(self, agent_count: int = 10):
        self.agent_count = agent_count
        self.agents = [{"id": i, "beliefs": {}} for i in range(agent_count)]

    def process(self, claims: List[Claim]) -> Dict[str, float]:
        """
        Executes the consensus process.
        Returns a mapping from each claim identifier to its final acceptance ratio.
        """
        # 1. Initialize agent beliefs stochastically based on claim clarity.
        for agent in self.agents:
            for claim in claims:
                accept_prob = 0.1 + (0.9 * claim.clarity)
                agent["beliefs"][claim.identifier] = 1 if np.random.random() < accept_prob else 0

        # 2. Iterate: agents adjust beliefs towards network average.
        for _ in range(5):
            network_avg = self._average_beliefs(claims)
            for agent in self.agents:
                for claim in claims:
                    current = agent["beliefs"][claim.identifier]
                    social_influence = network_avg[claim.identifier] - current
                    agent["beliefs"][claim.identifier] = np.clip(current + social_influence * 0.3, 0, 1)

        # 3. Calculate and return final consensus.
        final_state = {}
        for claim in claims:
            total_acceptance = sum(a["beliefs"][claim.identifier] for a in self.agents)
            final_state[claim.identifier] = round(total_acceptance / self.agent_count, 3)
        return final_state

    def _average_beliefs(self, claims: List[Claim]) -> Dict[str, float]:
        avg = {}
        for claim in claims:
            total = sum(a["beliefs"].get(claim.identifier, 0) for a in self.agents)
            avg[claim.identifier] = total / self.agent_count
        return avg

def run_demonstration():
    """
    Demonstrates the model's convergence on a small set of claims.
    """
    # Define claims with varying clarity.
    claim_data = [
        ("Water boils at 100°C at sea level.", 0.99),
        ("Democracy requires informed participation.", 0.75),
        ("Abstract concepts influence material outcomes.", 0.40),
    ]
    claims = [
        Claim(identifier=hashlib.sha256(text.encode()).hexdigest()[:16], clarity=score)
        for text, score in claim_data
    ]
    print("Claims:")
    for c in claims:
        print(f"  ID:{c.identifier} | Clarity:{c.clarity}")

    # Run the engine.
    engine = ConsensusEngine(agent_count=10)
    result = engine.process(claims)
    print("\nConsensus Results (Acceptance Ratio):")
    for claim_id, ratio in result.items():
        print(f"  {claim_id}: {ratio}")

if __name__ == "__main__":
    run_demonstration()