"""
Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions.
Supports multiple voting methods and weighted votes.
"""

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


class VotingMethod(Enum):
    """Type of voting method."""
    MAJORITY = "majority"  # Simple majority wins
    PLURALITY = "plurality"  # Most votes wins
    UNANIMOUS = "unanimous"  # All must agree
    WEIGHTED = "weighted"  # Votes have different weights
    RANKED_CHOICE = "ranked_choice"  # Rank preferences


@dataclass
class VoteOption:
    """Option that can be voted on."""
    option_id: str
    description: str
    proposed_by: str


@dataclass
class Vote:
    """Individual vote cast by an agent."""
    voter_name: str
    proposal_id: str
    choice: str  # option_id
    weight: float = 1.0
    ranking: Optional[List[str]] = None  # For ranked choice
    reasoning: str = ""
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class VotingResult:
    """Result of a vote."""
    proposal_id: str
    method: VotingMethod
    winner: Optional[str]
    all_votes: List[Vote]
    vote_counts: Dict[str, float]
    total_voters: int
    passed: bool


class VotingAgent:
    """Agent that participates in voting."""
    
    def __init__(self, name: str, role: str, expertise: List[str],
                 vote_weight: float = 1.0, model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.expertise = expertise
        self.vote_weight = vote_weight
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def cast_vote(self, proposal_id: str, options: List[VoteOption],
                  method: VotingMethod = VotingMethod.MAJORITY) -> Vote:
        """Cast vote on proposal."""
        
        options_text = "\n".join([f"- {opt.option_id}: {opt.description}" 
                                  for opt in options])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role} with expertise in {', '.join(self.expertise)}. "
                      f"Analyze the options and vote based on your expertise."),
            ("human", f"Proposal: {proposal_id}\n\n"
                     f"Options:\n{options_text}\n\n"
                     f"Choose the best option and explain why briefly.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        content = str(response.content)
        
        # Parse choice from response
        choice = options[0].option_id  # Default
        for opt in options:
            if opt.option_id.lower() in content.lower():
                choice = opt.option_id
                break
        
        # For ranked choice, create ranking
        ranking = None
        if method == VotingMethod.RANKED_CHOICE:
            ranking = [opt.option_id for opt in options]
        
        vote = Vote(
            voter_name=self.name,
            proposal_id=proposal_id,
            choice=choice,
            weight=self.vote_weight,
            ranking=ranking,
            reasoning=content[:150]
        )
        
        return vote


class VotingPattern:
    """
    Voting Pattern - Democratic decision-making through voting.
    
    Agents cast votes on proposals using various voting methods.
    Supports weighted votes, ranked choice, and different consensus rules.
    """
    
    def __init__(self, voting_method: VotingMethod = VotingMethod.MAJORITY):
        self.voting_method = voting_method
        self.agents: Dict[str, VotingAgent] = {}
        self.voting_history: List[VotingResult] = []
    
    def register_agent(self, agent: VotingAgent):
        """Register agent as voter."""
        self.agents[agent.name] = agent
    
    def conduct_vote(self, proposal_id: str, options: List[VoteOption]) -> VotingResult:
        """Conduct vote among registered agents."""
        print(f"\n   Voting on: {proposal_id}")
        print(f"   Options: {', '.join([opt.option_id for opt in options])}")
        print(f"   Method: {self.voting_method.value}")
        
        # Collect votes
        votes: List[Vote] = []
        for agent in self.agents.values():
            vote = agent.cast_vote(proposal_id, options, self.voting_method)
            votes.append(vote)
            print(f"   - {agent.name} voted for: {vote.choice}")
        
        # Count votes based on method
        vote_counts: Dict[str, float] = {}
        
        if self.voting_method == VotingMethod.WEIGHTED:
            for vote in votes:
                vote_counts[vote.choice] = vote_counts.get(vote.choice, 0) + vote.weight
        else:
            for vote in votes:
                vote_counts[vote.choice] = vote_counts.get(vote.choice, 0) + 1
        
        # Determine winner
        winner = None
        passed = False
        
        if self.voting_method == VotingMethod.MAJORITY:
            total = sum(vote_counts.values())
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                passed = vote_counts[winner] > total / 2
        
        elif self.voting_method == VotingMethod.PLURALITY:
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                passed = True
        
        elif self.voting_method == VotingMethod.UNANIMOUS:
            if len(set(v.choice for v in votes)) == 1:
                winner = votes[0].choice
                passed = True
        
        elif self.voting_method == VotingMethod.WEIGHTED:
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                total_weight = sum(vote_counts.values())
                passed = vote_counts[winner] > total_weight / 2
        
        if passed:
            print(f"   ✓ Winner: {winner} ({vote_counts.get(winner, 0)} votes)")
        else:
            print(f"   ✗ No winner (consensus not reached)")
        
        result = VotingResult(
            proposal_id=proposal_id,
            method=self.voting_method,
            winner=winner,
            all_votes=votes,
            vote_counts=vote_counts,
            total_voters=len(votes),
            passed=passed
        )
        
        self.voting_history.append(result)
        return result
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute voting pattern."""
        passed_votes = sum(1 for v in self.voting_history if v.passed)
        
        return {
            "pattern": "Voting",
            "voting_method": self.voting_method.value,
            "total_votes": len(self.voting_history),
            "passed": passed_votes,
            "failed": len(self.voting_history) - passed_votes,
            "registered_voters": len(self.agents),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Voting Pattern Demo ===\n")
    
    # Create pattern with weighted voting
    pattern = VotingPattern(voting_method=VotingMethod.WEIGHTED)
    
    # Register voters with different weights
    print("1. Registering voters...")
    
    tech_lead = VotingAgent("Alice", "tech lead", ["architecture", "security"], vote_weight=2.0)
    senior_dev = VotingAgent("Bob", "senior developer", ["backend", "databases"], vote_weight=1.5)
    designer = VotingAgent("Carol", "designer", ["UI", "UX"], vote_weight=1.0)
    pm = VotingAgent("Dave", "product manager", ["business", "strategy"], vote_weight=1.5)
    
    for agent in [tech_lead, senior_dev, designer, pm]:
        pattern.register_agent(agent)
    
    # Vote on proposals
    print("\n2. Conducting votes...")
    
    proposal1_options = [
        VoteOption("opt1", "Use microservices architecture", "Alice"),
        VoteOption("opt2", "Use monolithic architecture", "Bob"),
        VoteOption("opt3", "Use serverless architecture", "Dave")
    ]
    
    proposal2_options = [
        VoteOption("opt1", "React for frontend", "Carol"),
        VoteOption("opt2", "Vue for frontend", "Bob"),
        VoteOption("opt3", "Angular for frontend", "Alice")
    ]
    
    result1 = pattern.conduct_vote("architecture_decision", proposal1_options)
    result2 = pattern.conduct_vote("frontend_framework", proposal2_options)
    
    # Summary
    print("\n3. Voting Summary:")
    summary = pattern.run("Team technical decisions")
    print(f"   Method: {summary['voting_method']}")
    print(f"   Total Votes: {summary['total_votes']}")
    print(f"   Passed: {summary['passed']}")
    print(f"   Failed: {summary['failed']}")
    print(f"   Registered Voters: {summary['registered_voters']}")
    
    print("\n✓ Voting pattern enabled democratic decision-making!")

    

Agentic Architecture Pattern: Voting

Pattern 25: 25 Voting – Line-by-Line Explanation

Overview

Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions. Supports multiple voting methods and weighted votes.

Code Structure Analysis

File Information

• Pattern Number: 25
• Pattern Name: 25 Voting
• File Path: patterns\collaboration\25_voting.py
• Total Lines: 257

Detailed Code Explanation

Module Docstring

"""
Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions.
Supports multiple voting methods and weighted votes.
"""

Explanation: This module docstring describes the purpose and key features of the 25 Voting pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import Counter
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: 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

VotingMethod (Line 17)

class VotingMethod(Enum):
    MAJORITY
    PLURALITY
    UNANIMOUS
    WEIGHTED
    RANKED_CHOICE

Purpose: Type of voting method.

Values: – MAJORITY: Represents majority – PLURALITY: Represents plurality – UNANIMOUS: Represents unanimous – WEIGHTED: Represents weighted – RANKED_CHOICE: Represents ranked choice

Classes

VotingMethod (Line 17)

Description: Type of voting method.

VoteOption (Line 27)

Description: Option that can be voted on.

Vote (Line 35)

Description: Individual vote cast by an agent.

VotingResult (Line 47)

Description: Result of a vote.

VotingAgent (Line 58)

Description: Agent that participates in voting.

Methods:

__init__ (Line 61)

def __init__(self, name, role, expertise, vote_weight, model_name):
    """No method docstring."""

Purpose: No method docstring.

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

cast_vote (Line 69)

def cast_vote(self, proposal_id, options, method):
    """Cast vote on proposal."""

Purpose: Cast vote on proposal.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – options: Parameter for options – method: Parameter for method

VotingPattern (Line 112)

Description: Voting Pattern – Democratic decision-making through voting.

Agents cast votes on proposals using various voting methods. Supports weighted votes, ranked choice, and different consensus rules.

Methods:

__init__ (Line 120)

def __init__(self, voting_method):
    """No method docstring."""

Purpose: No method docstring.

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

register_agent (Line 125)

def register_agent(self, agent):
    """Register agent as voter."""

Purpose: Register agent as voter.

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

conduct_vote (Line 129)

def conduct_vote(self, proposal_id, options):
    """Conduct vote among registered agents."""

Purpose: Conduct vote among registered agents.

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

run (Line 196)

def run(self, scenario):
    """Execute voting pattern."""

Purpose: Execute voting 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 = VotingMethod(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