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