"""
Pattern 12: Contract Net Protocol
A negotiation protocol where agents bid for tasks.
"""

from typing import Dict, Any, List, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from enum import Enum


class BidStatus(str, Enum):
    PENDING = "pending"
    ACCEPTED = "accepted"
    REJECTED = "rejected"


class Bid(BaseModel):
    """Represents a bid from an agent."""
    agent_id: str
    task_id: str
    cost: float
    time_estimate: float
    quality_score: float
    proposal: str
    status: BidStatus = BidStatus.PENDING


class Task(BaseModel):
    """Represents a task to be contracted."""
    id: str
    description: str
    requirements: List[str]
    deadline: Optional[float] = None
    budget: Optional[float] = None


class ContractNetPattern:
    """
    Contract Net Protocol - Task allocation through bidding.
    Manager announces tasks, contractors bid, manager awards contracts.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.tasks: Dict[str, Task] = {}
        self.bids: Dict[str, List[Bid]] = {}
        self.contractors: Dict[str, 'ContractorAgent'] = {}
    
    def announce_task(self, task: Task) -> str:
        """Manager announces a task."""
        self.tasks[task.id] = task
        self.bids[task.id] = []
        return task.id
    
    def register_contractor(self, agent_id: str, capabilities: List[str]) -> 'ContractorAgent':
        """Register a contractor agent."""
        contractor = ContractorAgent(agent_id, capabilities, self, self.llm)
        self.contractors[agent_id] = contractor
        return contractor
    
    def receive_bid(self, bid: Bid):
        """Receive a bid from a contractor."""
        if bid.task_id in self.bids:
            self.bids[bid.task_id].append(bid)
    
    def evaluate_bids(self, task_id: str) -> Optional[Bid]:
        """Evaluate bids and select the best one."""
        if task_id not in self.bids or not self.bids[task_id]:
            return None
        
        task = self.tasks[task_id]
        bids = self.bids[task_id]
        
        # Simple evaluation: select lowest cost with acceptable quality
        acceptable_bids = [b for b in bids if b.quality_score >= 7.0]
        if acceptable_bids:
            best_bid = min(acceptable_bids, key=lambda b: b.cost)
        else:
            best_bid = min(bids, key=lambda b: b.cost)
        
        best_bid.status = BidStatus.ACCEPTED
        return best_bid
    
    def award_contract(self, task_id: str) -> Dict[str, Any]:
        """Award contract to the best bidder."""
        winning_bid = self.evaluate_bids(task_id)
        
        if not winning_bid:
            return {"success": False, "message": "No bids received"}
        
        # Mark other bids as rejected
        for bid in self.bids[task_id]:
            if bid != winning_bid:
                bid.status = BidStatus.REJECTED
        
        return {
            "success": True,
            "task_id": task_id,
            "winner": winning_bid.agent_id,
            "cost": winning_bid.cost,
            "time_estimate": winning_bid.time_estimate,
            "all_bids": len(self.bids[task_id])
        }


class ContractorAgent:
    """Agent that bids on tasks."""
    
    def __init__(self, agent_id: str, capabilities: List[str], 
                 contract_net: ContractNetPattern, llm):
        self.agent_id = agent_id
        self.capabilities = capabilities
        self.contract_net = contract_net
        self.llm = llm
    
    def submit_bid(self, task: Task) -> Bid:
        """Submit a bid for a task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are contractor {agent_id} with capabilities: {capabilities}
            
            Create a bid for this task. Provide cost (dollars), time (hours), quality score (1-10).
            Task: {task_description}
            Requirements: {requirements}"""),
            ("human", "Provide: Cost: X, Time: Y, Quality: Z, Proposal: [text]")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "agent_id": self.agent_id,
            "capabilities": ", ".join(self.capabilities),
            "task_description": task.description,
            "requirements": ", ".join(task.requirements)
        })
        
        # Parse response (simplified)
        import random
        bid = Bid(
            agent_id=self.agent_id,
            task_id=task.id,
            cost=random.uniform(500, 5000),
            time_estimate=random.uniform(10, 50),
            quality_score=random.uniform(6, 10),
            proposal=response.content[:200]
        )
        
        self.contract_net.receive_bid(bid)
        return bid


# Example usage
if __name__ == "__main__":
    cn = ContractNetPattern()
    
    # Register contractors
    c1 = cn.register_contractor("Agent_A", ["web_dev", "backend"])
    c2 = cn.register_contractor("Agent_B", ["web_dev", "frontend"])
    c3 = cn.register_contractor("Agent_C", ["fullstack", "devops"])
    
    # Announce task
    task = Task(
        id="task_001",
        description="Build REST API",
        requirements=["Python", "FastAPI"],
        budget=5000.0
    )
    
    cn.announce_task(task)
    
    # Submit bids
    c1.submit_bid(task)
    c2.submit_bid(task)
    c3.submit_bid(task)
    
    # Award
    result = cn.award_contract(task.id)
    print(f"Winner: {result['winner']}, Cost: ${result['cost']:.2f}")

    

Agentic Architecture Pattern: Contract Net

Pattern 12: 12 Contract Net – Line-by-Line Explanation

Overview

Pattern 12: Contract Net Protocol A negotiation protocol where agents bid for tasks.

Code Structure Analysis

File Information

• Pattern Number: 12
• Pattern Name: 12 Contract Net
• File Path: patterns\communication\12_contract_net.py
• Total Lines: 179

Detailed Code Explanation

Module Docstring

"""
Pattern 12: Contract Net Protocol
A negotiation protocol where agents bid for tasks.
"""

Explanation: This module docstring describes the purpose and key features of the 12 Contract Net pattern.

Imports Section

from typing import Dict, Any, List, Optional
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from enum import Enum
        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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from enum import Enum: Enumeration support for fixed sets of values

Enumerations

BidStatus (Line 13)

class BidStatus(Enum):
    PENDING
    ACCEPTED
    REJECTED

Purpose: No enum docstring.

Values: – PENDING: Initial state, not yet processed – ACCEPTED: Represents accepted – REJECTED: Represents rejected

Classes

BidStatus (Line 13)

Description: No class docstring.

Bid (Line 19)

Description: Represents a bid from an agent.

Task (Line 30)

Description: Represents a task to be contracted.

ContractNetPattern (Line 39)

Description: Contract Net Protocol – Task allocation through bidding. Manager announces tasks, contractors bid, manager awards contracts.

Methods:

__init__ (Line 45)

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”)

announce_task (Line 51)

def announce_task(self, task):
    """Manager announces a task."""

Purpose: Manager announces a task.

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

register_contractor (Line 57)

def register_contractor(self, agent_id, capabilities):
    """Register a contractor agent."""

Purpose: Register a contractor agent.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – capabilities: Parameter for capabilities

receive_bid (Line 63)

def receive_bid(self, bid):
    """Receive a bid from a contractor."""

Purpose: Receive a bid from a contractor.

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

evaluate_bids (Line 68)

def evaluate_bids(self, task_id):
    """Evaluate bids and select the best one."""

Purpose: Evaluate bids and select the best one.

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

award_contract (Line 86)

def award_contract(self, task_id):
    """Award contract to the best bidder."""

Purpose: Award contract to the best bidder.

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

ContractorAgent (Line 108)

Description: Agent that bids on tasks.

Methods:

__init__ (Line 111)

def __init__(self, agent_id, capabilities, contract_net, llm):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – capabilities: Parameter for capabilities – contract_net: Parameter for contract_net – llm: Parameter for llm

submit_bid (Line 118)

def submit_bid(self, task):
    """Submit a bid for a task."""

Purpose: Submit a bid for a task.

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

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