aishastr.live

Agentic Architecture Pattern: Resource Manager

Written by

admin

in


    """
Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens)
across agents with allocation, tracking, and optimization.
"""

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 ResourceType(Enum):
    """Types of resources to manage."""
    CPU = "cpu"
    MEMORY = "memory"
    API_CALLS = "api_calls"
    TOKENS = "tokens"
    STORAGE = "storage"


@dataclass
class Resource:
    """Resource definition."""
    type: ResourceType
    total_capacity: float
    available: float
    unit: str = ""
    
    def allocate(self, amount: float) -> bool:
        """Allocate resource."""
        if self.available >= amount:
            self.available -= amount
            return True
        return False
    
    def release(self, amount: float):
        """Release resource."""
        self.available = min(self.available + amount, self.total_capacity)
    
    def utilization(self) -> float:
        """Get utilization percentage."""
        return ((self.total_capacity - self.available) / self.total_capacity) * 100


@dataclass
class ResourceAllocation:
    """Track resource allocation."""
    agent_id: str
    resource_type: ResourceType
    amount: float
    timestamp: datetime = field(default_factory=datetime.now)
    released: bool = False


class ResourceManagerPattern:
    """
    Resource Manager Pattern - Centralized resource management.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.resources: Dict[ResourceType, Resource] = {}
        self.allocations: List[ResourceAllocation] = []
        self.agents: Dict[str, Dict[ResourceType, float]] = {}
    
    def add_resource(self, resource_type: ResourceType, capacity: float, unit: str = ""):
        """Add a resource pool."""
        self.resources[resource_type] = Resource(
            type=resource_type,
            total_capacity=capacity,
            available=capacity,
            unit=unit
        )
        print(f"Resource added: {resource_type.value} ({capacity} {unit})")
    
    def request_resources(self, agent_id: str, 
                         requests: Dict[ResourceType, float]) -> Dict[str, Any]:
        """Request resources for an agent."""
        # Check if all resources are available
        can_allocate = all(
            self.resources.get(r_type) and 
            self.resources[r_type].available >= amount
            for r_type, amount in requests.items()
        )
        
        if not can_allocate:
            return {
                "success": False,
                "message": "Insufficient resources",
                "requested": requests,
                "available": {
                    r_type: self.resources[r_type].available 
                    for r_type in requests.keys()
                    if r_type in self.resources
                }
            }
        
        # Allocate all resources
        allocated = {}
        for r_type, amount in requests.items():
            if self.resources[r_type].allocate(amount):
                allocation = ResourceAllocation(
                    agent_id=agent_id,
                    resource_type=r_type,
                    amount=amount
                )
                self.allocations.append(allocation)
                allocated[r_type] = amount
                
                # Track per agent
                if agent_id not in self.agents:
                    self.agents[agent_id] = {}
                self.agents[agent_id][r_type] = self.agents[agent_id].get(r_type, 0) + amount
        
        return {
            "success": True,
            "message": "Resources allocated",
            "agent_id": agent_id,
            "allocated": allocated
        }
    
    def release_resources(self, agent_id: str, 
                         releases: Optional[Dict[ResourceType, float]] = None):
        """Release resources from an agent."""
        if releases is None:
            # Release all resources for agent
            releases = self.agents.get(agent_id, {})
        
        for r_type, amount in releases.items():
            if r_type in self.resources:
                self.resources[r_type].release(amount)
                
                # Update agent tracking
                if agent_id in self.agents and r_type in self.agents[agent_id]:
                    self.agents[agent_id][r_type] -= amount
                    if self.agents[agent_id][r_type]  Dict[str, Any]:
        """Get status of all resources."""
        return {
            r_type.value: {
                "total": res.total_capacity,
                "available": res.available,
                "used": res.total_capacity - res.available,
                "utilization": f"{res.utilization():.1f}%",
                "unit": res.unit
            }
            for r_type, res in self.resources.items()
        }
    
    def get_agent_allocations(self, agent_id: str) -> Dict[str, Any]:
        """Get allocations for specific agent."""
        return {
            "agent_id": agent_id,
            "allocations": self.agents.get(agent_id, {}),
            "total_allocations": sum(self.agents.get(agent_id, {}).values())
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute resource management scenario."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a resource manager. Analyze the resource usage scenario."),
            ("human", "{scenario}\n\nCurrent status: {status}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "scenario": scenario,
            "status": str(self.get_resource_status())
        })
        
        return {
            "scenario": scenario,
            "analysis": response.content,
            "resource_status": self.get_resource_status()
        }


# Example usage
if __name__ == "__main__":
    # Create resource manager
    manager = ResourceManagerPattern()
    
    # Add resources
    manager.add_resource(ResourceType.CPU, 100.0, "cores")
    manager.add_resource(ResourceType.MEMORY, 16384, "MB")
    manager.add_resource(ResourceType.API_CALLS, 1000, "calls/hour")
    manager.add_resource(ResourceType.TOKENS, 1000000, "tokens")
    
    print("\n=== Resource Manager Pattern Demo ===\n")
    
    # Initial status
    print("Initial Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")
    
    # Agent 1 requests resources
    print("\n\n1. Agent 'DataProcessor' requesting resources:")
    result1 = manager.request_resources("DataProcessor", {
        ResourceType.CPU: 20.0,
        ResourceType.MEMORY: 4096,
        ResourceType.TOKENS: 100000
    })
    print(f"   Success: {result1['success']}")
    print(f"   Allocated: {result1['allocated']}")
    
    # Agent 2 requests resources
    print("\n2. Agent 'ModelTrainer' requesting resources:")
    result2 = manager.request_resources("ModelTrainer", {
        ResourceType.CPU: 50.0,
        ResourceType.MEMORY: 8192,
        ResourceType.API_CALLS: 500
    })
    print(f"   Success: {result2['success']}")
    print(f"   Allocated: {result2['allocated']}")
    
    # Show updated status
    print("\n\nUpdated Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")
    
    # Show agent allocations
    print("\n\nAgent Allocations:")
    for agent_id in ["DataProcessor", "ModelTrainer"]:
        allocs = manager.get_agent_allocations(agent_id)
        print(f"  {agent_id}: {allocs['allocations']}")
    
    # Release resources
    print("\n\n3. Releasing resources from 'DataProcessor':")
    release_result = manager.release_resources("DataProcessor")
    print(f"   Success: {release_result['success']}")
    print(f"   Released: {release_result['released']}")
    
    # Final status
    print("\n\nFinal Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")

Agentic Architecture Pattern: Resource Manager

Pattern 98: 98 Resource Manager – Line-by-Line Explanation

Overview

Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens) across agents with allocation, tracking, and optimization.

Code Structure Analysis

File Information

  • Pattern Number: 98
  • Pattern Name: 98 Resource Manager
  • File Path: patterns\tool_resource\98_resource_manager.py
  • Total Lines: 257

Detailed Code Explanation

Module Docstring

"""
Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens)
across agents with allocation, tracking, and optimization.
"""

Explanation: This module docstring describes the purpose and key features of the 98 Resource Manager 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

ResourceType (Line 16)

class ResourceType(Enum):
    CPU
    MEMORY
    API_CALLS
    TOKENS
    STORAGE

Purpose: Types of resources to manage.

Values: – CPU: Represents cpu – MEMORY: Represents memory – API_CALLS: Represents api calls – TOKENS: Represents tokens – STORAGE: Represents storage

Classes

ResourceType (Line 16)

Description: Types of resources to manage.

Resource (Line 26)

Description: Resource definition.

Methods:

allocate (Line 33)
def allocate(self, amount):
    """Allocate resource."""

Purpose: Allocate resource.

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

release (Line 40)
def release(self, amount):
    """Release resource."""

Purpose: Release resource.

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

utilization (Line 44)
def utilization(self):
    """Get utilization percentage."""

Purpose: Get utilization percentage.

Parameters: – self: Instance reference

ResourceAllocation (Line 50)

Description: Track resource allocation.

ResourceManagerPattern (Line 59)

Description: Resource Manager Pattern – Centralized resource management.

Methods:

__init__ (Line 64)
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”)

add_resource (Line 70)
def add_resource(self, resource_type, capacity, unit):
    """Add a resource pool."""

Purpose: Add a resource pool.

Parameters: – self: Instance reference – resource_type: Parameter for resource_type – capacity: Parameter for capacity – unit: Parameter for unit

request_resources (Line 80)
def request_resources(self, agent_id, requests):
    """Request resources for an agent."""

Purpose: Request resources for an agent.

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

release_resources (Line 126)
def release_resources(self, agent_id, releases):
    """Release resources from an agent."""

Purpose: Release resources from an agent.

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

get_resource_status (Line 157)
def get_resource_status(self):
    """Get status of all resources."""

Purpose: Get status of all resources.

Parameters: – self: Instance reference

get_agent_allocations (Line 170)
def get_agent_allocations(self, agent_id):
    """Get allocations for specific agent."""

Purpose: Get allocations for specific agent.

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

run (Line 178)
def run(self, scenario):
    """Execute resource management scenario."""

Purpose: Execute resource management 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 = ResourceType(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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts