aishastr.live

Agentic Architecture Pattern: Coalition Formation

Written by

admin

in


    """
Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals.
Features coalition evaluation, membership negotiation, and benefit distribution.
"""

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 CoalitionStatus(Enum):
    """Status of coalition."""
    FORMING = "forming"
    ACTIVE = "active"
    DISSOLVED = "dissolved"
    COMPLETED = "completed"


@dataclass
class CoalitionGoal:
    """Goal that coalition works toward."""
    goal_id: str
    description: str
    required_capabilities: List[str]
    reward: float
    deadline: Optional[datetime] = None


@dataclass
class Coalition:
    """Group of agents working together."""
    coalition_id: str
    name: str
    goal: CoalitionGoal
    members: Set[str] = field(default_factory=set)
    leader: Optional[str] = None
    status: CoalitionStatus = CoalitionStatus.FORMING
    total_reward: float = 0
    formed_at: datetime = field(default_factory=datetime.now)


@dataclass
class CoalitionAgent:
    """Agent that can form coalitions."""
    name: str
    capabilities: List[str]
    reputation: float = 5.0  # 0-10
    current_coalitions: Set[str] = field(default_factory=set)
    rewards_earned: float = 0


class CoalitionFormationPattern:
    """
    Coalition Formation Pattern - Dynamic alliance formation.
    
    Agents evaluate potential coalitions, negotiate membership,
    and work together to achieve shared goals with reward distribution.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.agents: Dict[str, CoalitionAgent] = {}
        self.coalitions: Dict[str, Coalition] = {}
        self.goals: List[CoalitionGoal] = []
    
    def register_agent(self, agent: CoalitionAgent):
        """Register agent for coalition formation."""
        self.agents[agent.name] = agent
    
    def add_goal(self, goal: CoalitionGoal):
        """Add goal that requires coalition."""
        self.goals.append(goal)
    
    def evaluate_coalition_value(self, agent: CoalitionAgent, 
                                 coalition_goal: CoalitionGoal,
                                 potential_members: List[str]) -> float:
        """Evaluate value of joining coalition for agent."""
        # Check capability match
        agent_capabilities_match = sum(
            1 for cap in coalition_goal.required_capabilities 
            if cap in agent.capabilities
        )
        
        # Check if coalition can meet requirements
        all_capabilities = set(agent.capabilities)
        for member_name in potential_members:
            if member_name in self.agents:
                all_capabilities.update(self.agents[member_name].capabilities)
        
        can_achieve = all(
            cap in all_capabilities 
            for cap in coalition_goal.required_capabilities
        )
        
        if not can_achieve:
            return 0.0
        
        # Estimate reward share (simple: divide by members)
        estimated_share = coalition_goal.reward / (len(potential_members) + 1)
        
        # Adjust by reputation
        value = estimated_share * (agent.reputation / 10.0)
        
        return value
    
    def form_coalition(self, goal: CoalitionGoal) -> Optional[Coalition]:
        """Attempt to form coalition for goal."""
        print(f"\n   Forming coalition for: {goal.description}")
        print(f"   Required: {', '.join(goal.required_capabilities)}")
        
        # Find capable agents
        interested_agents = []
        for agent in self.agents.values():
            # Check if agent has any required capability
            has_relevant_skill = any(
                cap in agent.capabilities 
                for cap in goal.required_capabilities
            )
            if has_relevant_skill:
                interested_agents.append(agent.name)
        
        if not interested_agents:
            print(f"   ✗ No agents with required capabilities")
            return None
        
        # Evaluate coalition value for each agent
        coalition_members = set()
        for agent_name in interested_agents:
            agent = self.agents[agent_name]
            value = self.evaluate_coalition_value(
                agent, goal, list(coalition_members)
            )
            
            if value > 0:
                coalition_members.add(agent_name)
                print(f"   + {agent_name} joins (value: {value:.1f})")
        
        # Check if coalition can achieve goal
        all_capabilities = set()
        for member in coalition_members:
            all_capabilities.update(self.agents[member].capabilities)
        
        can_achieve = all(
            cap in all_capabilities 
            for cap in goal.required_capabilities
        )
        
        if not can_achieve:
            print(f"   ✗ Coalition cannot meet all requirements")
            return None
        
        # Create coalition
        leader = max(coalition_members, 
                    key=lambda m: self.agents[m].reputation)
        
        coalition = Coalition(
            coalition_id=f"coal_{len(self.coalitions)}",
            name=f"Coalition for {goal.goal_id}",
            goal=goal,
            members=coalition_members,
            leader=leader,
            status=CoalitionStatus.ACTIVE
        )
        
        # Update agent memberships
        for member in coalition_members:
            self.agents[member].current_coalitions.add(coalition.coalition_id)
        
        self.coalitions[coalition.coalition_id] = coalition
        
        print(f"   ✓ Coalition formed: {len(coalition_members)} members, leader: {leader}")
        return coalition
    
    def execute_coalition_goal(self, coalition: Coalition) -> bool:
        """Execute coalition's goal."""
        print(f"\n   Executing: {coalition.name}")
        
        # Simulate work using LLM
        members_text = ", ".join(coalition.members)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are describing a coalition executing a goal."),
            ("human", f"Coalition members: {members_text}\n"
                     f"Leader: {coalition.leader}\n"
                     f"Goal: {coalition.goal.description}\n"
                     f"Required capabilities: {', '.join(coalition.goal.required_capabilities)}\n\n"
                     f"Describe how they work together (2 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        print(f"   Work: {str(response.content)[:150]}...")
        
        # Distribute rewards
        share_per_member = coalition.goal.reward / len(coalition.members)
        
        print(f"   Distributing {coalition.goal.reward} reward:")
        for member in coalition.members:
            self.agents[member].rewards_earned += share_per_member
            print(f"   - {member}: +{share_per_member:.1f}")
        
        coalition.total_reward = coalition.goal.reward
        coalition.status = CoalitionStatus.COMPLETED
        
        return True
    
    def dissolve_coalition(self, coalition_id: str):
        """Dissolve coalition."""
        if coalition_id in self.coalitions:
            coalition = self.coalitions[coalition_id]
            
            # Remove from agent memberships
            for member in coalition.members:
                if member in self.agents:
                    self.agents[member].current_coalitions.discard(coalition_id)
            
            coalition.status = CoalitionStatus.DISSOLVED
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute coalition formation pattern."""
        # Form coalitions for all goals
        formed_coalitions = []
        for goal in self.goals:
            coalition = self.form_coalition(goal)
            if coalition:
                formed_coalitions.append(coalition)
        
        # Execute coalition goals
        for coalition in formed_coalitions:
            self.execute_coalition_goal(coalition)
        
        # Statistics
        total_rewards_distributed = sum(
            agent.rewards_earned for agent in self.agents.values()
        )
        
        return {
            "pattern": "CoalitionFormation",
            "total_agents": len(self.agents),
            "goals_available": len(self.goals),
            "coalitions_formed": len(formed_coalitions),
            "coalitions_completed": sum(
                1 for c in self.coalitions.values() 
                if c.status == CoalitionStatus.COMPLETED
            ),
            "total_rewards_distributed": total_rewards_distributed,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Coalition Formation Pattern Demo ===\n")
    
    # Create pattern
    pattern = CoalitionFormationPattern()
    
    # Register agents with different capabilities
    print("1. Registering agents...")
    
    agents = [
        CoalitionAgent("Alice", ["programming", "architecture"], reputation=8.0),
        CoalitionAgent("Bob", ["programming", "testing"], reputation=7.0),
        CoalitionAgent("Carol", ["design", "frontend"], reputation=7.5),
        CoalitionAgent("Dave", ["devops", "cloud"], reputation=6.5),
        CoalitionAgent("Eve", ["data_analysis", "ml"], reputation=8.5),
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
        print(f"   - {agent.name}: {', '.join(agent.capabilities)} (rep: {agent.reputation})")
    
    # Add goals requiring coalitions
    print("\n2. Adding coalition goals...")
    
    goals = [
        CoalitionGoal(
            "g1",
            "Build web application with ML features",
            required_capabilities=["programming", "frontend", "ml"],
            reward=1000
        ),
        CoalitionGoal(
            "g2",
            "Deploy cloud infrastructure",
            required_capabilities=["devops", "cloud", "architecture"],
            reward=750
        ),
        CoalitionGoal(
            "g3",
            "Create data dashboard",
            required_capabilities=["data_analysis", "design", "programming"],
            reward=800
        ),
    ]
    
    for goal in goals:
        pattern.add_goal(goal)
    
    # Run coalition formation
    print("\n3. Forming and executing coalitions...")
    summary = pattern.run("Project team coalition formation")
    
    # Final results
    print("\n4. Coalition Results:")
    print(f"   Goals: {summary['goals_available']}")
    print(f"   Coalitions Formed: {summary['coalitions_formed']}")
    print(f"   Completed: {summary['coalitions_completed']}")
    print(f"   Total Rewards: ${summary['total_rewards_distributed']:.2f}")
    
    print("\n5. Agent Earnings:")
    for agent_name, agent in pattern.agents.items():
        coalitions = len(agent.current_coalitions)
        print(f"   {agent_name}: ${agent.rewards_earned:.2f} ({coalitions} coalitions)")
    
    print("\n✓ Agents formed coalitions and achieved shared goals!")

Agentic Architecture Pattern: Coalition Formation

Pattern 26: 26 Coalition Formation – Line-by-Line Explanation

Overview

Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals. Features coalition evaluation, membership negotiation, and benefit distribution.

Code Structure Analysis

File Information

  • Pattern Number: 26
  • Pattern Name: 26 Coalition Formation
  • File Path: patterns\collaboration\26_coalition_formation.py
  • Total Lines: 323

Detailed Code Explanation

Module Docstring

"""
Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals.
Features coalition evaluation, membership negotiation, and benefit distribution.
"""

Explanation: This module docstring describes the purpose and key features of the 26 Coalition Formation 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

CoalitionStatus (Line 16)

class CoalitionStatus(Enum):
    FORMING
    ACTIVE
    DISSOLVED
    COMPLETED

Purpose: Status of coalition.

Values: – FORMING: Represents forming – ACTIVE: Currently operational – DISSOLVED: Represents dissolved – COMPLETED: Successfully finished

Classes

CoalitionStatus (Line 16)

Description: Status of coalition.

CoalitionGoal (Line 25)

Description: Goal that coalition works toward.

Coalition (Line 35)

Description: Group of agents working together.

CoalitionAgent (Line 48)

Description: Agent that can form coalitions.

CoalitionFormationPattern (Line 57)

Description: Coalition Formation Pattern – Dynamic alliance formation.

Agents evaluate potential coalitions, negotiate membership, and work together to achieve shared goals with reward distribution.

Methods:

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

Purpose: No method docstring.

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

register_agent (Line 71)
def register_agent(self, agent):
    """Register agent for coalition formation."""

Purpose: Register agent for coalition formation.

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

add_goal (Line 75)
def add_goal(self, goal):
    """Add goal that requires coalition."""

Purpose: Add goal that requires coalition.

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

evaluate_coalition_value (Line 79)
def evaluate_coalition_value(self, agent, coalition_goal, potential_members):
    """Evaluate value of joining coalition for agent."""

Purpose: Evaluate value of joining coalition for agent.

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

form_coalition (Line 111)
def form_coalition(self, goal):
    """Attempt to form coalition for goal."""

Purpose: Attempt to form coalition for goal.

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

execute_coalition_goal (Line 179)
def execute_coalition_goal(self, coalition):
    """Execute coalition's goal."""

Purpose: Execute coalition’s goal.

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

dissolve_coalition (Line 213)
def dissolve_coalition(self, coalition_id):
    """Dissolve coalition."""

Purpose: Dissolve coalition.

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

run (Line 225)
def run(self, scenario):
    """Execute coalition formation pattern."""

Purpose: Execute coalition formation 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 = CoalitionStatus(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