aishastr.live

Agentic Architecture Pattern: Consensus

Written by

admin

in


    """
Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms.
Supports multiple consensus algorithms: unanimous, majority, quorum-based.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ConsensusType(Enum):
    """Type of consensus mechanism."""
    UNANIMOUS = "unanimous"  # All must agree
    MAJORITY = "majority"  # >50% must agree
    SUPERMAJORITY = "supermajority"  # 2/3 must agree
    QUORUM = "quorum"  # Minimum number must participate


class ProposalStatus(Enum):
    """Status of consensus proposal."""
    PROPOSED = "proposed"
    DISCUSSING = "discussing"
    VOTING = "voting"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    NO_CONSENSUS = "no_consensus"


@dataclass
class Proposal:
    """Proposal for consensus."""
    proposal_id: str
    proposer: str
    content: str
    status: ProposalStatus = ProposalStatus.PROPOSED
    votes_for: Set[str] = field(default_factory=set)
    votes_against: Set[str] = field(default_factory=set)
    abstentions: Set[str] = field(default_factory=set)
    discussion_rounds: int = 0
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class ConsensusRound:
    """Round of consensus building."""
    round_id: int
    proposal_id: str
    participants: List[str]
    votes_cast: int
    consensus_reached: bool
    timestamp: datetime = field(default_factory=datetime.now)


class ConsensusAgent:
    """Agent participating in consensus."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.proposals_voted: Set[str] = set()
    
    def propose(self, proposal_id: str, content: str) -> Proposal:
        """Make a proposal."""
        proposal = Proposal(
            proposal_id=proposal_id,
            proposer=self.name,
            content=content
        )
        print(f"   šŸ“‹ {self.name} proposed: {content}")
        return proposal
    
    def evaluate_proposal(self, proposal: Proposal, context: str = "") -> str:
        """Evaluate a proposal and provide stance."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, evaluating a proposal for consensus. "
                      f"Consider the proposal carefully and provide your stance."),
            ("human", f"Proposal: {proposal.content}\n"
                     f"Proposer: {proposal.proposer}\n"
                     f"Context: {context}\n\n"
                     f"Do you support this proposal? Answer with 'support', 'oppose', or 'abstain' "
                     f"and explain why in 1-2 sentences.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:200]
    
    def vote(self, proposal: Proposal, decision: str):
        """Cast vote on proposal."""
        self.proposals_voted.add(proposal.proposal_id)
        
        if "support" in decision.lower():
            proposal.votes_for.add(self.name)
            print(f"   āœ… {self.name} voted FOR")
        elif "oppose" in decision.lower():
            proposal.votes_against.add(self.name)
            print(f"   āŒ {self.name} voted AGAINST")
        else:
            proposal.abstentions.add(self.name)
            print(f"   āšŖ {self.name} ABSTAINED")


class ConsensusPattern:
    """
    Consensus Pattern - Agreement through iterative consensus.
    
    Agents work together to reach consensus on proposals through
    discussion, voting, and iterative refinement.
    """
    
    def __init__(self, consensus_type: ConsensusType = ConsensusType.MAJORITY,
                 quorum: Optional[int] = None):
        self.consensus_type = consensus_type
        self.quorum = quorum
        self.agents: Dict[str, ConsensusAgent] = {}
        self.proposals: Dict[str, Proposal] = {}
        self.consensus_rounds: List[ConsensusRound] = []
    
    def register_agent(self, agent: ConsensusAgent):
        """Register agent in consensus group."""
        self.agents[agent.name] = agent
    
    def submit_proposal(self, proposal: Proposal):
        """Submit proposal for consensus."""
        self.proposals[proposal.proposal_id] = proposal
        proposal.status = ProposalStatus.DISCUSSING
    
    def conduct_consensus(self, proposal_id: str, max_rounds: int = 3) -> bool:
        """Conduct consensus building rounds."""
        if proposal_id not in self.proposals:
            return False
        
        proposal = self.proposals[proposal_id]
        proposal.status = ProposalStatus.VOTING
        
        print(f"\n   šŸ¤ Starting consensus on: {proposal.content}")
        print(f"   Consensus type: {self.consensus_type.value}")
        
        for round_num in range(max_rounds):
            print(f"\n   === Round {round_num + 1} ===")
            proposal.discussion_rounds += 1
            
            # Each agent evaluates and votes
            for agent_name, agent in self.agents.items():
                evaluation = agent.evaluate_proposal(proposal)
                agent.vote(proposal, evaluation)
            
            # Check consensus
            consensus_reached = self._check_consensus(proposal)
            
            # Record round
            consensus_round = ConsensusRound(
                round_id=round_num,
                proposal_id=proposal_id,
                participants=list(self.agents.keys()),
                votes_cast=len(proposal.votes_for) + len(proposal.votes_against),
                consensus_reached=consensus_reached
            )
            self.consensus_rounds.append(consensus_round)
            
            if consensus_reached:
                proposal.status = ProposalStatus.ACCEPTED
                print(f"\n   āœ… Consensus REACHED after {round_num + 1} rounds!")
                return True
        
        proposal.status = ProposalStatus.NO_CONSENSUS
        print(f"\n   āŒ No consensus after {max_rounds} rounds")
        return False
    
    def _check_consensus(self, proposal: Proposal) -> bool:
        """Check if consensus is reached."""
        total_agents = len(self.agents)
        votes_for = len(proposal.votes_for)
        votes_against = len(proposal.votes_against)
        votes_cast = votes_for + votes_against
        
        print(f"      Votes: {votes_for} for, {votes_against} against, "
              f"{len(proposal.abstentions)} abstain")
        
        # Check quorum if required
        if self.quorum and votes_cast  total_agents / 2
        
        elif self.consensus_type == ConsensusType.SUPERMAJORITY:
            return votes_for >= (total_agents * 2 / 3)
        
        elif self.consensus_type == ConsensusType.QUORUM:
            return votes_cast >= (self.quorum or total_agents)
        
        return False
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute consensus pattern."""
        total_proposals = len(self.proposals)
        accepted = sum(1 for p in self.proposals.values() 
                      if p.status == ProposalStatus.ACCEPTED)
        
        return {
            "pattern": "Consensus",
            "scenario": scenario,
            "consensus_type": self.consensus_type.value,
            "total_agents": len(self.agents),
            "total_proposals": total_proposals,
            "accepted_proposals": accepted,
            "consensus_rounds": len(self.consensus_rounds)
        }


# Example usage
if __name__ == "__main__":
    print("=== Consensus Pattern Demo ===\n")
    
    # Create consensus group (majority rule)
    print("1. Creating consensus group (majority rule)...")
    pattern = ConsensusPattern(consensus_type=ConsensusType.MAJORITY)
    
    # Register agents
    agents = [
        ConsensusAgent("Alice"),
        ConsensusAgent("Bob"),
        ConsensusAgent("Carol"),
        ConsensusAgent("Dave"),
        ConsensusAgent("Eve")
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
    
    print(f"   Registered {len(agents)} agents")
    
    # Proposal 1: Should be accepted
    print("\n2. Proposal 1: Adopt new AI safety guidelines")
    proposal1 = agents[0].propose("prop1", "Adopt comprehensive AI safety guidelines for all projects")
    pattern.submit_proposal(proposal1)
    result1 = pattern.conduct_consensus("prop1", max_rounds=1)
    
    # Proposal 2: Different proposal
    print("\n3. Proposal 2: Increase project budget by 50%")
    proposal2 = agents[1].propose("prop2", "Increase project budget by 50% to accelerate development")
    pattern.submit_proposal(proposal2)
    result2 = pattern.conduct_consensus("prop2", max_rounds=1)
    
    # Proposal 3: Another proposal
    print("\n4. Proposal 3: Switch to open-source model")
    proposal3 = agents[2].propose("prop3", "Switch primary AI model from proprietary to open-source")
    pattern.submit_proposal(proposal3)
    result3 = pattern.conduct_consensus("prop3", max_rounds=1)
    
    # Try unanimous consensus
    print("\n\n5. Testing UNANIMOUS consensus (harder to achieve)...")
    unanimous_pattern = ConsensusPattern(consensus_type=ConsensusType.UNANIMOUS)
    
    for agent in agents[:3]:  # Smaller group for unanimous
        unanimous_pattern.register_agent(agent)
    
    proposal4 = agents[0].propose("prop4", "Schedule team meeting for next Monday at 2 PM")
    unanimous_pattern.submit_proposal(proposal4)
    result4 = unanimous_pattern.conduct_consensus("prop4", max_rounds=1)
    
    # Statistics
    print("\n\n6. Consensus Statistics:")
    summary = pattern.run("Team decision making")
    print(f"   Total Proposals: {summary['total_proposals']}")
    print(f"   Accepted: {summary['accepted_proposals']}")
    print(f"   Rejection rate: {((summary['total_proposals'] - summary['accepted_proposals']) / max(1, summary['total_proposals']) * 100):.1f}%")
    print(f"   Consensus rounds: {summary['consensus_rounds']}")
    
    # Vote distribution
    print("\n7. Vote Distribution:")
    for prop_id, proposal in pattern.proposals.items():
        print(f"   {prop_id}: {len(proposal.votes_for)} for, "
              f"{len(proposal.votes_against)} against, "
              f"{len(proposal.abstentions)} abstain -> {proposal.status.value}")
    
    print("\nāœ“ Consensus pattern enabled collaborative decision making!")

Agentic Architecture Pattern: Consensus

Pattern 29: 29 Consensus – Line-by-Line Explanation

Overview

Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms. Supports multiple consensus algorithms: unanimous, majority, quorum-based.

Code Structure Analysis

File Information

  • Pattern Number: 29
  • Pattern Name: 29 Consensus
  • File Path: patterns\collaboration\29_consensus.py
  • Total Lines: 289

Detailed Code Explanation

Module Docstring

"""
Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms.
Supports multiple consensus algorithms: unanimous, majority, quorum-based.
"""

Explanation: This module docstring describes the purpose and key features of the 29 Consensus pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ConsensusType (Line 16)

class ConsensusType(Enum):
    UNANIMOUS
    MAJORITY
    SUPERMAJORITY
    QUORUM

Purpose: Type of consensus mechanism.

Values: – UNANIMOUS: Represents unanimous – MAJORITY: Represents majority – SUPERMAJORITY: Represents supermajority – QUORUM: Represents quorum

ProposalStatus (Line 24)

class ProposalStatus(Enum):
    PROPOSED
    DISCUSSING
    VOTING
    ACCEPTED
    REJECTED
    NO_CONSENSUS

Purpose: Status of consensus proposal.

Values: – PROPOSED: Represents proposed – DISCUSSING: Represents discussing – VOTING: Represents voting – ACCEPTED: Represents accepted – REJECTED: Represents rejected – NO_CONSENSUS: Represents no consensus

Classes

ConsensusType (Line 16)

Description: Type of consensus mechanism.

ProposalStatus (Line 24)

Description: Status of consensus proposal.

Proposal (Line 35)

Description: Proposal for consensus.

ConsensusRound (Line 49)

Description: Round of consensus building.

ConsensusAgent (Line 59)

Description: Agent participating in consensus.

Methods:

__init__ (Line 62)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

propose (Line 67)
def propose(self, proposal_id, content):
    """Make a proposal."""

Purpose: Make a proposal.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – content: Parameter for content

evaluate_proposal (Line 77)
def evaluate_proposal(self, proposal, context):
    """Evaluate a proposal and provide stance."""

Purpose: Evaluate a proposal and provide stance.

Parameters: – self: Instance reference – proposal: Parameter for proposal – context: Parameter for context

vote (Line 94)
def vote(self, proposal, decision):
    """Cast vote on proposal."""

Purpose: Cast vote on proposal.

Parameters: – self: Instance reference – proposal: Parameter for proposal – decision: Parameter for decision

ConsensusPattern (Line 109)

Description: Consensus Pattern – Agreement through iterative consensus.

Agents work together to reach consensus on proposals through discussion, voting, and iterative refinement.

Methods:

__init__ (Line 117)
def __init__(self, consensus_type, quorum):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – consensus_type: Parameter for consensus_type – quorum: Parameter for quorum

register_agent (Line 125)
def register_agent(self, agent):
    """Register agent in consensus group."""

Purpose: Register agent in consensus group.

Parameters: – self: Instance reference – agent: Parameter for agent

submit_proposal (Line 129)
def submit_proposal(self, proposal):
    """Submit proposal for consensus."""

Purpose: Submit proposal for consensus.

Parameters: – self: Instance reference – proposal: Parameter for proposal

conduct_consensus (Line 134)
def conduct_consensus(self, proposal_id, max_rounds):
    """Conduct consensus building rounds."""

Purpose: Conduct consensus building rounds.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – max_rounds: Parameter for max_rounds

_check_consensus (Line 176)
def _check_consensus(self, proposal):
    """Check if consensus is reached."""

Purpose: Check if consensus is reached.

Parameters: – self: Instance reference – proposal: Parameter for proposal

run (Line 205)
def run(self, scenario):
    """Execute consensus pattern."""

Purpose: Execute consensus pattern.

Parameters: – self: Instance reference – scenario: Parameter for scenario

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = ConsensusType(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts