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