aishastr.live

Agentic Architecture Pattern: Proxy Agent

Written by

admin

in


    """
Pattern 89: Proxy Agent Pattern

Acts as an intermediary controlling access to another agent or service.
Features access control, caching, logging, and lazy initialization.
"""

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


class AccessLevel(Enum):
    """Access control levels."""
    PUBLIC = "public"
    PRIVATE = "private"
    RESTRICTED = "restricted"
    ADMIN = "admin"


@dataclass
class CacheEntry:
    """Cached response entry."""
    key: str
    value: Any
    timestamp: datetime = field(default_factory=datetime.now)
    hits: int = 0
    ttl_seconds: int = 300  # 5 minutes default
    
    def is_expired(self) -> bool:
        """Check if cache entry has expired."""
        return datetime.now() > self.timestamp + timedelta(seconds=self.ttl_seconds)


@dataclass
class AccessRequest:
    """Access request record."""
    user: str
    operation: str
    access_level: AccessLevel
    allowed: bool
    timestamp: datetime = field(default_factory=datetime.now)


class RealAgent:
    """The real agent being proxied."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.initialized = False
        self.operation_count = 0
    
    def initialize(self):
        """Expensive initialization (lazy loaded)."""
        if not self.initialized:
            print(f"   {self.name}: Performing expensive initialization...")
            self.initialized = True
    
    def execute(self, operation: str, params: Dict[str, Any]) -> str:
        """Execute real operation."""
        if not self.initialized:
            self.initialize()
        
        self.operation_count += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}. Execute the requested operation."),
            ("human", "Operation: {operation}\nParameters: {params}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"operation": operation, "params": str(params)})
        
        return response.content


class ProxyAgent:
    """Proxy agent that controls access to real agent."""
    
    def __init__(self, name: str, real_agent: RealAgent):
        self.name = name
        self.real_agent = real_agent
        self.cache: Dict[str, CacheEntry] = {}
        self.access_log: List[AccessRequest] = []
        self.permissions: Dict[str, List[str]] = {
            "public_ops": ["read", "list", "search"],
            "restricted_ops": ["write", "update"],
            "admin_ops": ["delete", "admin"]
        }
    
    def check_access(self, user: str, operation: str, user_level: AccessLevel) -> bool:
        """Check if user has access to operation."""
        # Public operations
        if operation in self.permissions["public_ops"]:
            allowed = True
        # Restricted operations
        elif operation in self.permissions["restricted_ops"]:
            allowed = user_level in [AccessLevel.RESTRICTED, AccessLevel.ADMIN]
        # Admin operations
        elif operation in self.permissions["admin_ops"]:
            allowed = user_level == AccessLevel.ADMIN
        else:
            allowed = True  # Unknown operations allowed by default
        
        # Log access request
        self.access_log.append(AccessRequest(
            user=user,
            operation=operation,
            access_level=user_level,
            allowed=allowed
        ))
        
        return allowed
    
    def get_from_cache(self, key: str) -> Optional[Any]:
        """Get value from cache if not expired."""
        if key in self.cache:
            entry = self.cache[key]
            if not entry.is_expired():
                entry.hits += 1
                print(f"   {self.name}: Cache hit! (hits: {entry.hits})")
                return entry.value
            else:
                del self.cache[key]
                print(f"   {self.name}: Cache expired")
        return None
    
    def put_in_cache(self, key: str, value: Any, ttl_seconds: int = 300):
        """Put value in cache."""
        self.cache[key] = CacheEntry(key=key, value=value, ttl_seconds=ttl_seconds)
        print(f"   {self.name}: Cached result (TTL: {ttl_seconds}s)")
    
    def execute(self, user: str, user_level: AccessLevel, operation: str, 
                params: Dict[str, Any], use_cache: bool = True) -> Optional[str]:
        """Execute operation through proxy."""
        print(f"\n   {self.name}: Proxy request from {user}")
        print(f"   Operation: {operation}, Access Level: {user_level.value}")
        
        # Check access
        if not self.check_access(user, operation, user_level):
            print(f"   {self.name}: Access DENIED")
            return None
        
        print(f"   {self.name}: Access granted")
        
        # Check cache
        if use_cache and operation == "read":
            cache_key = f"{operation}:{str(params)}"
            cached_result = self.get_from_cache(cache_key)
            if cached_result is not None:
                return cached_result
        
        # Forward to real agent
        print(f"   {self.name}: Forwarding to real agent...")
        result = self.real_agent.execute(operation, params)
        
        # Cache result for read operations
        if use_cache and operation == "read":
            cache_key = f"{operation}:{str(params)}"
            self.put_in_cache(cache_key, result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get proxy statistics."""
        total_requests = len(self.access_log)
        denied_requests = sum(1 for req in self.access_log if not req.allowed)
        cache_entries = len(self.cache)
        total_cache_hits = sum(entry.hits for entry in self.cache.values())
        
        return {
            "total_requests": total_requests,
            "denied_requests": denied_requests,
            "cache_entries": cache_entries,
            "total_cache_hits": total_cache_hits,
            "real_agent_calls": self.real_agent.operation_count
        }


class ProxyAgentPattern:
    """
    Proxy Agent Pattern - Controlled access intermediary.
    
    Proxy agents control access to real agents providing caching,
    access control, logging, and lazy initialization.
    """
    
    def __init__(self):
        self.proxies: Dict[str, ProxyAgent] = {}
    
    def register_proxy(self, proxy: ProxyAgent):
        """Register proxy agent."""
        self.proxies[proxy.name] = proxy
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute proxy scenarios."""
        print("=== Proxy Agent Pattern Demo ===\n")
        
        for scenario in scenarios:
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            proxy_name = scenario.get("proxy")
            if proxy_name not in self.proxies:
                continue
            
            proxy = self.proxies[proxy_name]
            user = scenario.get("user", "anonymous")
            user_level = AccessLevel(scenario.get("access_level", "public"))
            operation = scenario.get("operation")
            params = scenario.get("params", {})
            use_cache = scenario.get("use_cache", True)
            
            result = proxy.execute(user, user_level, operation, params, use_cache)
            
            if result:
                print(f"   Result: {result[:80]}...")
        
        return {
            "pattern": "ProxyAgent",
            "total_proxies": len(self.proxies),
            "scenarios": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Proxy Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = ProxyAgentPattern()
    
    # Create real agent
    print("1. Creating real agent (not initialized yet)...")
    real_agent = RealAgent("DatabaseAgent")
    
    # Create proxy
    print("\n2. Creating proxy agent...")
    proxy = ProxyAgent("DatabaseProxy", real_agent)
    pattern.register_proxy(proxy)
    
    # Test scenarios
    print("\n3. Executing operations through proxy...")
    
    scenarios = [
        {
            "description": "Public read (should be cached)",
            "proxy": "DatabaseProxy",
            "user": "alice",
            "access_level": "public",
            "operation": "read",
            "params": {"id": "user_123"}
        },
        {
            "description": "Same read (cache hit expected)",
            "proxy": "DatabaseProxy",
            "user": "bob",
            "access_level": "public",
            "operation": "read",
            "params": {"id": "user_123"}
        },
        {
            "description": "Write operation (restricted)",
            "proxy": "DatabaseProxy",
            "user": "alice",
            "access_level": "public",
            "operation": "write",
            "params": {"id": "user_456", "data": {"name": "Alice"}}
        },
        {
            "description": "Write with restricted access",
            "proxy": "DatabaseProxy",
            "user": "admin",
            "access_level": "restricted",
            "operation": "write",
            "params": {"id": "user_456", "data": {"name": "Alice"}}
        },
        {
            "description": "Delete operation (admin only)",
            "proxy": "DatabaseProxy",
            "user": "admin",
            "access_level": "admin",
            "operation": "delete",
            "params": {"id": "user_789"}
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n4. Summary:")
    print(f"   Total Proxies: {summary['total_proxies']}")
    print(f"   Scenarios Executed: {summary['scenarios']}")
    
    print("\n5. Proxy Statistics:")
    stats = proxy.get_statistics()
    print(f"   Total Requests: {stats['total_requests']}")
    print(f"   Denied Requests: {stats['denied_requests']}")
    print(f"   Cache Entries: {stats['cache_entries']}")
    print(f"   Cache Hits: {stats['total_cache_hits']}")
    print(f"   Real Agent Calls: {stats['real_agent_calls']}")
    
    print("\n6. Access Log:")
    for req in proxy.access_log[-5:]:  # Last 5 requests
        status = "✓" if req.allowed else "✗"
        print(f"   {status} {req.user} ({req.access_level.value}): {req.operation}")
    
    print("\n✓ Proxy agent successfully controlled access and caching!")

Agentic Architecture Pattern: Proxy Agent

Pattern 89: 89 Proxy Agent – Line-by-Line Explanation

Overview

Pattern 89: Proxy Agent Pattern

Acts as an intermediary controlling access to another agent or service. Features access control, caching, logging, and lazy initialization.

Code Structure Analysis

File Information

  • Pattern Number: 89
  • Pattern Name: 89 Proxy Agent
  • File Path: patterns\specialized_agent\89_proxy_agent.py
  • Total Lines: 311

Detailed Code Explanation

Module Docstring

"""
Pattern 89: Proxy Agent Pattern

Acts as an intermediary controlling access to another agent or service.
Features access control, caching, logging, and lazy initialization.
"""

Explanation: This module docstring describes the purpose and key features of the 89 Proxy Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
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, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: 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

AccessLevel (Line 16)

class AccessLevel(Enum):
    PUBLIC
    PRIVATE
    RESTRICTED
    ADMIN

Purpose: Access control levels.

Values: – PUBLIC: Represents public – PRIVATE: Represents private – RESTRICTED: Represents restricted – ADMIN: Represents admin

Classes

AccessLevel (Line 16)

Description: Access control levels.

CacheEntry (Line 25)

Description: Cached response entry.

Methods:

is_expired (Line 33)
def is_expired(self):
    """Check if cache entry has expired."""

Purpose: Check if cache entry has expired.

Parameters: – self: Instance reference

AccessRequest (Line 39)

Description: Access request record.

RealAgent (Line 48)

Description: The real agent being proxied.

Methods:

__init__ (Line 51)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

initialize (Line 57)
def initialize(self):
    """Expensive initialization (lazy loaded)."""

Purpose: Expensive initialization (lazy loaded).

Parameters: – self: Instance reference

execute (Line 63)
def execute(self, operation, params):
    """Execute real operation."""

Purpose: Execute real operation.

Parameters: – self: Instance reference – operation: Parameter for operation – params: Parameter for params

ProxyAgent (Line 81)

Description: Proxy agent that controls access to real agent.

Methods:

__init__ (Line 84)
def __init__(self, name, real_agent):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – real_agent: Parameter for real_agent

check_access (Line 95)
def check_access(self, user, operation, user_level):
    """Check if user has access to operation."""

Purpose: Check if user has access to operation.

Parameters: – self: Instance reference – user: Parameter for user – operation: Parameter for operation – user_level: Parameter for user_level

get_from_cache (Line 119)
def get_from_cache(self, key):
    """Get value from cache if not expired."""

Purpose: Get value from cache if not expired.

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

put_in_cache (Line 132)
def put_in_cache(self, key, value, ttl_seconds):
    """Put value in cache."""

Purpose: Put value in cache.

Parameters: – self: Instance reference – key: Parameter for key – value: Parameter for value – ttl_seconds: Parameter for ttl_seconds

execute (Line 137)
def execute(self, user, user_level, operation, params, use_cache):
    """Execute operation through proxy."""

Purpose: Execute operation through proxy.

Parameters: – self: Instance reference – user: Parameter for user – user_level: Parameter for user_level – operation: Parameter for operation – params: Parameter for params – use_cache: Parameter for use_cache

get_statistics (Line 168)
def get_statistics(self):
    """Get proxy statistics."""

Purpose: Get proxy statistics.

Parameters: – self: Instance reference

ProxyAgentPattern (Line 184)

Description: Proxy Agent Pattern – Controlled access intermediary.

Proxy agents control access to real agents providing caching, access control, logging, and lazy initialization.

Methods:

__init__ (Line 192)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_proxy (Line 195)
def register_proxy(self, proxy):
    """Register proxy agent."""

Purpose: Register proxy agent.

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

run (Line 199)
def run(self, scenarios):
    """Execute proxy scenarios."""

Purpose: Execute proxy scenarios.

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

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