"""
Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead.
Similar to connection pooling in databases.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


class PooledAgentStatus(Enum):
    """Status of pooled agent."""
    AVAILABLE = "available"
    IN_USE = "in_use"
    RETIRED = "retired"


@dataclass
class PooledAgent:
    """Agent in the pool."""
    id: str
    role: str
    status: PooledAgentStatus = PooledAgentStatus.AVAILABLE
    created_at: datetime = field(default_factory=datetime.now)
    last_used: datetime = field(default_factory=datetime.now)
    tasks_completed: int = 0


class AgentPoolPattern:
    """
    Agent Pool Pattern - Reusable agent instances.
    
    Features:
    - Pre-warmed agent pool
    - Checkout/checkin mechanism
    - Pool size management (min/max)
    - Idle agent cleanup
    - Usage statistics
    """
    
    def __init__(self, min_size: int = 2, max_size: int = 10, model: str = "gpt-4"):
        self.min_size = min_size
        self.max_size = max_size
        self.model = model
        self.pool: Dict[str, PooledAgent] = {}
        self.agent_llms: Dict[str, ChatOpenAI] = {}
        self.checkouts = 0
        self.checkins = 0
        
        # Initialize pool with minimum size
        for i in range(min_size):
            self._create_agent(f"pooled-agent-{i}")
    
    def _create_agent(self, agent_id: str):
        """Internal: Create a new pooled agent."""
        agent = PooledAgent(id=agent_id, role="general")
        self.pool[agent_id] = agent
        self.agent_llms[agent_id] = ChatOpenAI(model=self.model, temperature=0.7)
    
    def checkout(self) -> Optional[str]:
        """Checkout an available agent from pool."""
        # Find available agent
        for agent_id, agent in self.pool.items():
            if agent.status == PooledAgentStatus.AVAILABLE:
                agent.status = PooledAgentStatus.IN_USE
                agent.last_used = datetime.now()
                self.checkouts += 1
                return agent_id
        
        # Try to create new agent if under max size
        if len(self.pool)  bool:
        """Return agent to pool."""
        agent = self.pool.get(agent_id)
        if agent and agent.status == PooledAgentStatus.IN_USE:
            agent.status = PooledAgentStatus.AVAILABLE
            self.checkins += 1
            return True
        return False
    
    def execute_task(self, agent_id: str, task: str) -> str:
        """Execute task with checked-out agent."""
        agent = self.pool.get(agent_id)
        llm = self.agent_llms.get(agent_id)
        
        if not agent or not llm or agent.status != PooledAgentStatus.IN_USE:
            return "Agent not available"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {agent.id} from the agent pool"),
            ("human", "{task}")
        ])
        
        chain = prompt | llm
        response = chain.invoke({"task": task})
        agent.tasks_completed += 1
        
        return str(response.content)
    
    def cleanup_idle(self, max_idle_seconds: float = 600.0):
        """Remove agents that have been idle too long."""
        current_time = datetime.now()
        to_remove = []
        
        for agent_id, agent in self.pool.items():
            if agent.status == PooledAgentStatus.AVAILABLE:
                idle_time = (current_time - agent.last_used).total_seconds()
                if idle_time > max_idle_seconds and len(self.pool) > self.min_size:
                    to_remove.append(agent_id)
        
        for agent_id in to_remove:
            del self.pool[agent_id]
            del self.agent_llms[agent_id]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pool statistics."""
        available = sum(1 for a in self.pool.values() if a.status == PooledAgentStatus.AVAILABLE)
        in_use = sum(1 for a in self.pool.values() if a.status == PooledAgentStatus.IN_USE)
        
        return {
            "pool_size": len(self.pool),
            "available": available,
            "in_use": in_use,
            "total_checkouts": self.checkouts,
            "total_checkins": self.checkins,
            "utilization": in_use / len(self.pool) if self.pool else 0
        }


if __name__ == "__main__":
    print("=== Agent Pool Pattern Demo ===\n")
    
    pool = AgentPoolPattern(min_size=2, max_size=5)
    
    print(f"Pool initialized with {len(pool.pool)} agents\n")
    
    # Checkout agents and use them
    print("Checking out agents...")
    agent1 = pool.checkout()
    agent2 = pool.checkout()
    
    if agent1:
        print(f"Checked out: {agent1}")
        result = pool.execute_task(agent1, "Process data")
        print(f"Result: {result[:50]}...")
        pool.checkin(agent1)
        print(f"Checked in: {agent1}")
    
    if agent2:
        print(f"Checked out: {agent2}")
        result = pool.execute_task(agent2, "Generate report")
        print(f"Result: {result[:50]}...")
        pool.checkin(agent2)
        print(f"Checked in: {agent2}")
    
    # Statistics
    print("\n=== Statistics ===")
    stats = pool.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")

    

Agentic Architecture Pattern: Agent Pool

Pattern 165: 165 Agent Pool – Line-by-Line Explanation

Overview

Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead. Similar to connection pooling in databases.

Code Structure Analysis

File Information

• Pattern Number: 165
• Pattern Name: 165 Agent Pool
• File Path: patterns\scaling\165_agent_pool.py
• Total Lines: 172

Detailed Code Explanation

Module Docstring

"""
Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead.
Similar to connection pooling in databases.
"""

Explanation: This module docstring describes the purpose and key features of the 165 Agent Pool 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_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate  # 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_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Enumerations

PooledAgentStatus (Line 16)

class PooledAgentStatus(Enum):
    AVAILABLE
    IN_USE
    RETIRED

Purpose: Status of pooled agent.

Values: – AVAILABLE: Represents available – IN_USE: Represents in use – RETIRED: Represents retired

Classes

PooledAgentStatus (Line 16)

Description: Status of pooled agent.

PooledAgent (Line 24)

Description: Agent in the pool.

AgentPoolPattern (Line 34)

Description: Agent Pool Pattern – Reusable agent instances.

Features: – Pre-warmed agent pool – Checkout/checkin mechanism – Pool size management (min/max) – Idle agent cleanup – Usage statistics

Methods:

__init__ (Line 46)

def __init__(self, min_size, max_size, model):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – min_size: Parameter for min_size – max_size: Parameter for max_size – model: Parameter for model

_create_agent (Line 59)

def _create_agent(self, agent_id):
    """Internal: Create a new pooled agent."""

Purpose: Internal: Create a new pooled agent.

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

checkout (Line 65)

def checkout(self):
    """Checkout an available agent from pool."""

Purpose: Checkout an available agent from pool.

Parameters: – self: Instance reference

checkin (Line 83)

def checkin(self, agent_id):
    """Return agent to pool."""

Purpose: Return agent to pool.

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

execute_task (Line 92)

def execute_task(self, agent_id, task):
    """Execute task with checked-out agent."""

Purpose: Execute task with checked-out agent.

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

cleanup_idle (Line 111)

def cleanup_idle(self, max_idle_seconds):
    """Remove agents that have been idle too long."""

Purpose: Remove agents that have been idle too long.

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

get_statistics (Line 126)

def get_statistics(self):
    """Get pool statistics."""

Purpose: Get pool statistics.

Parameters: – self: Instance reference

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