Create consensus_verification.py

consensus_verification.py

@@ -0,0 +1,85 @@
+"""
+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()