Tag: llm

  • Agentic Architecture Pattern: Team Based

    
        """
    Pattern 27: Team-Based Pattern
    
    Agents organized into formal teams with defined roles and responsibilities.
    Teams have leaders, members, and coordinate to achieve team goals.
    """
    
    from typing import Dict, Any, List, Optional
    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 TeamRole(Enum):
        """Role within a team."""
        LEADER = "leader"
        MEMBER = "member"
        SPECIALIST = "specialist"
        COORDINATOR = "coordinator"
    
    
    @dataclass
    class TeamMember:
        """Member of a team."""
        name: str
        role: TeamRole
        skills: List[str]
        availability: float = 1.0  # 0.0 to 1.0
        tasks_assigned: List[str] = field(default_factory=list)
    
    
    @dataclass
    class TeamTask:
        """Task assigned to team."""
        task_id: str
        description: str
        required_skills: List[str]
        assigned_to: Optional[str] = None
        status: str = "pending"  # pending, in_progress, completed
        priority: int = 5  # 1-10
    
    
    @dataclass
    class Team:
        """Team of agents."""
        team_id: str
        name: str
        purpose: str
        members: List[TeamMember] = field(default_factory=list)
        tasks: List[TeamTask] = field(default_factory=list)
        achievements: List[str] = field(default_factory=list)
    
    
    class TeamAgent:
        """Agent that works as part of a team."""
        
        def __init__(self, name: str, role: TeamRole, skills: List[str], model_name: str = "gpt-4"):
            self.member = TeamMember(name=name, role=role, skills=skills)
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        def work_on_task(self, task: TeamTask, team: Team) -> str:
            """Work on assigned task."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.member.name}, a {self.member.role.value} with skills: {', '.join(self.member.skills)}. "
                          f"You're part of team '{team.name}' working on: {team.purpose}"),
                ("human", f"Task: {task.description}\n"
                         f"Required skills: {', '.join(task.required_skills)}\n"
                         f"Priority: {task.priority}/10\n\n"
                         f"Describe your approach to completing this task (2-3 sentences).")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            return str(response.content)[:200]
        
        def collaborate_with(self, other_member: TeamMember, context: str) -> str:
            """Collaborate with another team member."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.member.name} collaborating with {other_member.name}."),
                ("human", f"Context: {context}\n\nHow will you work together? (brief response)")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            return str(response.content)[:150]
    
    
    class TeamBasedPattern:
        """
        Team-Based Pattern - Structured team organization and coordination.
        
        Agents are organized into teams with formal roles and responsibilities.
        Teams have leaders, coordinate work, and achieve collective goals.
        """
        
        def __init__(self):
            self.teams: Dict[str, Team] = {}
            self.agents: Dict[str, TeamAgent] = {}
        
        def create_team(self, team_id: str, name: str, purpose: str) -> Team:
            """Create a new team."""
            team = Team(team_id=team_id, name=name, purpose=purpose)
            self.teams[team_id] = team
            return team
        
        def add_member_to_team(self, team_id: str, agent: TeamAgent):
            """Add agent to team."""
            if team_id in self.teams:
                self.teams[team_id].members.append(agent.member)
                self.agents[agent.member.name] = agent
        
        def assign_task(self, team_id: str, task: TeamTask) -> bool:
            """Assign task to appropriate team member."""
            if team_id not in self.teams:
                return False
            
            team = self.teams[team_id]
            team.tasks.append(task)
            
            # Find best member for task based on skills
            best_member = None
            best_match_score = 0
            
            for member in team.members:
                # Count matching skills
                matches = sum(1 for skill in task.required_skills if skill in member.skills)
                if matches > best_match_score and member.availability > 0.3:
                    best_match_score = matches
                    best_member = member
            
            if best_member:
                task.assigned_to = best_member.name
                task.status = "in_progress"
                best_member.tasks_assigned.append(task.task_id)
                return True
            
            return False
        
        def execute_team_work(self, team_id: str) -> Dict[str, Any]:
            """Execute work for team."""
            if team_id not in self.teams:
                return {"success": False, "error": "Team not found"}
            
            team = self.teams[team_id]
            results = []
            
            print(f"\n   Team '{team.name}' working on: {team.purpose}")
            
            # Execute tasks
            for task in team.tasks:
                if task.assigned_to and task.assigned_to in self.agents:
                    agent = self.agents[task.assigned_to]
                    print(f"   - {task.assigned_to} working on: {task.description}")
                    
                    work_result = agent.work_on_task(task, team)
                    results.append({
                        "task": task.task_id,
                        "member": task.assigned_to,
                        "result": work_result
                    })
                    
                    task.status = "completed"
                    team.achievements.append(f"Completed: {task.description}")
            
            # Team collaboration
            if len(team.members) >= 2:
                member1 = team.members[0]
                member2 = team.members[1]
                if member1.name in self.agents and member2.name in self.agents:
                    agent1 = self.agents[member1.name]
                    collab = agent1.collaborate_with(member2, f"Team goal: {team.purpose}")
                    print(f"   - Collaboration: {collab[:80]}...")
            
            return {
                "success": True,
                "team": team.name,
                "tasks_completed": sum(1 for t in team.tasks if t.status == "completed"),
                "achievements": len(team.achievements),
                "results": results
            }
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute team-based pattern."""
            total_tasks = sum(len(team.tasks) for team in self.teams.values())
            completed_tasks = sum(
                sum(1 for t in team.tasks if t.status == "completed")
                for team in self.teams.values()
            )
            
            return {
                "pattern": "TeamBased",
                "total_teams": len(self.teams),
                "total_agents": len(self.agents),
                "total_tasks": total_tasks,
                "completed_tasks": completed_tasks,
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Team-Based Pattern Demo ===\n")
        
        # Create pattern
        pattern = TeamBasedPattern()
        
        # Create teams
        print("1. Creating teams...")
        dev_team = pattern.create_team("dev1", "Development Team", "Build mobile app")
        design_team = pattern.create_team("design1", "Design Team", "Create UI/UX designs")
        
        # Add members to development team
        print("\n2. Adding team members...")
        dev_lead = TeamAgent("Alice", TeamRole.LEADER, ["python", "architecture", "testing"])
        backend_dev = TeamAgent("Bob", TeamRole.MEMBER, ["python", "databases", "APIs"])
        frontend_dev = TeamAgent("Carol", TeamRole.MEMBER, ["javascript", "react", "CSS"])
        
        pattern.add_member_to_team("dev1", dev_lead)
        pattern.add_member_to_team("dev1", backend_dev)
        pattern.add_member_to_team("dev1", frontend_dev)
        
        # Add members to design team
        design_lead = TeamAgent("Dave", TeamRole.LEADER, ["UI_design", "prototyping"])
        ux_specialist = TeamAgent("Eve", TeamRole.SPECIALIST, ["UX_research", "user_testing"])
        
        pattern.add_member_to_team("design1", design_lead)
        pattern.add_member_to_team("design1", ux_specialist)
        
        # Assign tasks
        print("\n3. Assigning tasks...")
        task1 = TeamTask("t1", "Implement authentication system", ["python", "security"], priority=9)
        task2 = TeamTask("t2", "Build REST API", ["python", "APIs"], priority=8)
        task3 = TeamTask("t3", "Create responsive UI", ["javascript", "CSS"], priority=7)
        task4 = TeamTask("t4", "Design app mockups", ["UI_design", "prototyping"], priority=9)
        task5 = TeamTask("t5", "Conduct user research", ["UX_research"], priority=6)
        
        pattern.assign_task("dev1", task1)
        pattern.assign_task("dev1", task2)
        pattern.assign_task("dev1", task3)
        pattern.assign_task("design1", task4)
        pattern.assign_task("design1", task5)
        
        # Execute team work
        print("\n4. Executing team work...")
        dev_result = pattern.execute_team_work("dev1")
        design_result = pattern.execute_team_work("design1")
        
        # Summary
        print("\n5. Team Performance Summary:")
        summary = pattern.run("Multi-team software development")
        print(f"   Total Teams: {summary['total_teams']}")
        print(f"   Total Agents: {summary['total_agents']}")
        print(f"   Tasks Completed: {summary['completed_tasks']}/{summary['total_tasks']}")
        
        print("\nāœ“ Team-based pattern organized agents into effective teams!")
    
        

    Agentic Architecture Pattern: Team Based

    Pattern 27: 27 Team Based – Line-by-Line Explanation

    Overview

    Pattern 27: Team-Based Pattern

    Agents organized into formal teams with defined roles and responsibilities. Teams have leaders, members, and coordinate to achieve team goals.


    Code Structure Analysis

    File Information

    • Pattern Number: 27
    • Pattern Name: 27 Team Based
    • File Path: patterns\collaboration\27_team_based.py
    • Total Lines: 258

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 27: Team-Based Pattern
    
    Agents organized into formal teams with defined roles and responsibilities.
    Teams have leaders, members, and coordinate to achieve team goals.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 27 Team Based pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    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: 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

    TeamRole (Line 16)

    class TeamRole(Enum):
        LEADER
        MEMBER
        SPECIALIST
        COORDINATOR
    

    Purpose: Role within a team.

    Values: – LEADER: Represents leader – MEMBER: Represents member – SPECIALIST: Represents specialist – COORDINATOR: Represents coordinator


    Classes

    TeamRole (Line 16)

    Description: Role within a team.

    TeamMember (Line 25)

    Description: Member of a team.

    TeamTask (Line 35)

    Description: Task assigned to team.

    Team (Line 46)

    Description: Team of agents.

    TeamAgent (Line 56)

    Description: Agent that works as part of a team.

    Methods:

    __init__ (Line 59)
    def __init__(self, name, role, skills, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    work_on_task (Line 63)
    def work_on_task(self, task, team):
        """Work on assigned task."""
    

    Purpose: Work on assigned task.

    Parameters: – self: Instance reference – task: Parameter for task – team: Parameter for team

    collaborate_with (Line 78)
    def collaborate_with(self, other_member, context):
        """Collaborate with another team member."""
    

    Purpose: Collaborate with another team member.

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

    TeamBasedPattern (Line 90)

    Description: Team-Based Pattern – Structured team organization and coordination.

    Agents are organized into teams with formal roles and responsibilities. Teams have leaders, coordinate work, and achieve collective goals.

    Methods:

    __init__ (Line 98)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    create_team (Line 102)
    def create_team(self, team_id, name, purpose):
        """Create a new team."""
    

    Purpose: Create a new team.

    Parameters: – self: Instance reference – team_id: Parameter for team_id – name: Identifier or label – purpose: Parameter for purpose

    add_member_to_team (Line 108)
    def add_member_to_team(self, team_id, agent):
        """Add agent to team."""
    

    Purpose: Add agent to team.

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

    assign_task (Line 114)
    def assign_task(self, team_id, task):
        """Assign task to appropriate team member."""
    

    Purpose: Assign task to appropriate team member.

    Parameters: – self: Instance reference – team_id: Parameter for team_id – task: Parameter for task

    execute_team_work (Line 141)
    def execute_team_work(self, team_id):
        """Execute work for team."""
    

    Purpose: Execute work for team.

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

    run (Line 184)
    def run(self, scenario):
        """Execute team-based pattern."""
    

    Purpose: Execute team-based 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 = TeamRole(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

  • Agentic Architecture Pattern: Coalition Formation

    
        """
    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

  • Agentic Architecture Pattern: Voting

    
        """
    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

  • Agentic Architecture Pattern: Auction Based

    
        """
    Pattern 24: Auction-Based Pattern
    
    Allocates tasks or resources through competitive bidding.
    Agents submit bids, and the best bid wins the right to perform the task.
    """
    
    from typing import Dict, Any, List, Optional
    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 AuctionType(Enum):
        """Type of auction."""
        FIRST_PRICE = "first_price"  # Highest bid wins, pays bid
        SECOND_PRICE = "second_price"  # Highest bid wins, pays second highest
        DUTCH = "dutch"  # Price decreases until someone accepts
        ENGLISH = "english"  # Price increases through competitive bidding
    
    
    @dataclass
    class AuctionItem:
        """Item being auctioned."""
        item_id: str
        task_description: str
        requirements: List[str]
        reserve_price: float = 0.0
        deadline: Optional[datetime] = None
    
    
    @dataclass
    class Bid:
        """Bid from an agent."""
        bidder_name: str
        item_id: str
        amount: float
        capabilities: List[str]
        proposed_timeline: str
        justification: str
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class AuctionResult:
        """Result of an auction."""
        item_id: str
        winner: Optional[str]
        winning_bid: Optional[float]
        all_bids: List[Bid]
        auction_type: AuctionType
        success: bool
    
    
    class AuctionAgent:
        """Agent that participates in auctions."""
        
        def __init__(self, name: str, role: str, capabilities: List[str], budget: float, model_name: str = "gpt-4"):
            self.name = name
            self.role = role
            self.capabilities = capabilities
            self.budget = budget
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        def evaluate_and_bid(self, item: AuctionItem) -> Optional[Bid]:
            """Evaluate item and submit bid if interested."""
            # Check if agent has required capabilities
            has_capabilities = all(req in self.capabilities for req in item.requirements)
            
            if not has_capabilities:
                return None
            
            # Use LLM to determine bid amount and strategy
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, a {self.role} with budget ${self.budget:.2f}. "
                          f"Your capabilities: {', '.join(self.capabilities)}. "
                          f"Decide how much to bid for this task."),
                ("human", f"Task: {item.task_description}\n"
                         f"Requirements: {', '.join(item.requirements)}\n"
                         f"Reserve Price: ${item.reserve_price:.2f}\n\n"
                         f"Provide:\n"
                         f"1. Bid amount (number only, less than your budget)\n"
                         f"2. Timeline estimate\n"
                         f"3. Brief justification")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            content = str(response.content)
            
            # Parse response for bid amount
            try:
                import re
                # Look for dollar amounts or numbers
                amounts = re.findall(r'\$?(\d+(?:\.\d{2})?)', content)
                if amounts:
                    bid_amount = float(amounts[0])
                    bid_amount = min(bid_amount, self.budget)
                    
                    if bid_amount >= item.reserve_price:
                        bid = Bid(
                            bidder_name=self.name,
                            item_id=item.item_id,
                            amount=bid_amount,
                            capabilities=self.capabilities,
                            proposed_timeline="2 weeks",  # Could parse from response
                            justification=content[:200]
                        )
                        return bid
            except:
                pass
            
            return None
    
    
    class AuctionBasedPattern:
        """
        Auction-Based Pattern - Task allocation through competitive bidding.
        
        Tasks are auctioned to agents who submit competitive bids.
        Various auction mechanisms determine winners and prices.
        """
        
        def __init__(self, auction_type: AuctionType = AuctionType.FIRST_PRICE):
            self.auction_type = auction_type
            self.agents: Dict[str, AuctionAgent] = {}
            self.auction_history: List[AuctionResult] = []
        
        def register_agent(self, agent: AuctionAgent):
            """Register agent to participate in auctions."""
            self.agents[agent.name] = agent
        
        def conduct_auction(self, item: AuctionItem) -> AuctionResult:
            """Conduct auction for item."""
            print(f"\n   Auctioning: {item.task_description}")
            print(f"   Reserve Price: ${item.reserve_price:.2f}")
            
            # Collect bids from agents
            bids: List[Bid] = []
            for agent in self.agents.values():
                bid = agent.evaluate_and_bid(item)
                if bid:
                    bids.append(bid)
                    print(f"   - {agent.name} bid: ${bid.amount:.2f}")
            
            # Determine winner based on auction type
            winner = None
            winning_bid_amount = None
            success = False
            
            if bids:
                # Sort bids by amount (descending)
                sorted_bids = sorted(bids, key=lambda b: b.amount, reverse=True)
                
                if self.auction_type == AuctionType.FIRST_PRICE:
                    # Highest bid wins and pays their bid
                    winner = sorted_bids[0].bidder_name
                    winning_bid_amount = sorted_bids[0].amount
                    success = True
                
                elif self.auction_type == AuctionType.SECOND_PRICE:
                    # Highest bid wins but pays second-highest bid
                    winner = sorted_bids[0].bidder_name
                    winning_bid_amount = sorted_bids[1].amount if len(sorted_bids) > 1 else sorted_bids[0].amount
                    success = True
                
                print(f"   āœ“ Winner: {winner} at ${winning_bid_amount:.2f}")
            else:
                print(f"   āœ— No valid bids received")
            
            result = AuctionResult(
                item_id=item.item_id,
                winner=winner,
                winning_bid=winning_bid_amount,
                all_bids=bids,
                auction_type=self.auction_type,
                success=success
            )
            
            self.auction_history.append(result)
            return result
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute auction-based pattern."""
            return {
                "pattern": "AuctionBased",
                "auction_type": self.auction_type.value,
                "registered_agents": len(self.agents),
                "total_auctions": len(self.auction_history),
                "successful_auctions": sum(1 for a in self.auction_history if a.success),
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Auction-Based Pattern Demo ===\n")
        
        # Create pattern
        pattern = AuctionBasedPattern(auction_type=AuctionType.FIRST_PRICE)
        
        # Register agents with different capabilities and budgets
        print("1. Registering agents...")
        agent1 = AuctionAgent("DevCorp", "development company", 
                             ["web_dev", "mobile_dev", "testing"], budget=5000)
        agent2 = AuctionAgent("DesignStudio", "design agency",
                             ["UI_design", "branding", "web_dev"], budget=3000)
        agent3 = AuctionAgent("TechSolutions", "tech consultancy",
                             ["web_dev", "cloud", "security"], budget=7000)
        
        for agent in [agent1, agent2, agent3]:
            pattern.register_agent(agent)
        
        # Auction tasks
        print("\n2. Conducting auctions...")
        
        task1 = AuctionItem(
            item_id="task1",
            task_description="Build e-commerce website",
            requirements=["web_dev"],
            reserve_price=2000.0
        )
        
        task2 = AuctionItem(
            item_id="task2",
            task_description="Design mobile app UI",
            requirements=["UI_design", "mobile_dev"],
            reserve_price=1500.0
        )
        
        # Run auctions
        result1 = pattern.conduct_auction(task1)
        result2 = pattern.conduct_auction(task2)
        
        # Summary
        print("\n3. Auction Summary:")
        summary = pattern.run("Task allocation via auction")
        print(f"   Total Agents: {summary['registered_agents']}")
        print(f"   Auctions Held: {summary['total_auctions']}")
        print(f"   Successful: {summary['successful_auctions']}")
        print(f"   Auction Type: {summary['auction_type']}")
        
        print("\nāœ“ Auction-based pattern allocated tasks through competitive bidding!")
    
        

    Agentic Architecture Pattern: Auction Based

    Pattern 24: 24 Auction Based – Line-by-Line Explanation

    Overview

    Pattern 24: Auction-Based Pattern

    Allocates tasks or resources through competitive bidding. Agents submit bids, and the best bid wins the right to perform the task.


    Code Structure Analysis

    File Information

    • Pattern Number: 24
    • Pattern Name: 24 Auction Based
    • File Path: patterns\collaboration\24_auction_based.py
    • Total Lines: 246

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 24: Auction-Based Pattern
    
    Allocates tasks or resources through competitive bidding.
    Agents submit bids, and the best bid wins the right to perform the task.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 24 Auction Based pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
                import re  # 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

    AuctionType (Line 16)

    class AuctionType(Enum):
        FIRST_PRICE
        SECOND_PRICE
        DUTCH
        ENGLISH
    

    Purpose: Type of auction.

    Values: – FIRST_PRICE: Represents first price – SECOND_PRICE: Represents second price – DUTCH: Represents dutch – ENGLISH: Represents english


    Classes

    AuctionType (Line 16)

    Description: Type of auction.

    AuctionItem (Line 25)

    Description: Item being auctioned.

    Bid (Line 35)

    Description: Bid from an agent.

    AuctionResult (Line 47)

    Description: Result of an auction.

    AuctionAgent (Line 57)

    Description: Agent that participates in auctions.

    Methods:

    __init__ (Line 60)
    def __init__(self, name, role, capabilities, budget, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    evaluate_and_bid (Line 67)
    def evaluate_and_bid(self, item):
        """Evaluate item and submit bid if interested."""
    

    Purpose: Evaluate item and submit bid if interested.

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

    AuctionBasedPattern (Line 118)

    Description: Auction-Based Pattern – Task allocation through competitive bidding.

    Tasks are auctioned to agents who submit competitive bids. Various auction mechanisms determine winners and prices.

    Methods:

    __init__ (Line 126)
    def __init__(self, auction_type):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    register_agent (Line 131)
    def register_agent(self, agent):
        """Register agent to participate in auctions."""
    

    Purpose: Register agent to participate in auctions.

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

    conduct_auction (Line 135)
    def conduct_auction(self, item):
        """Conduct auction for item."""
    

    Purpose: Conduct auction for item.

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

    run (Line 185)
    def run(self, scenario):
        """Execute auction-based pattern."""
    

    Purpose: Execute auction-based 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 = AuctionType(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

  • Agentic Architecture Pattern: Negotiation

    
        """
    Pattern 23: Negotiation Pattern
    
    Agents negotiate to reach mutually beneficial agreements.
    Supports offers, counteroffers, concessions, and conflict resolution.
    """
    
    from typing import Dict, Any, List, Optional
    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 NegotiationStatus(Enum):
        """Status of negotiation."""
        INITIATED = "initiated"
        IN_PROGRESS = "in_progress"
        AGREEMENT_REACHED = "agreement_reached"
        FAILED = "failed"
        DEADLOCK = "deadlock"
    
    
    @dataclass
    class NegotiationOffer:
        """Offer made during negotiation."""
        offer_id: str
        from_agent: str
        to_agent: str
        round_number: int
        terms: Dict[str, Any]
        reasoning: str
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    @dataclass
    class NegotiationIssue:
        """Issue being negotiated."""
        issue_id: str
        description: str
        proposer: str
        initial_terms: Dict[str, Any]
        priority: int = 5  # 1-10
    
    
    @dataclass
    class NegotiationResult:
        """Final result of negotiation."""
        issue_id: str
        status: NegotiationStatus
        participants: List[str]
        rounds: int
        final_terms: Optional[Dict[str, Any]]
        all_offers: List[NegotiationOffer]
        concessions_made: Dict[str, int]
    
    
    class NegotiationAgent:
        """Agent that negotiates with others."""
        
        def __init__(self, name: str, role: str, interests: Dict[str, int], 
                     flexibility: float = 0.5, model_name: str = "gpt-4"):
            self.name = name
            self.role = role
            self.interests = interests  # e.g., {"price": 8, "timeline": 6, "quality": 9}
            self.flexibility = flexibility  # 0.0-1.0, higher = more willing to compromise
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.concessions_made = 0
        
        def make_offer(self, issue: NegotiationIssue, round_number: int, 
                       previous_offer: Optional[NegotiationOffer] = None) -> NegotiationOffer:
            """Generate negotiation offer."""
            
            context = f"You are {self.name}, a {self.role}.\n"
            context += f"Your interests (1-10 priority): {self.interests}\n"
            context += f"Your flexibility level: {self.flexibility:.1f}\n"
            context += f"Negotiating: {issue.description}\n"
            
            if previous_offer:
                context += f"\nPrevious offer from {previous_offer.from_agent}:\n"
                context += f"Terms: {previous_offer.terms}\n"
                context += f"Their reasoning: {previous_offer.reasoning}\n"
            else:
                context += f"\nInitial terms proposed: {issue.initial_terms}\n"
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are a skilled negotiator. Make offers that balance your interests "
                          "with reaching agreement. Be specific about terms."),
                ("human", f"{context}\n\n"
                         f"Round {round_number} - Provide:\n"
                         f"1. Your proposed terms (be specific)\n"
                         f"2. Brief reasoning for your position\n"
                         f"Format: TERMS: [your terms] | REASONING: [your reasoning]")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            content = str(response.content)
            
            # Parse response
            terms_dict = issue.initial_terms.copy()
            reasoning = content[:150]
            
            if "TERMS:" in content and "REASONING:" in content:
                parts = content.split("REASONING:")
                reasoning = parts[1].strip()[:150] if len(parts) > 1 else reasoning
            
            # Adjust terms based on flexibility and concessions
            if previous_offer and self.flexibility > 0.3:
                self.concessions_made += 1
            
            offer = NegotiationOffer(
                offer_id=f"offer_{round_number}_{self.name}",
                from_agent=self.name,
                to_agent=previous_offer.from_agent if previous_offer else "all",
                round_number=round_number,
                terms=terms_dict,
                reasoning=reasoning
            )
            
            return offer
        
        def evaluate_offer(self, offer: NegotiationOffer) -> bool:
            """Evaluate if offer is acceptable."""
            # Simple evaluation: accept with probability based on flexibility and rounds
            import random
            acceptance_threshold = 0.5 + (self.flexibility * 0.3) + (self.concessions_made * 0.05)
            return random.random()  NegotiationResult:
            """Conduct negotiation between participants."""
            print(f"\n   Negotiating: {issue.description}")
            print(f"   Participants: {', '.join(participants)}")
            
            participant_agents = [self.agents[name] for name in participants if name in self.agents]
            if len(participant_agents) = 3:
                    recent_concessions = sum(a.concessions_made for a in participant_agents)
                    if recent_concessions == 0:
                        status = NegotiationStatus.DEADLOCK
                        print(f"   āœ— Deadlock detected - no progress")
                        break
            
            if status == NegotiationStatus.IN_PROGRESS:
                status = NegotiationStatus.FAILED
                print(f"   āœ— No agreement after {self.max_rounds} rounds")
            
            result = NegotiationResult(
                issue_id=issue.issue_id,
                status=status,
                participants=participants,
                rounds=round_num,
                final_terms=latest_offer.terms if status == NegotiationStatus.AGREEMENT_REACHED else None,
                all_offers=all_offers,
                concessions_made={a.name: a.concessions_made for a in participant_agents}
            )
            
            self.negotiations.append(result)
            return result
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute negotiation pattern."""
            successful = sum(1 for n in self.negotiations 
                            if n.status == NegotiationStatus.AGREEMENT_REACHED)
            
            return {
                "pattern": "Negotiation",
                "total_negotiations": len(self.negotiations),
                "successful": successful,
                "failed": len(self.negotiations) - successful,
                "average_rounds": sum(n.rounds for n in self.negotiations) / len(self.negotiations) 
                                if self.negotiations else 0,
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Negotiation Pattern Demo ===\n")
        
        # Create pattern
        pattern = NegotiationPattern(max_rounds=4)
        
        # Register agents with different interests
        print("1. Registering negotiating agents...")
        
        buyer = NegotiationAgent(
            "BuyerCorp", 
            "purchasing manager",
            interests={"price": 10, "quality": 7, "timeline": 5},
            flexibility=0.6
        )
        
        seller = NegotiationAgent(
            "SellerInc",
            "sales representative", 
            interests={"price": 9, "timeline": 8, "quality": 6},
            flexibility=0.5
        )
        
        consultant = NegotiationAgent(
            "ConsultantLLC",
            "business consultant",
            interests={"price": 5, "quality": 9, "timeline": 7},
            flexibility=0.7
        )
        
        for agent in [buyer, seller, consultant]:
            pattern.register_agent(agent)
        
        # Negotiate issues
        print("\n2. Conducting negotiations...")
        
        issue1 = NegotiationIssue(
            issue_id="contract1",
            description="Software development contract terms",
            proposer="BuyerCorp",
            initial_terms={
                "price": "$50,000",
                "timeline": "6 months", 
                "quality": "enterprise-grade"
            },
            priority=8
        )
        
        issue2 = NegotiationIssue(
            issue_id="partnership1",
            description="Strategic partnership agreement",
            proposer="ConsultantLLC",
            initial_terms={
                "revenue_split": "60-40",
                "duration": "2 years",
                "exclusivity": "non-exclusive"
            },
            priority=7
        )
        
        # Run negotiations
        result1 = pattern.negotiate(issue1, ["BuyerCorp", "SellerInc"])
        result2 = pattern.negotiate(issue2, ["SellerInc", "ConsultantLLC"])
        
        # Summary
        print("\n3. Negotiation Summary:")
        summary = pattern.run("Multi-party contract negotiations")
        print(f"   Total Negotiations: {summary['total_negotiations']}")
        print(f"   Successful: {summary['successful']}")
        print(f"   Failed: {summary['failed']}")
        print(f"   Average Rounds: {summary['average_rounds']:.1f}")
        
        print("\nāœ“ Negotiation pattern facilitated multi-agent conflict resolution!")
    
        

    Agentic Architecture Pattern: Negotiation

    Pattern 23: 23 Negotiation – Line-by-Line Explanation

    Overview

    Pattern 23: Negotiation Pattern

    Agents negotiate to reach mutually beneficial agreements. Supports offers, counteroffers, concessions, and conflict resolution.


    Code Structure Analysis

    File Information

    • Pattern Number: 23
    • Pattern Name: 23 Negotiation
    • File Path: patterns\collaboration\23_negotiation.py
    • Total Lines: 312

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 23: Negotiation Pattern
    
    Agents negotiate to reach mutually beneficial agreements.
    Supports offers, counteroffers, concessions, and conflict resolution.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 23 Negotiation pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
            import random  # 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

    NegotiationStatus (Line 16)

    class NegotiationStatus(Enum):
        INITIATED
        IN_PROGRESS
        AGREEMENT_REACHED
        FAILED
        DEADLOCK
    

    Purpose: Status of negotiation.

    Values: – INITIATED: Represents initiated – IN_PROGRESS: Represents in progress – AGREEMENT_REACHED: Represents agreement reached – FAILED: Execution failed – DEADLOCK: Represents deadlock


    Classes

    NegotiationStatus (Line 16)

    Description: Status of negotiation.

    NegotiationOffer (Line 26)

    Description: Offer made during negotiation.

    NegotiationIssue (Line 38)

    Description: Issue being negotiated.

    NegotiationResult (Line 48)

    Description: Final result of negotiation.

    NegotiationAgent (Line 59)

    Description: Agent that negotiates with others.

    Methods:

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

    Purpose: No method docstring.

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

    make_offer (Line 71)
    def make_offer(self, issue, round_number, previous_offer):
        """Generate negotiation offer."""
    

    Purpose: Generate negotiation offer.

    Parameters: – self: Instance reference – issue: Parameter for issue – round_number: Parameter for round_number – previous_offer: Parameter for previous_offer

    evaluate_offer (Line 124)
    def evaluate_offer(self, offer):
        """Evaluate if offer is acceptable."""
    

    Purpose: Evaluate if offer is acceptable.

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

    NegotiationPattern (Line 132)

    Description: Negotiation Pattern – Multi-agent negotiation and conflict resolution.

    Agents make offers and counteroffers to reach mutually acceptable agreements. Supports multiple negotiation rounds and concession tracking.

    Methods:

    __init__ (Line 140)
    def __init__(self, max_rounds):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    register_agent (Line 145)
    def register_agent(self, agent):
        """Register agent for negotiations."""
    

    Purpose: Register agent for negotiations.

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

    negotiate (Line 149)
    def negotiate(self, issue, participants):
        """Conduct negotiation between participants."""
    

    Purpose: Conduct negotiation between participants.

    Parameters: – self: Instance reference – issue: Parameter for issue – participants: Parameter for participants

    run (Line 222)
    def run(self, scenario):
        """Execute negotiation pattern."""
    

    Purpose: Execute negotiation 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 = NegotiationStatus(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

  • Agentic Architecture Pattern: Competitive Agent

    
        """
    Pattern 22: Competitive Agent Pattern
    
    Agents compete for resources, tasks, or market share.
    Features strategies, performance metrics, and competitive dynamics.
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    import random
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    class CompetitionType(Enum):
        """Type of competition."""
        ZERO_SUM = "zero_sum"  # One's gain is another's loss
        NON_ZERO_SUM = "non_zero_sum"  # Mutual benefits possible
        TOURNAMENT = "tournament"  # Elimination rounds
        CONTINUOUS = "continuous"  # Ongoing competition
    
    
    @dataclass
    class CompetitiveResource:
        """Resource being competed for."""
        resource_id: str
        name: str
        value: float
        quantity: int
        owner: Optional[str] = None
    
    
    @dataclass
    class CompetitiveStrategy:
        """Strategy used by competitive agent."""
        name: str
        aggression: float  # 0.0 to 1.0
        risk_tolerance: float  # 0.0 to 1.0
        cooperation_willingness: float  # 0.0 to 1.0
    
    
    @dataclass
    class CompetitionResult:
        """Result of competitive action."""
        round_number: int
        agent_name: str
        action: str
        success: bool
        score_change: float
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class CompetitiveAgent:
        """Agent that competes with others."""
        
        def __init__(self, name: str, strategy: CompetitiveStrategy, 
                     initial_score: float = 0, model_name: str = "gpt-4"):
            self.name = name
            self.strategy = strategy
            self.score = initial_score
            self.resources: List[CompetitiveResource] = []
            self.history: List[CompetitionResult] = []
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        def compete_for_resource(self, resource: CompetitiveResource, 
                                opponents: List[str]) -> float:
            """Compete for a resource, returns bid/effort."""
            # Use strategy to determine competitive effort
            base_effort = resource.value * self.strategy.aggression
            
            # Add randomness based on risk tolerance
            variation = random.uniform(-0.2, 0.2) * self.strategy.risk_tolerance
            effort = base_effort * (1 + variation)
            
            # Adjust based on current score (catch-up effect)
            if self.score  str:
            """Make strategic decision using LLM."""
            competitors_text = "\n".join([f"- {name}: {score}" 
                                         for name, score in competitor_scores.items()])
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, a competitive agent with this strategy:\n"
                          f"- Aggression: {self.strategy.aggression:.1f}\n"
                          f"- Risk tolerance: {self.strategy.risk_tolerance:.1f}\n"
                          f"- Cooperation willingness: {self.strategy.cooperation_willingness:.1f}\n"
                          f"Your current score: {self.score}"),
                ("human", f"Situation: {situation}\n\n"
                         f"Competitor scores:\n{competitors_text}\n\n"
                         f"What's your strategic move? (one sentence)")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            
            return str(response.content)[:150]
    
    
    class CompetitiveAgentPattern:
        """
        Competitive Agent Pattern - Agent competition and rivalry.
        
        Agents compete for resources using different strategies.
        Supports various competition types and dynamic interactions.
        """
        
        def __init__(self, competition_type: CompetitionType = CompetitionType.CONTINUOUS):
            self.competition_type = competition_type
            self.agents: Dict[str, CompetitiveAgent] = {}
            self.resources: List[CompetitiveResource] = []
            self.round_number = 0
        
        def register_agent(self, agent: CompetitiveAgent):
            """Register agent for competition."""
            self.agents[agent.name] = agent
        
        def add_resource(self, resource: CompetitiveResource):
            """Add resource to compete for."""
            self.resources.append(resource)
        
        def run_competition_round(self) -> List[CompetitionResult]:
            """Run one round of competition."""
            self.round_number += 1
            results = []
            
            print(f"\n   === Round {self.round_number} ===")
            
            # Compete for each available resource
            for resource in self.resources:
                if resource.owner:
                    continue  # Already claimed
                
                print(f"   Competing for: {resource.name} (value: {resource.value})")
                
                # Each agent submits their effort/bid
                efforts = {}
                for agent_name, agent in self.agents.items():
                    opponents = [n for n in self.agents.keys() if n != agent_name]
                    effort = agent.compete_for_resource(resource, opponents)
                    efforts[agent_name] = effort
                    print(f"   - {agent_name}: effort {effort:.1f}")
                
                # Winner is highest effort
                if efforts:
                    winner = max(efforts, key=efforts.get)
                    winner_effort = efforts[winner]
                    
                    # Award resource
                    resource.owner = winner
                    self.agents[winner].resources.append(resource)
                    self.agents[winner].score += resource.value
                    
                    print(f"   āœ“ Winner: {winner} (effort: {winner_effort:.1f})")
                    
                    # Record results
                    for agent_name, effort in efforts.items():
                        is_winner = agent_name == winner
                        score_change = resource.value if is_winner else -effort * 0.1  # Penalty for losing
                        
                        result = CompetitionResult(
                            round_number=self.round_number,
                            agent_name=agent_name,
                            action=f"Competed for {resource.name}",
                            success=is_winner,
                            score_change=score_change
                        )
                        results.append(result)
                        self.agents[agent_name].history.append(result)
            
            return results
        
        def get_leaderboard(self) -> List[tuple[str, float]]:
            """Get current leaderboard."""
            scores = [(agent.name, agent.score) for agent in self.agents.values()]
            return sorted(scores, key=lambda x: x[1], reverse=True)
        
        def strategic_interaction(self):
            """Have agents make strategic decisions."""
            print(f"\n   Strategic Planning Round {self.round_number}...")
            
            # Get all scores
            scores = {agent.name: agent.score for agent in self.agents.values()}
            
            # Each agent makes a decision
            for agent in self.agents.values():
                other_scores = {n: s for n, s in scores.items() if n != agent.name}
                
                situation = f"Resource competition, {len(self.resources)} resources available"
                decision = agent.make_strategic_decision(situation, other_scores)
                
                print(f"   - {agent.name}: {decision}")
        
        def run(self, scenario: str, num_rounds: int = 3) -> Dict[str, Any]:
            """Execute competitive agent pattern."""
            
            for round_num in range(num_rounds):
                results = self.run_competition_round()
                
                # Strategic interaction every other round
                if round_num % 2 == 1:
                    self.strategic_interaction()
            
            # Final leaderboard
            leaderboard = self.get_leaderboard()
            
            return {
                "pattern": "CompetitiveAgent",
                "competition_type": self.competition_type.value,
                "total_agents": len(self.agents),
                "rounds_played": self.round_number,
                "winner": leaderboard[0][0] if leaderboard else None,
                "final_scores": dict(leaderboard),
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Competitive Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = CompetitiveAgentPattern(CompetitionType.CONTINUOUS)
        
        # Create agents with different strategies
        print("1. Creating competitive agents...")
        
        aggressive = CompetitiveAgent(
            "AggressiveCorp",
            CompetitiveStrategy("aggressive", aggression=0.9, risk_tolerance=0.8, cooperation_willingness=0.2),
            initial_score=100
        )
        
        balanced = CompetitiveAgent(
            "BalancedInc",
            CompetitiveStrategy("balanced", aggression=0.5, risk_tolerance=0.5, cooperation_willingness=0.5),
            initial_score=100
        )
        
        conservative = CompetitiveAgent(
            "ConservativeLLC",
            CompetitiveStrategy("conservative", aggression=0.3, risk_tolerance=0.2, cooperation_willingness=0.7),
            initial_score=100
        )
        
        for agent in [aggressive, balanced, conservative]:
            pattern.register_agent(agent)
            print(f"   - {agent.name}: {agent.strategy.name} strategy")
        
        # Add resources to compete for
        print("\n2. Adding resources...")
        
        resources = [
            CompetitiveResource("r1", "Contract A", value=50, quantity=1),
            CompetitiveResource("r2", "Contract B", value=75, quantity=1),
            CompetitiveResource("r3", "Contract C", value=100, quantity=1),
            CompetitiveResource("r4", "Contract D", value=60, quantity=1),
        ]
        
        for res in resources:
            pattern.add_resource(res)
        
        # Run competition
        print("\n3. Running competition...")
        summary = pattern.run("Market competition for contracts", num_rounds=2)
        
        # Final results
        print("\n4. Final Leaderboard:")
        for rank, (agent_name, score) in enumerate(summary["final_scores"].items(), 1):
            agent = pattern.agents[agent_name]
            resources_won = len(agent.resources)
            print(f"   {rank}. {agent_name}: {score:.1f} points ({resources_won} resources)")
        
        print(f"\n5. Competition Summary:")
        print(f"   Winner: {summary['winner']}")
        print(f"   Rounds: {summary['rounds_played']}")
        print(f"   Type: {summary['competition_type']}")
        
        print("\nāœ“ Competitive agents battled for resources and market share!")
    
        

    Agentic Architecture Pattern: Competitive Agent

    Pattern 22: 22 Competitive Agent – Line-by-Line Explanation

    Overview

    Pattern 22: Competitive Agent Pattern

    Agents compete for resources, tasks, or market share. Features strategies, performance metrics, and competitive dynamics.


    Code Structure Analysis

    File Information

    • Pattern Number: 22
    • Pattern Name: 22 Competitive Agent
    • File Path: patterns\collaboration\22_competitive_agent.py
    • Total Lines: 286

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 22: Competitive Agent Pattern
    
    Agents compete for resources, tasks, or market share.
    Features strategies, performance metrics, and competitive dynamics.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 22 Competitive Agent pattern.


    Imports Section

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

    CompetitionType (Line 17)

    class CompetitionType(Enum):
        ZERO_SUM
        NON_ZERO_SUM
        TOURNAMENT
        CONTINUOUS
    

    Purpose: Type of competition.

    Values: – ZERO_SUM: Represents zero sum – NON_ZERO_SUM: Represents non zero sum – TOURNAMENT: Represents tournament – CONTINUOUS: Represents continuous


    Classes

    CompetitionType (Line 17)

    Description: Type of competition.

    CompetitiveResource (Line 26)

    Description: Resource being competed for.

    CompetitiveStrategy (Line 36)

    Description: Strategy used by competitive agent.

    CompetitionResult (Line 45)

    Description: Result of competitive action.

    CompetitiveAgent (Line 55)

    Description: Agent that competes with others.

    Methods:

    __init__ (Line 58)
    def __init__(self, name, strategy, initial_score, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    compete_for_resource (Line 67)
    def compete_for_resource(self, resource, opponents):
        """Compete for a resource, returns bid/effort."""
    

    Purpose: Compete for a resource, returns bid/effort.

    Parameters: – self: Instance reference – resource: Parameter for resource – opponents: Parameter for opponents

    make_strategic_decision (Line 83)
    def make_strategic_decision(self, situation, competitor_scores):
        """Make strategic decision using LLM."""
    

    Purpose: Make strategic decision using LLM.

    Parameters: – self: Instance reference – situation: Parameter for situation – competitor_scores: Parameter for competitor_scores

    CompetitiveAgentPattern (Line 106)

    Description: Competitive Agent Pattern – Agent competition and rivalry.

    Agents compete for resources using different strategies. Supports various competition types and dynamic interactions.

    Methods:

    __init__ (Line 114)
    def __init__(self, competition_type):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    register_agent (Line 120)
    def register_agent(self, agent):
        """Register agent for competition."""
    

    Purpose: Register agent for competition.

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

    add_resource (Line 124)
    def add_resource(self, resource):
        """Add resource to compete for."""
    

    Purpose: Add resource to compete for.

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

    run_competition_round (Line 128)
    def run_competition_round(self):
        """Run one round of competition."""
    

    Purpose: Run one round of competition.

    Parameters: – self: Instance reference

    get_leaderboard (Line 179)
    def get_leaderboard(self):
        """Get current leaderboard."""
    

    Purpose: Get current leaderboard.

    Parameters: – self: Instance reference

    strategic_interaction (Line 184)
    def strategic_interaction(self):
        """Have agents make strategic decisions."""
    

    Purpose: Have agents make strategic decisions.

    Parameters: – self: Instance reference

    run (Line 200)
    def run(self, scenario, num_rounds):
        """Execute competitive agent pattern."""
    

    Purpose: Execute competitive agent pattern.

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


    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 = CompetitionType(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

  • Agentic Architecture Pattern: Cooperative Agent

    
        """
    Pattern 21: Cooperative Agent Pattern
    
    Agents work together toward shared goals, sharing resources and information
    to achieve outcomes that benefit the group.
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    @dataclass
    class SharedGoal:
        """Goal shared among cooperative agents."""
        goal_id: str
        description: str
        target: Any
        contributors: List[str] = field(default_factory=list)
        contributions: Dict[str, Any] = field(default_factory=dict)
        status: str = "active"  # active, achieved, failed
        progress: float = 0.0
    
    
    @dataclass
    class Contribution:
        """Contribution from an agent."""
        agent_name: str
        goal_id: str
        contribution_type: str
        data: Any
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class CooperativeAgent:
        """Agent that cooperates with others toward shared goals."""
        
        def __init__(self, name: str, role: str, skills: List[str], model_name: str = "gpt-4"):
            self.name = name
            self.role = role
            self.skills = skills
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.pattern: Optional['CooperativeAgentPattern'] = None
        
        def contribute_to_goal(self, goal: SharedGoal) -> Contribution:
            """Contribute to shared goal based on skills."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, a {self.role} with skills in {', '.join(self.skills)}. "
                          f"Work cooperatively to achieve the shared goal. "
                          f"Provide your contribution based on your skills."),
                ("human", f"Shared Goal: {goal.description}\n"
                         f"Target: {goal.target}\n"
                         f"Other Contributors: {', '.join(goal.contributors)}\n"
                         f"Current Progress: {goal.progress:.0%}\n\n"
                         f"What is your contribution?")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            
            contribution = Contribution(
                agent_name=self.name,
                goal_id=goal.goal_id,
                contribution_type=self.skills[0] if self.skills else "general",
                data=str(response.content)
            )
            
            return contribution
        
        def share_resource(self, resource_name: str, resource_data: Any):
            """Share resource with team."""
            if self.pattern:
                self.pattern.shared_resources[resource_name] = {
                    "owner": self.name,
                    "data": resource_data,
                    "timestamp": datetime.now()
                }
        
        def request_help(self, task: str) -> List[str]:
            """Request help from other agents."""
            if not self.pattern:
                return []
            
            responses = []
            for agent in self.pattern.agents.values():
                if agent.name != self.name:
                    # Ask if agent can help
                    prompt = ChatPromptTemplate.from_messages([
                        ("system", f"You are {agent.name}. {self.name} is requesting help. "
                                  f"Based on your skills ({', '.join(agent.skills)}), can you help?"),
                        ("human", f"Request: {task}\n\nCan you help? Respond yes/no and briefly explain how.")
                    ])
                    
                    chain = prompt | agent.llm
                    response = chain.invoke({})
                    responses.append(f"{agent.name}: {str(response.content)[:100]}")
            
            return responses
    
    
    class CooperativeAgentPattern:
        """
        Cooperative Agent Pattern - Team collaboration for shared goals.
        
        Agents work together, share resources, and contribute their
        skills toward achieving common objectives.
        """
        
        def __init__(self):
            self.agents: Dict[str, CooperativeAgent] = {}
            self.goals: Dict[str, SharedGoal] = {}
            self.shared_resources: Dict[str, Dict[str, Any]] = {}
            self.contribution_history: List[Contribution] = []
        
        def register_agent(self, agent: CooperativeAgent):
            """Register cooperative agent."""
            self.agents[agent.name] = agent
            agent.pattern = self
        
        def add_shared_goal(self, goal: SharedGoal):
            """Add shared goal for team."""
            self.goals[goal.goal_id] = goal
        
        def work_on_goal(self, goal_id: str):
            """Have all agents contribute to goal."""
            if goal_id not in self.goals:
                return
            
            goal = self.goals[goal_id]
            print(f"\n   Working on goal: {goal.description}")
            
            for agent in self.agents.values():
                print(f"   - {agent.name} contributing...")
                
                # Agent contributes
                contribution = agent.contribute_to_goal(goal)
                self.contribution_history.append(contribution)
                
                # Record contribution
                if agent.name not in goal.contributors:
                    goal.contributors.append(agent.name)
                goal.contributions[agent.name] = contribution.data
                
                # Update progress
                goal.progress = len(goal.contributors) / max(len(self.agents), 1)
            
            # Check if goal achieved
            if goal.progress >= 1.0:
                goal.status = "achieved"
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute cooperative work scenario."""
            return {
                "pattern": "Cooperative",
                "total_agents": len(self.agents),
                "shared_goals": len(self.goals),
                "achieved_goals": sum(1 for g in self.goals.values() if g.status == "achieved"),
                "total_contributions": len(self.contribution_history),
                "shared_resources": len(self.shared_resources),
                "scenario": scenario
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Cooperative Agent Pattern Demo ===\n")
        
        # Create pattern
        pattern = CooperativeAgentPattern()
        
        # Create cooperative agents
        print("1. Creating cooperative team...")
        researcher = CooperativeAgent("Alice", "researcher", ["research", "analysis"])
        developer = CooperativeAgent("Bob", "developer", ["coding", "testing"])
        designer = CooperativeAgent("Carol", "designer", ["UI/UX", "graphics"])
        writer = CooperativeAgent("Dave", "writer", ["documentation", "communication"])
        
        # Register agents
        for agent in [researcher, developer, designer, writer]:
            pattern.register_agent(agent)
        
        # Add shared goal
        print("\n2. Setting shared goal...")
        goal = SharedGoal(
            goal_id="proj1",
            description="Create a user-friendly mobile app",
            target="Launch MVP in 3 months"
        )
        pattern.add_shared_goal(goal)
        
        # Work on goal cooperatively
        print("\n3. Agents working cooperatively...")
        pattern.work_on_goal("proj1")
        
        # Agent requests help
        print("\n4. Agent requesting help from team...")
        responses = developer.request_help("Need help optimizing database queries")
        for response in responses[:2]:  # Show first 2 responses
            print(f"   {response}...")
        
        # Share resource
        print("\n5. Agent sharing resource...")
        researcher.share_resource("market_research", {"users": 1000, "feedback": "positive"})
        
        # Results
        result = pattern.run("Team collaboration")
        print(f"\n6. Cooperation Results:")
        print(f"   Team Size: {result['total_agents']}")
        print(f"   Goals Achieved: {result['achieved_goals']}/{result['shared_goals']}")
        print(f"   Total Contributions: {result['total_contributions']}")
        print(f"   Shared Resources: {result['shared_resources']}")
        
        print("\nāœ“ Cooperative agents worked together toward shared goals!")
    
        

    Agentic Architecture Pattern: Cooperative Agent

    Pattern 21: 21 Cooperative Agent – Line-by-Line Explanation

    Overview

    Pattern 21: Cooperative Agent Pattern

    Agents work together toward shared goals, sharing resources and information to achieve outcomes that benefit the group.


    Code Structure Analysis

    File Information

    • Pattern Number: 21
    • Pattern Name: 21 Cooperative Agent
    • File Path: patterns\collaboration\21_cooperative_agent.py
    • Total Lines: 216

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 21: Cooperative Agent Pattern
    
    Agents work together toward shared goals, sharing resources and information
    to achieve outcomes that benefit the group.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 21 Cooperative Agent pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Classes

    SharedGoal (Line 16)

    Description: Goal shared among cooperative agents.

    Contribution (Line 28)

    Description: Contribution from an agent.

    CooperativeAgent (Line 37)

    Description: Agent that cooperates with others toward shared goals.

    Methods:

    __init__ (Line 40)
    def __init__(self, name, role, skills, model_name):
        """No method docstring."""
    

    Purpose: No method docstring.

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

    contribute_to_goal (Line 47)
    def contribute_to_goal(self, goal):
        """Contribute to shared goal based on skills."""
    

    Purpose: Contribute to shared goal based on skills.

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

    share_resource (Line 72)
    def share_resource(self, resource_name, resource_data):
        """Share resource with team."""
    

    Purpose: Share resource with team.

    Parameters: – self: Instance reference – resource_name: Parameter for resource_name – resource_data: Parameter for resource_data

    request_help (Line 81)
    def request_help(self, task):
        """Request help from other agents."""
    

    Purpose: Request help from other agents.

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

    CooperativeAgentPattern (Line 103)

    Description: Cooperative Agent Pattern – Team collaboration for shared goals.

    Agents work together, share resources, and contribute their skills toward achieving common objectives.

    Methods:

    __init__ (Line 111)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 117)
    def register_agent(self, agent):
        """Register cooperative agent."""
    

    Purpose: Register cooperative agent.

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

    add_shared_goal (Line 122)
    def add_shared_goal(self, goal):
        """Add shared goal for team."""
    

    Purpose: Add shared goal for team.

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

    work_on_goal (Line 126)
    def work_on_goal(self, goal_id):
        """Have all agents contribute to goal."""
    

    Purpose: Have all agents contribute to goal.

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

    run (Line 153)
    def run(self, scenario):
        """Execute cooperative work scenario."""
    

    Purpose: Execute cooperative work scenario.

    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 = SharedGoal(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

  • Agentic Architecture Pattern: Request Response

    
        """
    Pattern 20: Request-Response Pattern
    
    Synchronous communication where requestor waits for response from responder.
    Implements classic client-server request-response with timeout handling.
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict
    
    
    class RequestStatus(Enum):
        """Status of request."""
        PENDING = "pending"
        PROCESSING = "processing"
        COMPLETED = "completed"
        FAILED = "failed"
        TIMEOUT = "timeout"
    
    
    @dataclass
    class Request:
        """Request from client to server."""
        request_id: str
        sender: str
        receiver: str
        operation: str
        payload: Any
        timestamp: datetime = field(default_factory=datetime.now)
        timeout: float = 30.0
    
    
    @dataclass
    class Response:
        """Response from server to client."""
        request_id: str
        sender: str
        receiver: str
        result: Any
        status: RequestStatus
        error: Optional[str] = None
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class RequestResponseState(TypedDict):
        """State for request-response workflow."""
        request: Optional[Request]
        response: Optional[Response]
        status: RequestStatus
    
    
    class RequestResponseAgent:
        """Agent with request-response capabilities."""
        
        def __init__(self, name: str, role: str, model_name: str = "gpt-4"):
            self.name = name
            self.role = role
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.pattern: Optional['RequestResponsePattern'] = None
        
        def request(self, receiver: str, operation: str, payload: Any, timeout: float = 30.0) -> Response:
            """Send synchronous request and wait for response."""
            if not self.pattern:
                raise ValueError("Agent not connected to pattern")
            
            request_id = f"{self.name}_{datetime.now().timestamp()}"
            request = Request(
                request_id=request_id,
                sender=self.name,
                receiver=receiver,
                operation=operation,
                payload=payload,
                timeout=timeout
            )
            
            return self.pattern.process_request(request)
        
        def handle_request(self, request: Request) -> Response:
            """Handle incoming request and generate response."""
            try:
                # Process request using LLM
                prompt = ChatPromptTemplate.from_messages([
                    ("system", f"You are {self.name}, a {self.role}. "
                              f"Process this request and provide a response."),
                    ("human", f"Operation: {request.operation}\n"
                             f"From: {request.sender}\n"
                             f"Payload: {request.payload}")
                ])
                
                chain = prompt | self.llm
                result = chain.invoke({})
                
                return Response(
                    request_id=request.request_id,
                    sender=self.name,
                    receiver=request.sender,
                    result=str(result.content),
                    status=RequestStatus.COMPLETED
                )
                
            except Exception as e:
                return Response(
                    request_id=request.request_id,
                    sender=self.name,
                    receiver=request.sender,
                    result=None,
                    status=RequestStatus.FAILED,
                    error=str(e)
                )
    
    
    class RequestResponsePattern:
        """
        Request-Response Pattern - Synchronous agent communication.
        
        Uses LangGraph to model the request-response workflow with
        timeout handling and error recovery.
        """
        
        def __init__(self):
            self.agents: Dict[str, RequestResponseAgent] = {}
            self.request_log: List[Request] = []
            self.response_log: List[Response] = []
            self.graph = self._build_graph()
        
        def register_agent(self, agent: RequestResponseAgent):
            """Register agent with pattern."""
            self.agents[agent.name] = agent
            agent.pattern = self
        
        def _build_graph(self) -> StateGraph:
            """Build request-response workflow graph."""
            workflow = StateGraph(RequestResponseState)
            
            def route_request(state: RequestResponseState) -> RequestResponseState:
                """Route request to target agent."""
                request = state["request"]
                if not request:
                    state["status"] = RequestStatus.FAILED
                    return state
                
                state["status"] = RequestStatus.PROCESSING
                
                if request.receiver in self.agents:
                    receiver = self.agents[request.receiver]
                    response = receiver.handle_request(request)
                    state["response"] = response
                    state["status"] = response.status
                else:
                    state["response"] = Response(
                        request_id=request.request_id,
                        sender="system",
                        receiver=request.sender,
                        result=None,
                        status=RequestStatus.FAILED,
                        error=f"Receiver {request.receiver} not found"
                    )
                    state["status"] = RequestStatus.FAILED
                
                return state
            
            workflow.add_node("process", route_request)
            workflow.set_entry_point("process")
            workflow.add_edge("process", END)
            
            return workflow.compile()
        
        def process_request(self, request: Request) -> Response:
            """Process request through pattern."""
            self.request_log.append(request)
            
            initial_state: RequestResponseState = {
                "request": request,
                "response": None,
                "status": RequestStatus.PENDING
            }
            
            # Execute workflow
            try:
                final_state = self.graph.invoke(initial_state)
                response = final_state["response"]
                
                if response:
                    self.response_log.append(response)
                    return response
                else:
                    # Timeout or error
                    timeout_response = Response(
                        request_id=request.request_id,
                        sender="system",
                        receiver=request.sender,
                        result=None,
                        status=RequestStatus.TIMEOUT,
                        error="Request timed out"
                    )
                    self.response_log.append(timeout_response)
                    return timeout_response
                    
            except Exception as e:
                error_response = Response(
                    request_id=request.request_id,
                    sender="system",
                    receiver=request.sender,
                    result=None,
                    status=RequestStatus.FAILED,
                    error=str(e)
                )
                self.response_log.append(error_response)
                return error_response
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get pattern statistics."""
            return {
                "total_requests": len(self.request_log),
                "total_responses": len(self.response_log),
                "completed": sum(1 for r in self.response_log if r.status == RequestStatus.COMPLETED),
                "failed": sum(1 for r in self.response_log if r.status == RequestStatus.FAILED),
                "timeout": sum(1 for r in self.response_log if r.status == RequestStatus.TIMEOUT)
            }
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute request-response scenario."""
            return {
                "pattern": "RequestResponse",
                "scenario": scenario,
                **self.get_statistics()
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Request-Response Pattern Demo ===\n")
        
        # Create pattern
        pattern = RequestResponsePattern()
        
        # Create agents
        client = RequestResponseAgent("Client", "service client")
        server = RequestResponseAgent("Server", "data server")
        auth_service = RequestResponseAgent("AuthService", "authentication service")
        
        # Register agents
        pattern.register_agent(client)
        pattern.register_agent(server)
        pattern.register_agent(auth_service)
        
        print("1. Client making requests...")
        
        # Request 1: Authentication
        print("\n   Request 1: Authentication")
        response1 = client.request("AuthService", "authenticate", {"user": "john", "password": "***"})
        print(f"   Status: {response1.status.value}")
        if response1.status == RequestStatus.COMPLETED:
            print(f"   Result: {response1.result[:50]}...")
        
        # Request 2: Data query
        print("\n   Request 2: Data query")
        response2 = client.request("Server", "get_data", {"query": "SELECT * FROM users"})
        print(f"   Status: {response2.status.value}")
        if response2.status == RequestStatus.COMPLETED:
            print(f"   Result: {response2.result[:50]}...")
        
        # Request 3: Non-existent service (will fail)
        print("\n   Request 3: Non-existent service")
        response3 = client.request("InvalidService", "operation", {})
        print(f"   Status: {response3.status.value}")
        if response3.error:
            print(f"   Error: {response3.error}")
        
        # Statistics
        stats = pattern.get_statistics()
        print(f"\n2. Pattern Statistics:")
        print(f"   Total Requests: {stats['total_requests']}")
        print(f"   Completed: {stats['completed']}")
        print(f"   Failed: {stats['failed']}")
        print(f"   Timeout: {stats['timeout']}")
        
        result = pattern.run("Service communication")
        print(f"\nāœ“ Request-response pattern enabled synchronous agent communication!")
    
        

    Agentic Architecture Pattern: Request Response

    Pattern 20: 20 Request Response – Line-by-Line Explanation

    Overview

    Pattern 20: Request-Response Pattern

    Synchronous communication where requestor waits for response from responder. Implements classic client-server request-response with timeout handling.


    Code Structure Analysis

    File Information

    • Pattern Number: 20
    • Pattern Name: 20 Request Response
    • File Path: patterns\communication\20_request_response.py
    • Total Lines: 286

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 20: Request-Response Pattern
    
    Synchronous communication where requestor waits for response from responder.
    Implements classic client-server request-response with timeout handling.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 20 Request Response pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


    Enumerations

    RequestStatus (Line 18)

    class RequestStatus(Enum):
        PENDING
        PROCESSING
        COMPLETED
        FAILED
        TIMEOUT
    

    Purpose: Status of request.

    Values: – PENDING: Initial state, not yet processed – PROCESSING: Represents processing – COMPLETED: Successfully finished – FAILED: Execution failed – TIMEOUT: Represents timeout


    Classes

    RequestStatus (Line 18)

    Description: Status of request.

    Request (Line 28)

    Description: Request from client to server.

    Response (Line 40)

    Description: Response from server to client.

    RequestResponseState (Line 51)

    Description: State for request-response workflow.

    RequestResponseAgent (Line 58)

    Description: Agent with request-response capabilities.

    Methods:

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

    Purpose: No method docstring.

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

    request (Line 67)
    def request(self, receiver, operation, payload, timeout):
        """Send synchronous request and wait for response."""
    

    Purpose: Send synchronous request and wait for response.

    Parameters: – self: Instance reference – receiver: Parameter for receiver – operation: Parameter for operation – payload: Parameter for payload – timeout: Parameter for timeout

    handle_request (Line 84)
    def handle_request(self, request):
        """Handle incoming request and generate response."""
    

    Purpose: Handle incoming request and generate response.

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

    RequestResponsePattern (Line 118)

    Description: Request-Response Pattern – Synchronous agent communication.

    Uses LangGraph to model the request-response workflow with timeout handling and error recovery.

    Methods:

    __init__ (Line 126)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 132)
    def register_agent(self, agent):
        """Register agent with pattern."""
    

    Purpose: Register agent with pattern.

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

    _build_graph (Line 137)
    def _build_graph(self):
        """Build request-response workflow graph."""
    

    Purpose: Build request-response workflow graph.

    Parameters: – self: Instance reference

    process_request (Line 174)
    def process_request(self, request):
        """Process request through pattern."""
    

    Purpose: Process request through pattern.

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

    get_statistics (Line 217)
    def get_statistics(self):
        """Get pattern statistics."""
    

    Purpose: Get pattern statistics.

    Parameters: – self: Instance reference

    run (Line 227)
    def run(self, scenario):
        """Execute request-response scenario."""
    

    Purpose: Execute request-response scenario.

    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 = RequestStatus(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

  • Agentic Architecture Pattern: Event Driven

    
        """
    Pattern 19: Event-Driven Pattern
    
    Agents react to events in the system asynchronously.
    Uses event bus for decoupled event publishing and subscription.
    """
    
    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from collections import defaultdict
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    
    
    @dataclass
    class Event:
        """Event in the system."""
        event_type: str
        source: str
        data: Any
        timestamp: datetime = field(default_factory=datetime.now)
        priority: int = 1
    
    
    class EventDrivenAgent:
        """Agent that reacts to events."""
        
        def __init__(self, name: str, role: str, model_name: str = "gpt-4"):
            self.name = name
            self.role = role
            self.llm = ChatOpenAI(model=model_name, temperature=0.7)
            self.event_handlers: Dict[str, Callable] = {}
            self.event_bus: Optional['EventBus'] = None
        
        def subscribe(self, event_type: str, handler: Optional[Callable] = None):
            """Subscribe to event type."""
            if handler:
                self.event_handlers[event_type] = handler
            else:
                # Use default LLM-based handler
                self.event_handlers[event_type] = self._default_handler
            
            if self.event_bus:
                self.event_bus.subscribe(event_type, self)
        
        def _default_handler(self, event: Event) -> str:
            """Default event handler using LLM."""
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, a {self.role}. "
                          f"React to this event appropriately."),
                ("human", f"Event: {event.event_type}\n"
                         f"Source: {event.source}\n"
                         f"Data: {event.data}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            return str(response.content)
        
        def handle_event(self, event: Event) -> str:
            """Handle received event."""
            if event.event_type in self.event_handlers:
                return self.event_handlers[event.event_type](event)
            return f"{self.name}: Unhandled event {event.event_type}"
        
        def publish(self, event_type: str, data: Any, priority: int = 1):
            """Publish event to event bus."""
            if self.event_bus:
                event = Event(
                    event_type=event_type,
                    source=self.name,
                    data=data,
                    priority=priority
                )
                self.event_bus.publish(event)
    
    
    class EventBus:
        """Central event bus for event-driven communication."""
        
        def __init__(self):
            self.subscribers: Dict[str, List[EventDrivenAgent]] = defaultdict(list)
            self.event_log: List[Event] = []
        
        def subscribe(self, event_type: str, agent: EventDrivenAgent):
            """Subscribe agent to event type."""
            if agent not in self.subscribers[event_type]:
                self.subscribers[event_type].append(agent)
                agent.event_bus = self
        
        def unsubscribe(self, event_type: str, agent: EventDrivenAgent):
            """Unsubscribe agent from event type."""
            if agent in self.subscribers[event_type]:
                self.subscribers[event_type].remove(agent)
        
        def publish(self, event: Event):
            """Publish event to all subscribers."""
            self.event_log.append(event)
            
            # Notify all subscribers
            if event.event_type in self.subscribers:
                for agent in self.subscribers[event.event_type]:
                    response = agent.handle_event(event)
                    # Optionally create follow-up event from response
                    if response and "event:" in response.lower():
                        # Agent can trigger new events in response
                        pass
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get event bus statistics."""
            return {
                "total_events": len(self.event_log),
                "event_types": len(self.subscribers),
                "total_subscriptions": sum(len(agents) for agents in self.subscribers.values()),
                "events_by_type": {
                    event_type: sum(1 for e in self.event_log if e.event_type == event_type)
                    for event_type in self.subscribers.keys()
                }
            }
    
    
    class EventDrivenPattern:
        """
        Event-Driven Pattern - Asynchronous event-based communication.
        
        Agents publish events to a central bus and subscribe to events
        they're interested in, enabling loose coupling and reactivity.
        """
        
        def __init__(self):
            self.event_bus = EventBus()
        
        def register_agent(self, agent: EventDrivenAgent):
            """Register agent with event bus."""
            agent.event_bus = self.event_bus
        
        def run(self, scenario: str) -> Dict[str, Any]:
            """Execute event-driven scenario."""
            stats = self.event_bus.get_statistics()
            return {
                "pattern": "EventDriven",
                "scenario": scenario,
                **stats
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Event-Driven Pattern Demo ===\n")
        
        # Create pattern
        pattern = EventDrivenPattern()
        event_bus = pattern.event_bus
        
        # Create agents
        monitor = EventDrivenAgent("Monitor", "system monitor")
        logger = EventDrivenAgent("Logger", "event logger")
        alerter = EventDrivenAgent("Alerter", "alert manager")
        
        # Register and subscribe
        print("1. Setting up agents and subscriptions...")
        pattern.register_agent(monitor)
        pattern.register_agent(logger)
        pattern.register_agent(alerter)
        
        # Custom handler for critical events
        def critical_handler(event: Event) -> str:
            return f"ALERT: Critical event from {event.source}: {event.data}"
        
        monitor.subscribe("system_event")
        monitor.subscribe("user_action")
        logger.subscribe("system_event")
        logger.subscribe("user_action")
        logger.subscribe("error")
        alerter.subscribe("error", critical_handler)
        
        # Publish events
        print("\n2. Publishing events...")
        event_bus.publish(Event("system_event", "System", "Server started"))
        event_bus.publish(Event("user_action", "User1", "Login successful"))
        event_bus.publish(Event("error", "System", "Database connection failed", priority=5))
        
        # Statistics
        stats = event_bus.get_statistics()
        print(f"\n3. Event Bus Statistics:")
        print(f"   Total Events: {stats['total_events']}")
        print(f"   Event Types: {stats['event_types']}")
        print(f"   Subscriptions: {stats['total_subscriptions']}")
        print(f"   By Type: {stats['events_by_type']}")
        
        result = pattern.run("System monitoring")
        print(f"\nāœ“ Event-driven pattern enabled reactive agent communication!")
    
        

    Agentic Architecture Pattern: Event Driven

    Pattern 19: 19 Event Driven – Line-by-Line Explanation

    Overview

    Pattern 19: Event-Driven Pattern

    Agents react to events in the system asynchronously. Uses event bus for decoupled event publishing and subscription.


    Code Structure Analysis

    File Information

    • Pattern Number: 19
    • Pattern Name: 19 Event Driven
    • File Path: patterns\communication\19_event_driven.py
    • Total Lines: 194

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 19: Event-Driven Pattern
    
    Agents react to events in the system asynchronously.
    Uses event bus for decoupled event publishing and subscription.
    """
    

    Explanation: This module docstring describes the purpose and key features of the 19 Event Driven pattern.


    Imports Section

    from typing import Dict, Any, List, Optional, Callable
    from dataclasses import dataclass, field
    from datetime import datetime
    from collections import defaultdict
    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, Callable: 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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration


    Classes

    Event (Line 17)

    Description: Event in the system.

    EventDrivenAgent (Line 26)

    Description: Agent that reacts to events.

    Methods:

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

    Purpose: No method docstring.

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

    subscribe (Line 36)
    def subscribe(self, event_type, handler):
        """Subscribe to event type."""
    

    Purpose: Subscribe to event type.

    Parameters: – self: Instance reference – event_type: Parameter for event_type – handler: Parameter for handler

    _default_handler (Line 47)
    def _default_handler(self, event):
        """Default event handler using LLM."""
    

    Purpose: Default event handler using LLM.

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

    handle_event (Line 61)
    def handle_event(self, event):
        """Handle received event."""
    

    Purpose: Handle received event.

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

    publish (Line 67)
    def publish(self, event_type, data, priority):
        """Publish event to event bus."""
    

    Purpose: Publish event to event bus.

    Parameters: – self: Instance reference – event_type: Parameter for event_type – data: Input data to process – priority: Parameter for priority

    EventBus (Line 79)

    Description: Central event bus for event-driven communication.

    Methods:

    __init__ (Line 82)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    subscribe (Line 86)
    def subscribe(self, event_type, agent):
        """Subscribe agent to event type."""
    

    Purpose: Subscribe agent to event type.

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

    unsubscribe (Line 92)
    def unsubscribe(self, event_type, agent):
        """Unsubscribe agent from event type."""
    

    Purpose: Unsubscribe agent from event type.

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

    publish (Line 97)
    def publish(self, event):
        """Publish event to all subscribers."""
    

    Purpose: Publish event to all subscribers.

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

    get_statistics (Line 110)
    def get_statistics(self):
        """Get event bus statistics."""
    

    Purpose: Get event bus statistics.

    Parameters: – self: Instance reference

    EventDrivenPattern (Line 123)

    Description: Event-Driven Pattern – Asynchronous event-based communication.

    Agents publish events to a central bus and subscribe to events they’re interested in, enabling loose coupling and reactivity.

    Methods:

    __init__ (Line 131)
    def __init__(self):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference

    register_agent (Line 134)
    def register_agent(self, agent):
        """Register agent with event bus."""
    

    Purpose: Register agent with event bus.

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

    run (Line 138)
    def run(self, scenario):
        """Execute event-driven scenario."""
    

    Purpose: Execute event-driven scenario.

    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 = Event(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

  • Agentic Architecture Pattern: Tuple Space

    
        """
    Pattern 18: Tuple Space
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    
    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict
    
    
    @dataclass
    class TupleSpaceConfig:
        """Configuration for Tuple Space."""
        model_name: str = "gpt-4"
        temperature: float = 0.7
        max_iterations: int = 10
        enable_logging: bool = True
    
    
    @dataclass
    class TupleSpaceState:
        """State information for Tuple Space."""
        current_step: int = 0
        data: Dict[str, Any] = field(default_factory=dict)
        results: List[Any] = field(default_factory=list)
        metadata: Dict[str, Any] = field(default_factory=dict)
        timestamp: datetime = field(default_factory=datetime.now)
    
    
    class TupleSpaceAgent:
        """Agent implementing Tuple Space."""
        
        def __init__(self, name: str, config: TupleSpaceConfig = None):
            self.name = name
            self.config = config or TupleSpaceConfig()
            self.llm = ChatOpenAI(
                model=self.config.model_name,
                temperature=self.config.temperature
            )
            self.state = TupleSpaceState()
            self.execution_history: List[Dict[str, Any]] = []
        
        def process(self, input_data: Any) -> Dict[str, Any]:
            """Process input according to Tuple Space."""
            self.state.current_step += 1
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, implementing Tuple Space. "
                          f"Provide clear, structured responses."),
                ("human", "Process this input: {input_data}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({"input_data": str(input_data)})
            
            result = {
                "agent": self.name,
                "step": self.state.current_step,
                "input": input_data,
                "output": str(response.content),
                "timestamp": datetime.now()
            }
            
            self.execution_history.append(result)
            self.state.results.append(result)
            
            return result
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get agent statistics."""
            return {
                "name": self.name,
                "total_executions": len(self.execution_history),
                "current_step": self.state.current_step,
                "results_count": len(self.state.results)
            }
    
    
    class TupleSpacePattern:
        """
        Tuple Space - Enhanced implementation
        
        This pattern provides comprehensive functionality for tuple space
        with full LangChain/LangGraph integration.
        
        Features:
        - Multi-agent coordination
        - State management
        - Execution tracking
        - LangGraph workflow integration
        """
        
        def __init__(self, config: TupleSpaceConfig = None):
            self.config = config or TupleSpaceConfig()
            self.agents: Dict[str, TupleSpaceAgent] = {}
            self.workflow = self._build_workflow()
            self.execution_log: List[Dict[str, Any]] = []
        
        def _build_workflow(self) -> StateGraph:
            """Build LangGraph workflow for pattern execution."""
            
            class WorkflowState(TypedDict):
                input: str
                processed: str
                output: str
                metadata: Dict[str, Any]
            
            def process_input(state: WorkflowState) -> WorkflowState:
                """Process input node."""
                return {
                    **state,
                    "processed": f"Processed: {state['input']}",
                    "metadata": {"step": "process_input"}
                }
            
            def generate_output(state: WorkflowState) -> WorkflowState:
                """Generate output node."""
                llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
                
                prompt = ChatPromptTemplate.from_messages([
                    ("system", "You are implementing Tuple Space. Generate comprehensive output."),
                    ("human", "{input}")
                ])
                
                chain = prompt | llm
                response = chain.invoke({"input": state["processed"]})
                
                return {
                    **state,
                    "output": str(response.content),
                    "metadata": {**state.get("metadata", {}), "step": "generate_output"}
                }
            
            # Build workflow graph
            workflow = StateGraph(WorkflowState)
            workflow.add_node("process", process_input)
            workflow.add_node("generate", generate_output)
            workflow.set_entry_point("process")
            workflow.add_edge("process", "generate")
            workflow.add_edge("generate", END)
            
            return workflow.compile()
        
        def register_agent(self, agent: TupleSpaceAgent):
            """Register an agent with the pattern."""
            self.agents[agent.name] = agent
            if self.config.enable_logging:
                print(f"Registered agent: {agent.name}")
        
        def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
            """Execute the pattern with given input."""
            
            # Use workflow if no specific agent requested
            if agent_name is None:
                result = self.workflow.invoke({
                    "input": str(input_data),
                    "processed": "",
                    "output": "",
                    "metadata": {}
                })
                
                execution_record = {
                    "pattern": "Tuple Space",
                    "input": input_data,
                    "output": result.get("output", ""),
                    "metadata": result.get("metadata", {}),
                    "timestamp": datetime.now()
                }
                
                self.execution_log.append(execution_record)
                return execution_record
            
            # Execute with specific agent
            agent = self.agents.get(agent_name)
            if not agent:
                raise ValueError(f"Agent {agent_name} not found")
            
            result = agent.process(input_data)
            self.execution_log.append(result)
            return result
        
        def run(self, input_data: Any) -> Dict[str, Any]:
            """Main execution method."""
            return self.execute(input_data)
        
        def get_statistics(self) -> Dict[str, Any]:
            """Get pattern statistics."""
            return {
                "pattern": "Tuple Space",
                "total_executions": len(self.execution_log),
                "registered_agents": len(self.agents),
                "agents": [agent.get_statistics() for agent in self.agents.values()]
            }
    
    
    # Example usage
    if __name__ == "__main__":
        print("=== Tuple Space Demo ===\n")
        
        # Create configuration
        config = TupleSpaceConfig(
            model_name="gpt-4",
            temperature=0.7,
            enable_logging=True
        )
        
        # Create pattern
        pattern = TupleSpacePattern(config)
        
        # Create and register agents
        agent1 = TupleSpaceAgent("Agent1", config)
        agent2 = TupleSpaceAgent("Agent2", config)
        pattern.register_agent(agent1)
        pattern.register_agent(agent2)
        
        print("\nExecuting pattern workflow...")
        
        # Execute with workflow
        result1 = pattern.run("Demonstrate tuple space functionality")
        print(f"\nWorkflow Result:")
        print(f"  Output: {result1['output'][:150]}...")
        
        # Execute with specific agents
        result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
        print(f"\nAgent1 Result:")
        print(f"  Output: {result2['output'][:150]}...")
        
        result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
        print(f"\nAgent2 Result:")
        print(f"  Output: {result3['output'][:150]}...")
        
        # Show statistics
        print("\n=== Statistics ===")
        stats = pattern.get_statistics()
        print(f"Pattern: {stats['pattern']}")
        print(f"Total Executions: {stats['total_executions']}")
        print(f"Registered Agents: {stats['registered_agents']}")
        print(f"\nAgent Statistics:")
        for agent_stats in stats['agents']:
            print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")
    
        

    Agentic Architecture Pattern: Tuple Space

    Pattern 18: 18 Tuple Space – Line-by-Line Explanation

    Overview

    Pattern 18: Tuple Space

    Purpose: Enhanced implementation with full LangChain/LangGraph integration


    Code Structure Analysis

    File Information

    • Pattern Number: 18
    • Pattern Name: 18 Tuple Space
    • File Path: patterns\communication\18_tuple_space.py
    • Total Lines: 247

    Detailed Code Explanation

    Module Docstring

    """
    Pattern 18: Tuple Space
    
    Purpose: Enhanced implementation with full LangChain/LangGraph integration
    """
    

    Explanation: This module docstring describes the purpose and key features of the 18 Tuple Space pattern.


    Imports Section

    from typing import Dict, Any, List, Optional
    from dataclasses import dataclass, field
    from datetime import datetime
    from enum import Enum
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_openai import ChatOpenAI
    from langgraph.graph import StateGraph, END
    from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation


    Classes

    TupleSpaceConfig (Line 18)

    Description: Configuration for Tuple Space.

    TupleSpaceState (Line 27)

    Description: State information for Tuple Space.

    TupleSpaceAgent (Line 36)

    Description: Agent implementing Tuple Space.

    Methods:

    __init__ (Line 39)
    def __init__(self, name, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

    process (Line 49)
    def process(self, input_data):
        """Process input according to Tuple Space."""
    

    Purpose: Process input according to Tuple Space.

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

    get_statistics (Line 75)
    def get_statistics(self):
        """Get agent statistics."""
    

    Purpose: Get agent statistics.

    Parameters: – self: Instance reference

    TupleSpacePattern (Line 85)

    Description: Tuple Space – Enhanced implementation

    This pattern provides comprehensive functionality for tuple space with full LangChain/LangGraph integration.

    Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

    Methods:

    __init__ (Line 99)
    def __init__(self, config):
        """No method docstring."""
    

    Purpose: No method docstring.

    Parameters: – self: Instance reference – config: Configuration settings

    _build_workflow (Line 105)
    def _build_workflow(self):
        """Build LangGraph workflow for pattern execution."""
    

    Purpose: Build LangGraph workflow for pattern execution.

    Parameters: – self: Instance reference

    register_agent (Line 150)
    def register_agent(self, agent):
        """Register an agent with the pattern."""
    

    Purpose: Register an agent with the pattern.

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

    execute (Line 156)
    def execute(self, input_data, agent_name):
        """Execute the pattern with given input."""
    

    Purpose: Execute the pattern with given input.

    Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

    run (Line 188)
    def run(self, input_data):
        """Main execution method."""
    

    Purpose: Main execution method.

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

    get_statistics (Line 192)
    def get_statistics(self):
        """Get pattern statistics."""
    

    Purpose: Get pattern statistics.

    Parameters: – self: Instance reference

    WorkflowState (Line 108)

    Description: No class docstring.


    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 = TupleSpaceConfig(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