"""
Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements.
Features task routing, load balancing, and result aggregation.
"""

from typing import Dict, Any, List, Optional, Tuple
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 TaskPriority(Enum):
    """Task priority levels."""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class Task:
    """Represents a task to be delegated."""
    task_id: str
    description: str
    required_skills: List[str]
    priority: TaskPriority = TaskPriority.MEDIUM
    params: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class WorkerAgent:
    """Worker agent that can execute tasks."""
    name: str
    skills: List[str]
    capacity: int = 5
    current_load: int = 0
    tasks_completed: int = 0
    
    def can_accept(self) -> bool:
        """Check if worker can accept more tasks."""
        return self.current_load  bool:
        """Check if worker has required skills."""
        return all(skill in self.skills for skill in required_skills)
    
    def match_score(self, required_skills: List[str]) -> float:
        """Calculate match score (0.0 to 1.0)."""
        if not required_skills:
            return 1.0
        matching = sum(1 for skill in required_skills if skill in self.skills)
        return matching / len(required_skills)


@dataclass
class DelegationRecord:
    """Records a delegation event."""
    task: Task
    worker: str
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    result: Optional[str] = None
    success: bool = False


class DelegateAgent:
    """Agent that delegates tasks to workers."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.workers: Dict[str, WorkerAgent] = {}
        self.delegation_history: List[DelegationRecord] = []
        self.pending_tasks: List[Task] = []
    
    def register_worker(self, worker: WorkerAgent):
        """Register a worker agent."""
        self.workers[worker.name] = worker
        print(f"   {self.name}: Registered worker '{worker.name}' with skills: {', '.join(worker.skills)}")
    
    def find_best_worker(self, task: Task) -> Optional[WorkerAgent]:
        """Find the best worker for a task."""
        candidates = []
        
        for worker in self.workers.values():
            if worker.can_accept() and worker.has_skills(task.required_skills):
                score = worker.match_score(task.required_skills)
                # Consider load in scoring
                load_factor = 1.0 - (worker.current_load / worker.capacity)
                final_score = score * 0.7 + load_factor * 0.3
                candidates.append((worker, final_score))
        
        if not candidates:
            return None
        
        # Sort by score (higher is better)
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]
    
    def delegate_task(self, task: Task) -> Optional[str]:
        """Delegate task to best available worker."""
        print(f"\n   {self.name}: Delegating task '{task.task_id}'")
        print(f"   Description: {task.description}")
        print(f"   Required skills: {', '.join(task.required_skills)}")
        print(f"   Priority: {task.priority.name}")
        
        # Find best worker
        worker = self.find_best_worker(task)
        
        if not worker:
            print(f"   {self.name}: No suitable worker found")
            self.pending_tasks.append(task)
            return None
        
        print(f"   {self.name}: Assigned to '{worker.name}' (match: {worker.match_score(task.required_skills):.2f})")
        
        # Update worker load
        worker.current_load += 1
        
        # Execute task via worker
        result = self._execute_task_via_worker(task, worker)
        
        # Update worker
        worker.current_load -= 1
        worker.tasks_completed += 1
        
        # Record delegation
        record = DelegationRecord(
            task=task,
            worker=worker.name,
            end_time=datetime.now(),
            result=result,
            success=True
        )
        self.delegation_history.append(record)
        
        print(f"   {self.name}: Task completed by '{worker.name}'")
        return result
    
    def _execute_task_via_worker(self, task: Task, worker: WorkerAgent) -> str:
        """Execute task through worker agent."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {worker.name}, a worker agent with skills: {', '.join(worker.skills)}. "
                      f"Execute the delegated task."),
            ("human", "Task: {description}\nParameters: {params}\n\nExecute this task.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "description": task.description,
            "params": str(task.params)
        })
        
        return response.content
    
    def process_pending_tasks(self):
        """Try to process pending tasks."""
        if not self.pending_tasks:
            return
        
        print(f"\n   {self.name}: Processing {len(self.pending_tasks)} pending tasks...")
        
        processed = []
        for task in self.pending_tasks:
            result = self.delegate_task(task)
            if result:
                processed.append(task)
        
        # Remove processed tasks
        for task in processed:
            self.pending_tasks.remove(task)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get delegation statistics."""
        total_delegations = len(self.delegation_history)
        successful = sum(1 for d in self.delegation_history if d.success)
        
        worker_stats = {
            name: {
                "tasks_completed": worker.tasks_completed,
                "current_load": worker.current_load,
                "utilization": f"{worker.current_load / worker.capacity * 100:.1f}%"
            }
            for name, worker in self.workers.items()
        }
        
        return {
            "total_delegations": total_delegations,
            "successful_delegations": successful,
            "pending_tasks": len(self.pending_tasks),
            "worker_stats": worker_stats
        }


class DelegateAgentPattern:
    """
    Delegate Agent Pattern - Task delegation and routing.
    
    Delegate agents route tasks to specialized workers based on
    skills, availability, and load balancing.
    """
    
    def __init__(self):
        self.delegates: Dict[str, DelegateAgent] = {}
    
    def register_delegate(self, delegate: DelegateAgent):
        """Register delegate agent."""
        self.delegates[delegate.name] = delegate
    
    def run(self, tasks: List[Task]) -> Dict[str, Any]:
        """Execute delegation scenarios."""
        print("=== Delegate Agent Pattern Demo ===\n")
        
        for delegate in self.delegates.values():
            for task in tasks:
                delegate.delegate_task(task)
            
            # Try pending tasks
            delegate.process_pending_tasks()
        
        total_delegations = sum(
            len(d.delegation_history)
            for d in self.delegates.values()
        )
        
        return {
            "pattern": "DelegateAgent",
            "total_delegates": len(self.delegates),
            "total_delegations": total_delegations,
            "tasks_processed": len(tasks)
        }


# Example usage
if __name__ == "__main__":
    print("=== Delegate Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = DelegateAgentPattern()
    
    # Create delegate agent
    print("1. Creating delegate agent...")
    delegate = DelegateAgent("TaskCoordinator")
    
    # Register workers
    print("\n2. Registering worker agents...")
    
    delegate.register_worker(WorkerAgent(
        name="DataWorker",
        skills=["data_processing", "analysis", "sql"],
        capacity=3
    ))
    
    delegate.register_worker(WorkerAgent(
        name="WebWorker",
        skills=["web_scraping", "api_calls", "http"],
        capacity=5
    ))
    
    delegate.register_worker(WorkerAgent(
        name="MLWorker",
        skills=["machine_learning", "training", "prediction"],
        capacity=2
    ))
    
    delegate.register_worker(WorkerAgent(
        name="GeneralWorker",
        skills=["data_processing", "web_scraping", "text_processing"],
        capacity=4
    ))
    
    pattern.register_delegate(delegate)
    
    # Create tasks
    print("\n3. Creating tasks to delegate...")
    
    tasks = [
        Task(
            task_id="T001",
            description="Analyze sales data and generate report",
            required_skills=["data_processing", "analysis"],
            priority=TaskPriority.HIGH
        ),
        Task(
            task_id="T002",
            description="Scrape product information from website",
            required_skills=["web_scraping"],
            priority=TaskPriority.MEDIUM
        ),
        Task(
            task_id="T003",
            description="Train predictive model on customer data",
            required_skills=["machine_learning", "training"],
            priority=TaskPriority.HIGH
        ),
        Task(
            task_id="T004",
            description="Extract data from SQL database",
            required_skills=["sql", "data_processing"],
            priority=TaskPriority.LOW
        ),
        Task(
            task_id="T005",
            description="Process text documents",
            required_skills=["text_processing"],
            priority=TaskPriority.MEDIUM
        )
    ]
    
    # Execute
    print("\n4. Delegating tasks...")
    summary = pattern.run(tasks)
    
    print(f"\n5. Summary:")
    print(f"   Total Delegations: {summary['total_delegations']}")
    print(f"   Tasks Processed: {summary['tasks_processed']}")
    
    print("\n6. Delegation Statistics:")
    stats = delegate.get_statistics()
    print(f"   Successful Delegations: {stats['successful_delegations']}")
    print(f"   Pending Tasks: {stats['pending_tasks']}")
    
    print("\n7. Worker Performance:")
    for worker_name, worker_stat in stats['worker_stats'].items():
        print(f"   {worker_name}:")
        print(f"     Tasks Completed: {worker_stat['tasks_completed']}")
        print(f"     Current Load: {worker_stat['current_load']}")
        print(f"     Utilization: {worker_stat['utilization']}")
    
    print("\n✓ Delegate agent successfully routed tasks to workers!")

    

Agentic Architecture Pattern: Delegate Agent

Pattern 90: 90 Delegate Agent – Line-by-Line Explanation

Overview

Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements. Features task routing, load balancing, and result aggregation.

Code Structure Analysis

File Information

• Pattern Number: 90
• Pattern Name: 90 Delegate Agent
• File Path: patterns\specialized_agent\90_delegate_agent.py
• Total Lines: 336

Detailed Code Explanation

Module Docstring

"""
Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements.
Features task routing, load balancing, and result aggregation.
"""

Explanation: This module docstring describes the purpose and key features of the 90 Delegate Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
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, Tuple: 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

TaskPriority (Line 16)

class TaskPriority(Enum):
    LOW
    MEDIUM
    HIGH
    CRITICAL

Purpose: Task priority levels.

Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical

Classes

TaskPriority (Line 16)

Description: Task priority levels.

Task (Line 25)

Description: Represents a task to be delegated.

WorkerAgent (Line 36)

Description: Worker agent that can execute tasks.

Methods:

can_accept (Line 44)

def can_accept(self):
    """Check if worker can accept more tasks."""

Purpose: Check if worker can accept more tasks.

Parameters: – self: Instance reference

has_skills (Line 48)

def has_skills(self, required_skills):
    """Check if worker has required skills."""

Purpose: Check if worker has required skills.

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

match_score (Line 52)

def match_score(self, required_skills):
    """Calculate match score (0.0 to 1.0)."""

Purpose: Calculate match score (0.0 to 1.0).

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

DelegationRecord (Line 61)

Description: Records a delegation event.

DelegateAgent (Line 71)

Description: Agent that delegates tasks to workers.

Methods:

__init__ (Line 74)

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

register_worker (Line 81)

def register_worker(self, worker):
    """Register a worker agent."""

Purpose: Register a worker agent.

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

find_best_worker (Line 86)

def find_best_worker(self, task):
    """Find the best worker for a task."""

Purpose: Find the best worker for a task.

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

delegate_task (Line 105)

def delegate_task(self, task):
    """Delegate task to best available worker."""

Purpose: Delegate task to best available worker.

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

_execute_task_via_worker (Line 145)

def _execute_task_via_worker(self, task, worker):
    """Execute task through worker agent."""

Purpose: Execute task through worker agent.

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

process_pending_tasks (Line 161)

def process_pending_tasks(self):
    """Try to process pending tasks."""

Purpose: Try to process pending tasks.

Parameters: – self: Instance reference

get_statistics (Line 178)

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

Purpose: Get delegation statistics.

Parameters: – self: Instance reference

DelegateAgentPattern (Line 200)

Description: Delegate Agent Pattern – Task delegation and routing.

Delegate agents route tasks to specialized workers based on skills, availability, and load balancing.

Methods:

__init__ (Line 208)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_delegate (Line 211)

def register_delegate(self, delegate):
    """Register delegate agent."""

Purpose: Register delegate agent.

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

run (Line 215)

def run(self, tasks):
    """Execute delegation scenarios."""

Purpose: Execute delegation scenarios.

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

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

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

    """
Pattern 88: Wrapper Agent Pattern

Wraps legacy systems providing modern interfaces while maintaining backward compatibility.
Features interface adaptation, protocol translation, and behavior enhancement.
"""

from typing import Dict, Any, List, Optional, Callable
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 InterfaceType(Enum):
    """Types of interfaces."""
    LEGACY = "legacy"
    MODERN = "modern"
    REST_API = "rest_api"
    CLI = "cli"
    GUI = "gui"


@dataclass
class LegacySystem:
    """Represents a legacy system."""
    name: str
    interface_type: InterfaceType
    operations: List[str]
    limitations: List[str] = field(default_factory=list)
    
    def execute_legacy_operation(self, operation: str, params: Dict[str, Any]) -> str:
        """Simulate legacy system operation."""
        return f"Legacy {operation} executed with {params}"


@dataclass
class WrapperCall:
    """Records a wrapper invocation."""
    operation: str
    modern_params: Dict[str, Any]
    legacy_params: Dict[str, Any]
    result: Any
    timestamp: datetime = field(default_factory=datetime.now)


class WrapperAgent:
    """Agent that wraps legacy systems with modern interfaces."""
    
    def __init__(self, name: str, legacy_system: LegacySystem, model_name: str = "gpt-4"):
        self.name = name
        self.legacy_system = legacy_system
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.call_history: List[WrapperCall] = []
        self.modern_operations: Dict[str, Callable] = {}
        self.parameter_mappings: Dict[str, Dict[str, str]] = {}
    
    def register_modern_operation(self, modern_name: str, legacy_name: str, 
                                   param_mapping: Optional[Dict[str, str]] = None):
        """Register a modern operation that wraps a legacy one."""
        self.parameter_mappings[modern_name] = param_mapping or {}
        
        def wrapper(**modern_params):
            return self.invoke_legacy(modern_name, legacy_name, modern_params)
        
        self.modern_operations[modern_name] = wrapper
        print(f"   {self.name}: Registered '{modern_name}' -> '{legacy_name}'")
    
    def invoke_legacy(self, modern_op: str, legacy_op: str, 
                      modern_params: Dict[str, Any]) -> Any:
        """Invoke legacy operation with parameter translation."""
        print(f"\n   {self.name}: Wrapping call to '{modern_op}'")
        
        # Translate parameters
        legacy_params = self._translate_parameters(modern_op, modern_params)
        print(f"   Translated parameters: {legacy_params}")
        
        # Call legacy system
        result = self.legacy_system.execute_legacy_operation(legacy_op, legacy_params)
        
        # Enhance result with LLM
        enhanced_result = self._enhance_result(result, modern_op)
        
        # Record call
        self.call_history.append(WrapperCall(
            operation=modern_op,
            modern_params=modern_params,
            legacy_params=legacy_params,
            result=enhanced_result
        ))
        
        print(f"   Result: {enhanced_result[:80]}...")
        return enhanced_result
    
    def _translate_parameters(self, operation: str, modern_params: Dict[str, Any]) -> Dict[str, Any]:
        """Translate modern parameters to legacy format."""
        mapping = self.parameter_mappings.get(operation, {})
        legacy_params = {}
        
        for modern_key, value in modern_params.items():
            legacy_key = mapping.get(modern_key, modern_key)
            legacy_params[legacy_key] = value
        
        return legacy_params
    
    def _enhance_result(self, legacy_result: str, operation: str) -> str:
        """Enhance legacy result with modern formatting."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are enhancing legacy system output for modern applications. "
                      "Format the result in a more user-friendly way."),
            ("human", "Operation: {operation}\nLegacy Result: {result}\n\n"
                     "Provide an enhanced, user-friendly version.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "operation": operation,
            "result": legacy_result
        })
        
        return response.content
    
    def call(self, operation: str, **params) -> Any:
        """Call a modern operation."""
        if operation not in self.modern_operations:
            raise ValueError(f"Operation '{operation}' not found")
        
        return self.modern_operations[operation](**params)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get wrapper statistics."""
        operation_counts = {}
        for call in self.call_history:
            operation_counts[call.operation] = operation_counts.get(call.operation, 0) + 1
        
        return {
            "total_calls": len(self.call_history),
            "operation_counts": operation_counts,
            "wrapped_system": self.legacy_system.name,
            "modern_operations": list(self.modern_operations.keys())
        }


class WrapperAgentPattern:
    """
    Wrapper Agent Pattern - Legacy system integration.
    
    Wrapper agents provide modern interfaces to legacy systems,
    enabling integration without modifying the legacy code.
    """
    
    def __init__(self):
        self.agents: Dict[str, WrapperAgent] = {}
    
    def register_agent(self, agent: WrapperAgent):
        """Register wrapper agent."""
        self.agents[agent.name] = agent
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute wrapper scenarios."""
        print("=== Wrapper Agent Pattern Demo ===\n")
        
        for scenario in scenarios:
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            agent_name = scenario.get("agent")
            if agent_name not in self.agents:
                continue
            
            agent = self.agents[agent_name]
            operation = scenario.get("operation")
            params = scenario.get("params", {})
            
            try:
                result = agent.call(operation, **params)
            except Exception as e:
                print(f"   Error: {e}")
        
        # Statistics
        total_calls = sum(
            len(agent.call_history)
            for agent in self.agents.values()
        )
        
        return {
            "pattern": "WrapperAgent",
            "total_agents": len(self.agents),
            "total_calls": total_calls,
            "scenarios": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Wrapper Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = WrapperAgentPattern()
    
    # Create legacy system
    print("1. Defining legacy system...")
    legacy = LegacySystem(
        name="MainframeDB",
        interface_type=InterfaceType.CLI,
        operations=["FETCH_REC", "UPD_REC", "DEL_REC", "INS_REC"],
        limitations=["Fixed-width fields", "No JSON support", "Cryptic error codes"]
    )
    print(f"   Legacy system: {legacy.name}")
    print(f"   Operations: {', '.join(legacy.operations)}")
    
    # Create wrapper agent
    print("\n2. Creating wrapper agent...")
    wrapper = WrapperAgent("ModernDBWrapper", legacy)
    
    # Register modern operations
    print("\n3. Registering modern operations...")
    wrapper.register_modern_operation(
        modern_name="get_user",
        legacy_name="FETCH_REC",
        param_mapping={"user_id": "REC_ID", "fields": "FLD_LIST"}
    )
    
    wrapper.register_modern_operation(
        modern_name="update_user",
        legacy_name="UPD_REC",
        param_mapping={"user_id": "REC_ID", "data": "FLD_DATA"}
    )
    
    wrapper.register_modern_operation(
        modern_name="delete_user",
        legacy_name="DEL_REC",
        param_mapping={"user_id": "REC_ID"}
    )
    
    pattern.register_agent(wrapper)
    
    # Test scenarios
    print("\n4. Executing modern operations...")
    
    scenarios = [
        {
            "description": "Get user by ID",
            "agent": "ModernDBWrapper",
            "operation": "get_user",
            "params": {"user_id": "12345", "fields": ["name", "email", "age"]}
        },
        {
            "description": "Update user data",
            "agent": "ModernDBWrapper",
            "operation": "update_user",
            "params": {"user_id": "12345", "data": {"email": "new@example.com"}}
        },
        {
            "description": "Delete user",
            "agent": "ModernDBWrapper",
            "operation": "delete_user",
            "params": {"user_id": "67890"}
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n5. Summary:")
    print(f"   Total Calls: {summary['total_calls']}")
    print(f"   Scenarios Executed: {summary['scenarios']}")
    
    print("\n6. Wrapper Statistics:")
    stats = wrapper.get_statistics()
    print(f"   Wrapped System: {stats['wrapped_system']}")
    print(f"   Modern Operations: {', '.join(stats['modern_operations'])}")
    print(f"   Operation Counts:")
    for op, count in stats['operation_counts'].items():
        print(f"     {op}: {count}")
    
    print("\n✓ Wrapper agent successfully modernized legacy system!")

    

Agentic Architecture Pattern: Wrapper Agent

Pattern 88: 88 Wrapper Agent – Line-by-Line Explanation

Overview

Pattern 88: Wrapper Agent Pattern

Wraps legacy systems providing modern interfaces while maintaining backward compatibility. Features interface adaptation, protocol translation, and behavior enhancement.

Code Structure Analysis

File Information

• Pattern Number: 88
• Pattern Name: 88 Wrapper Agent
• File Path: patterns\specialized_agent\88_wrapper_agent.py
• Total Lines: 277

Detailed Code Explanation

Module Docstring

"""
Pattern 88: Wrapper Agent Pattern

Wraps legacy systems providing modern interfaces while maintaining backward compatibility.
Features interface adaptation, protocol translation, and behavior enhancement.
"""

Explanation: This module docstring describes the purpose and key features of the 88 Wrapper Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
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, Callable: 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

InterfaceType (Line 16)

class InterfaceType(Enum):
    LEGACY
    MODERN
    REST_API
    CLI
    GUI

Purpose: Types of interfaces.

Values: – LEGACY: Represents legacy – MODERN: Represents modern – REST_API: Represents rest api – CLI: Represents cli – GUI: Represents gui

Classes

InterfaceType (Line 16)

Description: Types of interfaces.

LegacySystem (Line 26)

Description: Represents a legacy system.

Methods:

execute_legacy_operation (Line 33)

def execute_legacy_operation(self, operation, params):
    """Simulate legacy system operation."""

Purpose: Simulate legacy system operation.

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

WrapperCall (Line 39)

Description: Records a wrapper invocation.

WrapperAgent (Line 48)

Description: Agent that wraps legacy systems with modern interfaces.

Methods:

__init__ (Line 51)

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

Purpose: No method docstring.

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

register_modern_operation (Line 59)

def register_modern_operation(self, modern_name, legacy_name, param_mapping):
    """Register a modern operation that wraps a legacy one."""

Purpose: Register a modern operation that wraps a legacy one.

Parameters: – self: Instance reference – modern_name: Parameter for modern_name – legacy_name: Parameter for legacy_name – param_mapping: Parameter for param_mapping

invoke_legacy (Line 70)

def invoke_legacy(self, modern_op, legacy_op, modern_params):
    """Invoke legacy operation with parameter translation."""

Purpose: Invoke legacy operation with parameter translation.

Parameters: – self: Instance reference – modern_op: Parameter for modern_op – legacy_op: Parameter for legacy_op – modern_params: Parameter for modern_params

_translate_parameters (Line 96)

def _translate_parameters(self, operation, modern_params):
    """Translate modern parameters to legacy format."""

Purpose: Translate modern parameters to legacy format.

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

_enhance_result (Line 107)

def _enhance_result(self, legacy_result, operation):
    """Enhance legacy result with modern formatting."""

Purpose: Enhance legacy result with modern formatting.

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

call (Line 124)

def call(self, operation):
    """Call a modern operation."""

Purpose: Call a modern operation.

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

get_statistics (Line 131)

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

Purpose: Get wrapper statistics.

Parameters: – self: Instance reference

WrapperAgentPattern (Line 145)

Description: Wrapper Agent Pattern – Legacy system integration.

Wrapper agents provide modern interfaces to legacy systems, enabling integration without modifying the legacy code.

Methods:

__init__ (Line 153)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 156)

def register_agent(self, agent):
    """Register wrapper agent."""

Purpose: Register wrapper agent.

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

run (Line 160)

def run(self, scenarios):
    """Execute wrapper scenarios."""

Purpose: Execute wrapper 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 = InterfaceType(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

    """
Pattern 87: Gateway Agent Pattern

Acts as a protocol translator between different systems or communication protocols.
Features protocol translation, message routing, and format conversion.
"""

from typing import Dict, Any, List, Optional, Callable
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 Protocol(Enum):
    """Communication protocols."""
    HTTP = "http"
    GRPC = "grpc"
    MQTT = "mqtt"
    WEBSOCKET = "websocket"
    REST = "rest"
    SOAP = "soap"
    CUSTOM = "custom"


@dataclass
class Message:
    """Message with protocol information."""
    content: Any
    protocol: Protocol
    headers: Dict[str, str] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Translation:
    """Translation record."""
    original_message: Message
    translated_message: Message
    from_protocol: Protocol
    to_protocol: Protocol
    timestamp: datetime = field(default_factory=datetime.now)


class ProtocolTranslator:
    """Translates between protocols."""
    
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.translations: List[Translation] = []
    
    def translate(self, message: Message, target_protocol: Protocol) -> Message:
        """Translate message to target protocol."""
        if message.protocol == target_protocol:
            return message
        
        # Use LLM for intelligent translation
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a protocol translator. Convert messages between protocols "
                      "while preserving semantic meaning."),
            ("human", """Translate this message:
From Protocol: {from_protocol}
To Protocol: {to_protocol}
Content: {content}
Headers: {headers}

Provide the translated content and appropriate headers for the target protocol.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "from_protocol": message.protocol.value,
            "to_protocol": target_protocol.value,
            "content": str(message.content),
            "headers": str(message.headers)
        })
        
        translated = Message(
            content=response.content,
            protocol=target_protocol,
            headers={"translated": "true"},
            metadata={"original_protocol": message.protocol.value}
        )
        
        # Record translation
        self.translations.append(Translation(
            original_message=message,
            translated_message=translated,
            from_protocol=message.protocol,
            to_protocol=target_protocol
        ))
        
        return translated


class GatewayAgent:
    """Gateway agent for protocol translation and routing."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.translator = ProtocolTranslator(self.llm)
        self.supported_protocols: List[Protocol] = []
        self.routes: Dict[Protocol, Callable[[Message], None]] = {}
        self.messages_processed = 0
    
    def add_protocol_support(self, protocol: Protocol):
        """Add support for a protocol."""
        if protocol not in self.supported_protocols:
            self.supported_protocols.append(protocol)
            print(f"   {self.name}: Added support for {protocol.value}")
    
    def register_route(self, protocol: Protocol, handler: Callable[[Message], None]):
        """Register a message handler for a protocol."""
        self.routes[protocol] = handler
    
    def process_message(self, message: Message, target_protocol: Optional[Protocol] = None) -> Message:
        """Process and translate message."""
        self.messages_processed += 1
        
        print(f"\n   {self.name}: Processing message")
        print(f"   Source protocol: {message.protocol.value}")
        
        # Translate if needed
        if target_protocol and target_protocol != message.protocol:
            if target_protocol not in self.supported_protocols:
                raise ValueError(f"Protocol {target_protocol.value} not supported")
            
            print(f"   Translating to: {target_protocol.value}")
            translated = self.translator.translate(message, target_protocol)
            
            # Route to handler
            if target_protocol in self.routes:
                self.routes[target_protocol](translated)
            
            return translated
        
        return message
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get gateway statistics."""
        translation_counts = {}
        for trans in self.translator.translations:
            key = f"{trans.from_protocol.value} -> {trans.to_protocol.value}"
            translation_counts[key] = translation_counts.get(key, 0) + 1
        
        return {
            "messages_processed": self.messages_processed,
            "total_translations": len(self.translator.translations),
            "supported_protocols": [p.value for p in self.supported_protocols],
            "translation_routes": translation_counts
        }


class GatewayAgentPattern:
    """
    Gateway Agent Pattern - Protocol translation gateway.
    
    Gateway agents translate between different communication protocols,
    enabling interoperability between heterogeneous systems.
    """
    
    def __init__(self):
        self.agents: Dict[str, GatewayAgent] = {}
    
    def register_agent(self, agent: GatewayAgent):
        """Register gateway agent."""
        self.agents[agent.name] = agent
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute gateway scenarios."""
        print("=== Gateway Agent Pattern Demo ===\n")
        
        for scenario in scenarios:
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            agent_name = scenario.get("agent")
            if agent_name not in self.agents:
                continue
            
            agent = self.agents[agent_name]
            message = Message(
                content=scenario.get("content"),
                protocol=Protocol(scenario.get("source_protocol")),
                headers=scenario.get("headers", {})
            )
            
            target = Protocol(scenario.get("target_protocol")) if scenario.get("target_protocol") else None
            result = agent.process_message(message, target)
            
            print(f"   Result protocol: {result.protocol.value}")
        
        # Statistics
        total_translations = sum(
            len(agent.translator.translations)
            for agent in self.agents.values()
        )
        
        return {
            "pattern": "GatewayAgent",
            "total_agents": len(self.agents),
            "total_translations": total_translations,
            "scenarios": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Gateway Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = GatewayAgentPattern()
    
    # Create gateway agent
    print("1. Creating gateway agent...")
    gateway = GatewayAgent("APIGateway")
    
    # Add protocol support
    print("\n2. Adding protocol support...")
    gateway.add_protocol_support(Protocol.HTTP)
    gateway.add_protocol_support(Protocol.GRPC)
    gateway.add_protocol_support(Protocol.MQTT)
    gateway.add_protocol_support(Protocol.WEBSOCKET)
    
    # Register handlers
    def http_handler(msg: Message):
        print(f"     HTTP handler received: {str(msg.content)[:50]}...")
    
    def grpc_handler(msg: Message):
        print(f"     gRPC handler received: {str(msg.content)[:50]}...")
    
    gateway.register_route(Protocol.HTTP, http_handler)
    gateway.register_route(Protocol.GRPC, grpc_handler)
    
    pattern.register_agent(gateway)
    
    # Test scenarios
    print("\n3. Processing messages...")
    
    scenarios = [
        {
            "description": "Translate REST to gRPC",
            "agent": "APIGateway",
            "content": {"method": "GET", "path": "/api/users", "params": {"id": 123}},
            "source_protocol": "rest",
            "target_protocol": "grpc",
            "headers": {"Content-Type": "application/json"}
        },
        {
            "description": "Translate MQTT to WebSocket",
            "agent": "APIGateway",
            "content": {"topic": "sensors/temperature", "value": 23.5},
            "source_protocol": "mqtt",
            "target_protocol": "websocket",
            "headers": {"qos": "1"}
        },
        {
            "description": "Translate HTTP to MQTT",
            "agent": "APIGateway",
            "content": {"event": "user.login", "data": {"user_id": 456}},
            "source_protocol": "http",
            "target_protocol": "mqtt",
            "headers": {"Authorization": "Bearer token123"}
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n4. Summary:")
    print(f"   Total Translations: {summary['total_translations']}")
    print(f"   Scenarios Processed: {summary['scenarios']}")
    
    print("\n5. Gateway Statistics:")
    stats = gateway.get_statistics()
    print(f"   Messages Processed: {stats['messages_processed']}")
    print(f"   Supported Protocols: {', '.join(stats['supported_protocols'])}")
    print(f"   Translation Routes:")
    for route, count in stats['translation_routes'].items():
        print(f"     {route}: {count}")
    
    print("\n✓ Gateway agent successfully translated between protocols!")

    

Agentic Architecture Pattern: Gateway Agent

Pattern 87: 87 Gateway Agent – Line-by-Line Explanation

Overview

Pattern 87: Gateway Agent Pattern

Acts as a protocol translator between different systems or communication protocols. Features protocol translation, message routing, and format conversion.

Code Structure Analysis

File Information

• Pattern Number: 87
• Pattern Name: 87 Gateway Agent
• File Path: patterns\specialized_agent\87_gateway_agent.py
• Total Lines: 284

Detailed Code Explanation

Module Docstring

"""
Pattern 87: Gateway Agent Pattern

Acts as a protocol translator between different systems or communication protocols.
Features protocol translation, message routing, and format conversion.
"""

Explanation: This module docstring describes the purpose and key features of the 87 Gateway Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
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, Callable: 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

Protocol (Line 16)

class Protocol(Enum):
    HTTP
    GRPC
    MQTT
    WEBSOCKET
    REST
    SOAP
    CUSTOM

Purpose: Communication protocols.

Values: – HTTP: Represents http – GRPC: Represents grpc – MQTT: Represents mqtt – WEBSOCKET: Represents websocket – REST: Represents rest – SOAP: Represents soap – CUSTOM: Represents custom

Classes

Protocol (Line 16)

Description: Communication protocols.

Message (Line 28)

Description: Message with protocol information.

Translation (Line 38)

Description: Translation record.

ProtocolTranslator (Line 47)

Description: Translates between protocols.

Methods:

__init__ (Line 50)

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

Purpose: No method docstring.

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

translate (Line 54)

def translate(self, message, target_protocol):
    """Translate message to target protocol."""

Purpose: Translate message to target protocol.

Parameters: – self: Instance reference – message: Parameter for message – target_protocol: Parameter for target_protocol

GatewayAgent (Line 98)

Description: Gateway agent for protocol translation and routing.

Methods:

__init__ (Line 101)

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

add_protocol_support (Line 109)

def add_protocol_support(self, protocol):
    """Add support for a protocol."""

Purpose: Add support for a protocol.

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

register_route (Line 115)

def register_route(self, protocol, handler):
    """Register a message handler for a protocol."""

Purpose: Register a message handler for a protocol.

Parameters: – self: Instance reference – protocol: Parameter for protocol – handler: Parameter for handler

process_message (Line 119)

def process_message(self, message, target_protocol):
    """Process and translate message."""

Purpose: Process and translate message.

Parameters: – self: Instance reference – message: Parameter for message – target_protocol: Parameter for target_protocol

get_statistics (Line 142)

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

Purpose: Get gateway statistics.

Parameters: – self: Instance reference

GatewayAgentPattern (Line 157)

Description: Gateway Agent Pattern – Protocol translation gateway.

Gateway agents translate between different communication protocols, enabling interoperability between heterogeneous systems.

Methods:

__init__ (Line 165)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 168)

def register_agent(self, agent):
    """Register gateway agent."""

Purpose: Register gateway agent.

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

run (Line 172)

def run(self, scenarios):
    """Execute gateway scenarios."""

Purpose: Execute gateway 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 = Protocol(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

    """
Pattern 86: Stationary Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class StationaryAgentConfig:
    """Configuration for Stationary Agent."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class StationaryAgentState:
    """State information for Stationary Agent."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class StationaryAgentAgent:
    """Agent implementing Stationary Agent."""
    
    def __init__(self, name: str, config: StationaryAgentConfig = None):
        self.name = name
        self.config = config or StationaryAgentConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = StationaryAgentState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Stationary Agent."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Stationary Agent. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class StationaryAgentPattern:
    """
    Stationary Agent - Enhanced implementation
    
    This pattern provides comprehensive functionality for stationary agent
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: StationaryAgentConfig = None):
        self.config = config or StationaryAgentConfig()
        self.agents: Dict[str, StationaryAgentAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Stationary Agent. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: StationaryAgentAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Stationary Agent",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Stationary Agent",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Stationary Agent Demo ===\n")
    
    # Create configuration
    config = StationaryAgentConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = StationaryAgentPattern(config)
    
    # Create and register agents
    agent1 = StationaryAgentAgent("Agent1", config)
    agent2 = StationaryAgentAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate stationary agent functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Stationary Agent

Pattern 86: 86 Stationary Agent – Line-by-Line Explanation

Overview

Pattern 86: Stationary Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 86
• Pattern Name: 86 Stationary Agent
• File Path: patterns\specialized_agent\86_stationary_agent.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 86: Stationary Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 86 Stationary Agent 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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

StationaryAgentConfig (Line 18)

Description: Configuration for Stationary Agent.

StationaryAgentState (Line 27)

Description: State information for Stationary Agent.

StationaryAgentAgent (Line 36)

Description: Agent implementing Stationary Agent.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Stationary Agent."""

Purpose: Process input according to Stationary Agent.

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

get_statistics (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

StationaryAgentPattern (Line 85)

Description: Stationary Agent – Enhanced implementation

This pattern provides comprehensive functionality for stationary agent with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

    """
Pattern 85: Mobile Agent Pattern with LangChain/LangGraph

Purpose: Agent migration between execution environments with intelligent routing.
Enables agents to move across environments while preserving state and optimizing placement.
"""

from typing import Dict, Any, List, Optional, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json
import time


class AgentStatus(Enum):
    """Agent execution status."""
    IDLE = "idle"
    EXECUTING = "executing"
    MIGRATING = "migrating"
    SUSPENDED = "suspended"
    TERMINATED = "terminated"


class MigrationType(Enum):
    """Types of migration."""
    MANUAL = "manual"
    LOAD_BALANCING = "load_balancing"
    RESOURCE_OPTIMIZATION = "resource_optimization"
    FAULT_TOLERANCE = "fault_tolerance"
    STRATEGIC = "strategic"


class EnvironmentCapability(Enum):
    """Environment capabilities."""
    COMPUTE = "compute"
    STORAGE = "storage"
    NETWORK = "network"
    GPU = "gpu"
    MEMORY = "memory"


@dataclass
class ExecutionEnvironment:
    """Execution environment definition."""
    env_id: str
    name: str
    capabilities: List[EnvironmentCapability]
    current_load: float = 0.0
    max_load: float = 100.0
    latency_ms: float = 10.0
    available: bool = True
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class MigrationRecord:
    """Record of agent migration."""
    migration_id: str
    agent_id: str
    from_env: str
    to_env: str
    migration_type: MigrationType
    reason: str
    timestamp: datetime
    duration_ms: float
    success: bool
    state_size_bytes: int


@dataclass
class AgentTask:
    """Task executed by agent."""
    task_id: str
    description: str
    required_capabilities: List[EnvironmentCapability]
    execution_time: float
    result: Optional[str] = None


class MigrationState(TypedDict):
    """State for migration workflow."""
    agent_id: str
    current_env: str
    target_env: Optional[str]
    migration_reason: str
    agent_state: Dict[str, Any]
    task_history: List[Dict[str, Any]]
    environments: Dict[str, Dict[str, Any]]
    migration_score: float
    recommendations: List[str]
    approved: bool


class MobileAgent:
    """
    Mobile Agent with state preservation and migration capabilities.
    """
    
    def __init__(self, agent_id: str, name: str, initial_env: str):
        self.agent_id = agent_id
        self.name = name
        self.current_env = initial_env
        self.status = AgentStatus.IDLE
        self.state_data: Dict[str, Any] = {}
        self.task_history: List[AgentTask] = []
        self.migration_history: List[MigrationRecord] = []
        self.created_at = datetime.now()
    
    def serialize_state(self) -> Dict[str, Any]:
        """Serialize agent state for migration."""
        return {
            "agent_id": self.agent_id,
            "name": self.name,
            "current_env": self.current_env,
            "status": self.status.value,
            "state_data": self.state_data,
            "task_history": [
                {
                    "task_id": t.task_id,
                    "description": t.description,
                    "result": t.result
                } for t in self.task_history
            ],
            "created_at": self.created_at.isoformat()
        }
    
    def deserialize_state(self, state: Dict[str, Any], new_env: str):
        """Restore agent state after migration."""
        self.current_env = new_env
        self.state_data = state.get("state_data", {})
        # Task history restored from serialized data
    
    def execute_task(self, task: AgentTask) -> str:
        """Execute task in current environment."""
        self.status = AgentStatus.EXECUTING
        
        # Simulate task execution
        time.sleep(0.01)  # Simulate work
        
        result = f"Task {task.task_id} completed in {self.current_env}"
        task.result = result
        self.task_history.append(task)
        
        self.status = AgentStatus.IDLE
        return result


class IntelligentMigrationManager:
    """
    Intelligent Migration Manager with LangGraph workflow.
    
    Features:
    - Environment analysis
    - Load balancing
    - Intelligent routing
    - State preservation
    - Migration optimization
    """
    
    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.3):
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.agents: Dict[str, MobileAgent] = {}
        self.environments: Dict[str, ExecutionEnvironment] = {}
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_migrations": 0,
            "migrations_by_type": {t.value: 0 for t in MigrationType},
            "successful_migrations": 0,
            "failed_migrations": 0,
            "total_agents": 0,
            "avg_migration_time_ms": 0.0
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for intelligent migration."""
        workflow = StateGraph(MigrationState)
        
        workflow.add_node("analyze_environments", self._analyze_environments_node)
        workflow.add_node("evaluate_migration", self._evaluate_migration_node)
        workflow.add_node("select_target", self._select_target_node)
        workflow.add_node("validate_migration", self._validate_migration_node)
        workflow.add_node("generate_recommendations", self._generate_recommendations_node)
        
        workflow.set_entry_point("analyze_environments")
        workflow.add_edge("analyze_environments", "evaluate_migration")
        workflow.add_edge("evaluate_migration", "select_target")
        workflow.add_edge("select_target", "validate_migration")
        workflow.add_edge("validate_migration", "generate_recommendations")
        workflow.add_edge("generate_recommendations", END)
        
        return workflow.compile()
    
    def _analyze_environments_node(self, state: MigrationState) -> MigrationState:
        """Analyze available environments."""
        env_analysis = {}
        
        for env_id, env in self.environments.items():
            env_analysis[env_id] = {
                "available": env.available,
                "load": env.current_load,
                "capacity": env.max_load - env.current_load,
                "latency_ms": env.latency_ms,
                "capabilities": [c.value for c in env.capabilities]
            }
        
        state["environments"] = env_analysis
        return state
    
    def _evaluate_migration_node(self, state: MigrationState) -> MigrationState:
        """Evaluate need for migration using LLM."""
        current_env = state["current_env"]
        reason = state["migration_reason"]
        environments = state["environments"]
        
        # Use LLM to evaluate migration necessity
        prompt = ChatPromptTemplate.from_template(
            "Evaluate if agent migration is beneficial:\n"
            "Current Environment: {current_env}\n"
            "Reason: {reason}\n"
            "Available Environments: {envs}\n\n"
            "Provide migration score (0.0-1.0) and brief analysis."
        )
        
        chain = prompt | self.llm
        response = chain.invoke({
            "current_env": current_env,
            "reason": reason,
            "envs": str(environments)
        })
        
        # Extract score (simplified - in production, parse LLM response)
        analysis = str(response.content)
        
        # Calculate score based on environment conditions
        if current_env in environments:
            current_load = environments[current_env]["load"]
            capacity = environments[current_env]["capacity"]
            
            if current_load > 80:
                score = 0.9  # High load, strong migration candidate
            elif capacity  MigrationState:
        """Select optimal target environment."""
        current_env = state["current_env"]
        environments = state["environments"]
        migration_score = state["migration_score"]
        
        if migration_score  best_score:
                best_score = total_score
                best_env = env_id
        
        state["target_env"] = best_env
        return state
    
    def _validate_migration_node(self, state: MigrationState) -> MigrationState:
        """Validate migration is safe and beneficial."""
        target_env = state["target_env"]
        
        if target_env is None:
            state["approved"] = False
        else:
            # Check if target environment can accept agent
            target_data = state["environments"].get(target_env, {})
            
            if target_data.get("available") and target_data.get("capacity", 0) > 10:
                state["approved"] = True
            else:
                state["approved"] = False
                state["target_env"] = None
        
        return state
    
    def _generate_recommendations_node(self, state: MigrationState) -> MigrationState:
        """Generate migration recommendations."""
        recommendations = []
        
        if state["approved"]:
            recommendations.append(f"Migrate to {state['target_env']} - optimal environment identified")
            recommendations.append("State preservation verified")
            recommendations.append("Target environment has sufficient capacity")
        else:
            recommendations.append("Migration not recommended at this time")
            
            if state["migration_score"]  Optional[MigrationRecord]:
        """Migrate agent with intelligent routing."""
        
        if agent_id not in self.agents:
            return None
        
        agent = self.agents[agent_id]
        start_time = time.time()
        
        # Prepare workflow state
        initial_state: MigrationState = {
            "agent_id": agent_id,
            "current_env": agent.current_env,
            "target_env": target_env,
            "migration_reason": reason,
            "agent_state": agent.serialize_state(),
            "task_history": [
                {"task_id": t.task_id, "result": t.result}
                for t in agent.task_history
            ],
            "environments": {},
            "migration_score": 0.0,
            "recommendations": [],
            "approved": False
        }
        
        # Run workflow
        result_state = self.workflow.invoke(initial_state)
        
        # Check if migration approved
        if not result_state["approved"]:
            return None
        
        final_target = result_state["target_env"]
        if not final_target or final_target not in self.environments:
            return None
        
        # Perform migration
        agent.status = AgentStatus.MIGRATING
        
        # Serialize state
        serialized_state = agent.serialize_state()
        state_size = len(json.dumps(serialized_state))
        
        # Update environment loads
        old_env = self.environments.get(agent.current_env)
        if old_env:
            old_env.current_load = max(0, old_env.current_load - 10)
        
        new_env = self.environments[final_target]
        new_env.current_load = min(new_env.max_load, new_env.current_load + 10)
        
        # Migrate agent
        agent.deserialize_state(serialized_state, final_target)
        agent.status = AgentStatus.IDLE
        
        duration = (time.time() - start_time) * 1000  # ms
        
        # Record migration
        record = MigrationRecord(
            migration_id=f"mig-{len(agent.migration_history) + 1}",
            agent_id=agent_id,
            from_env=result_state["current_env"],
            to_env=final_target,
            migration_type=migration_type,
            reason=reason,
            timestamp=datetime.now(),
            duration_ms=duration,
            success=True,
            state_size_bytes=state_size
        )
        
        agent.migration_history.append(record)
        
        # Update stats
        self.stats["total_migrations"] += 1
        self.stats["migrations_by_type"][migration_type.value] += 1
        self.stats["successful_migrations"] += 1
        
        total_time = self.stats["avg_migration_time_ms"] * (self.stats["total_migrations"] - 1) + duration
        self.stats["avg_migration_time_ms"] = total_time / self.stats["total_migrations"]
        
        return record
    
    def auto_balance(self):
        """Automatically balance agents across environments."""
        # Find overloaded environments
        overloaded = [
            env_id for env_id, env in self.environments.items()
            if env.current_load > 80
        ]
        
        # Migrate agents from overloaded environments
        for agent in self.agents.values():
            if agent.current_env in overloaded:
                self.migrate_agent(
                    agent.agent_id,
                    reason="load_balancing",
                    migration_type=MigrationType.LOAD_BALANCING
                )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get migration statistics."""
        return {
            **self.stats,
            "total_environments": len(self.environments),
            "active_agents": sum(1 for a in self.agents.values() if a.status != AgentStatus.TERMINATED)
        }


class MobileAgentPattern:
    """
    Comprehensive Mobile Agent Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.manager = IntelligentMigrationManager(model_name=model_name)
    
    def create_environment(
        self,
        env_id: str,
        name: str,
        capabilities: List[EnvironmentCapability],
        max_load: float = 100.0
    ) -> ExecutionEnvironment:
        """Create execution environment."""
        env = ExecutionEnvironment(
            env_id=env_id,
            name=name,
            capabilities=capabilities,
            max_load=max_load
        )
        self.manager.register_environment(env)
        return env
    
    def create_agent(
        self,
        agent_id: str,
        name: str,
        initial_env: str
    ) -> MobileAgent:
        """Create mobile agent."""
        agent = MobileAgent(agent_id, name, initial_env)
        self.manager.register_agent(agent)
        return agent
    
    def migrate(
        self,
        agent_id: str,
        target_env: Optional[str] = None,
        reason: str = "manual"
    ) -> Optional[MigrationRecord]:
        """Migrate agent."""
        return self.manager.migrate_agent(agent_id, target_env, reason)
    
    def auto_balance(self):
        """Auto-balance agents."""
        self.manager.auto_balance()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics."""
        return self.manager.get_statistics()


if __name__ == "__main__":
    print("=== Mobile Agent Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Environment Setup and Agent Creation")
    print("-" * 50)
    pattern = MobileAgentPattern(model_name="gpt-4")
    
    # Create environments
    env1 = pattern.create_environment(
        "env-001",
        "Production Server",
        [EnvironmentCapability.COMPUTE, EnvironmentCapability.STORAGE, EnvironmentCapability.NETWORK],
        max_load=100.0
    )
    print(f"✓ Created environment: {env1.name}")
    
    env2 = pattern.create_environment(
        "env-002",
        "Edge Node",
        [EnvironmentCapability.COMPUTE, EnvironmentCapability.NETWORK],
        max_load=80.0
    )
    print(f"✓ Created environment: {env2.name}")
    
    env3 = pattern.create_environment(
        "env-003",
        "GPU Cluster",
        [EnvironmentCapability.COMPUTE, EnvironmentCapability.GPU, EnvironmentCapability.MEMORY],
        max_load=120.0
    )
    print(f"✓ Created environment: {env3.name}")
    
    # Create agents
    agent1 = pattern.create_agent("agent-001", "DataProcessor", "env-001")
    print(f"\n✓ Created agent: {agent1.name} in {agent1.current_env}")
    
    agent2 = pattern.create_agent("agent-002", "APIWorker", "env-001")
    print(f"✓ Created agent: {agent2.name} in {agent2.current_env}")
    
    agent3 = pattern.create_agent("agent-003", "MLAgent", "env-002")
    print(f"✓ Created agent: {agent3.name} in {agent3.current_env}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Task Execution")
    print("-" * 50)
    
    # Execute tasks
    task1 = AgentTask(
        task_id="task-001",
        description="Process data batch",
        required_capabilities=[EnvironmentCapability.COMPUTE],
        execution_time=0.1
    )
    
    result = agent1.execute_task(task1)
    print(f"Agent {agent1.name}: {result}")
    
    task2 = AgentTask(
        task_id="task-002",
        description="Handle API request",
        required_capabilities=[EnvironmentCapability.NETWORK],
        execution_time=0.05
    )
    
    result = agent2.execute_task(task2)
    print(f"Agent {agent2.name}: {result}")
    
    print("\n" + "=" * 50)
    print("\nScenario 3: Intelligent Migration")
    print("-" * 50)
    
    # Simulate high load on env-001
    pattern.manager.environments["env-001"].current_load = 85.0
    print(f"Simulated high load on {env1.name}: {env1.current_load}%")
    
    # Migrate agent1 with intelligent routing
    print(f"\nMigrating {agent1.name} with intelligent routing...")
    record = pattern.migrate(agent1.agent_id, reason="high_load_detected")
    
    if record:
        print(f"✓ Migration successful!")
        print(f"  From: {record.from_env}")
        print(f"  To: {record.to_env}")
        print(f"  Duration: {record.duration_ms:.2f}ms")
        print(f"  State Size: {record.state_size_bytes} bytes")
    else:
        print("✗ Migration not recommended")
    
    print("\n" + "=" * 50)
    print("\nScenario 4: Manual Migration")
    print("-" * 50)
    
    # Manual migration to GPU cluster
    print(f"Migrating {agent3.name} to GPU cluster...")
    record = pattern.migrate(agent3.agent_id, target_env="env-003", reason="gpu_required")
    
    if record:
        print(f"✓ Migration successful!")
        print(f"  From: {record.from_env}")
        print(f"  To: {record.to_env}")
        print(f"  Reason: {record.reason}")
    else:
        print("✗ Migration failed or not approved")
    
    print("\n" + "=" * 50)
    print("\nScenario 5: Auto Load Balancing")
    print("-" * 50)
    
    print("Running auto load balancing...")
    pattern.auto_balance()
    print("✓ Load balancing complete")
    
    print("\n" + "=" * 50)
    print("\nScenario 6: Migration History")
    print("-" * 50)
    
    for agent in pattern.manager.agents.values():
        if agent.migration_history:
            print(f"\n{agent.name} migration history:")
            for i, mig in enumerate(agent.migration_history, 1):
                print(f"  {i}. {mig.from_env} → {mig.to_env} ({mig.migration_type.value})")
    
    print("\n" + "=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    stats = pattern.get_statistics()
    print(f"Total Agents: {stats['total_agents']}")
    print(f"Active Agents: {stats['active_agents']}")
    print(f"Total Migrations: {stats['total_migrations']}")
    print(f"Successful Migrations: {stats['successful_migrations']}")
    print(f"Average Migration Time: {stats['avg_migration_time_ms']:.2f}ms")
    print(f"Total Environments: {stats['total_environments']}")
    
    print(f"\nMigrations by Type:")
    for mig_type, count in stats['migrations_by_type'].items():
        if count > 0:
            print(f"  {mig_type}: {count}")
    
    print("\n" + "=" * 50)
    print("Mobile agent pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Mobile Agent

Pattern 85: Mobile Agent Pattern – Line-by-Line Explanation

Overview

This pattern implements mobile agents that can migrate between execution environments, carrying their state and logic. It includes intelligent routing, load balancing, state serialization, and LLM-powered migration decisions.

Code Explanation

Lines 1-17: Imports

"""
Pattern 85: Mobile Agent Pattern with LangChain/LangGraph

Purpose: Mobile agents with migration, state transfer, and intelligent routing.
Provides agent mobility with LLM-powered migration decisions and load balancing.
"""

Explanation: Module docstring – pattern for agents that move between environments.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import pickle
import base64
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Includes pickle for object serialization and base64 for encoding serialized state.

Lines 19-34: Enum Definitions

class AgentStatus(Enum):
    """Status of mobile agent."""
    IDLE = "idle"
    MIGRATING = "migrating"
    EXECUTING = "executing"
    WAITING = "waiting"
    TERMINATED = "terminated"

Explanation: Five agent lifecycle states.

class MigrationStrategy(Enum):
    """Strategy for agent migration."""
    LOAD_BALANCED = "load_balanced"  # Migrate to least loaded
    PROXIMITY = "proximity"           # Migrate to closest
    CAPABILITY = "capability"         # Migrate to most capable
    COST = "cost"                     # Migrate to cheapest
    INTELLIGENT = "intelligent"       # LLM decides

Explanation: Five migration strategies: – LOAD_BALANCED: Choose environment with lowest load (CPU, memory, queue size) – PROXIMITY: Choose geographically closest (minimize latency) – CAPABILITY: Choose environment with best resources (GPU, special hardware) – COST: Choose cheapest environment (cloud cost optimization) – INTELLIGENT: Use LLM to analyze requirements and choose best environment

Lines 36-92: Dataclasses

@dataclass
class Environment:
    """Execution environment for mobile agents."""
    id: str
    name: str
    capabilities: List[str]
    load: float = 0.0  # 0.0 to 1.0
    cost_per_hour: float = 0.0
    location: str = "unknown"
    active_agents: List[str] = field(default_factory=list)

Explanation: Represents an execution environment: – capabilities: What this environment can do (e.g., [“gpu”, “high_memory”, “python3.11”]) – load: Current utilization (0.0 = idle, 1.0 = maxed out) – cost_per_hour: For cloud environments (e.g., $0.05/hour) – location: Geographic location (e.g., “us-east-1”, “europe-west”) – active_agents: List of agent IDs currently running

@dataclass
class AgentTask:
    """Task for mobile agent."""
    id: str
    code: Callable
    data: Dict[str, Any]
    requirements: List[str] = field(default_factory=list)

Explanation: Task agent will execute: – code: Actual function to run – data: Input data – requirements: Required capabilities (e.g., [“gpu”, “tensorflow”])

@dataclass
class MigrationDecision:
    """Decision about agent migration."""
    should_migrate: bool
    target_environment: Optional[str]
    reasoning: str
    strategy: MigrationStrategy
    confidence: float

Explanation: Migration decision made by system or LLM: – should_migrate: Yes/no decision – target_environment: Destination environment ID – reasoning: Why this decision was made – strategy: Which strategy was used – confidence: 0.0-1.0 confidence in decision

@dataclass
class MigrationRecord:
    """Record of agent migration."""
    agent_id: str
    from_environment: str
    to_environment: str
    timestamp: datetime = field(default_factory=datetime.now)
    state_size_bytes: int = 0
    migration_time_ms: float = 0.0
    success: bool = True

Explanation: Historical record of migration event. Used for analysis and optimization.

Lines 94-110: TypedDict State

class MobileAgentState(TypedDict):
    """State for mobile agent workflow."""
    agent_id: str
    current_environment: str
    task: AgentTask
    agent_state: Dict[str, Any]
    serialized_state: Optional[str]
    migration_decision: Optional[MigrationDecision]
    target_environment: Optional[str]
    status: AgentStatus
    execution_result: Any
    insights: List[str]

Explanation: Workflow state tracking agent identity, location, state, and migration decisions.

Lines 112-160: MobileAgent Initialization

class MobileAgent:
    """
    Mobile Agent with LangGraph workflow integration.

    Features:
    - Agent migration between environments
    - State serialization and transfer
    - Intelligent routing decisions
    - Load balancing
    - Capability-based placement
    - Cost optimization
    - Proximity awareness
    - Migration history tracking
    """

Explanation: Main mobile agent class with intelligent migration.

def __init__(
    self,
    agent_id: str,
    initial_environment: str,
    model_name: str = "gpt-4",
    temperature: float = 0.5
):
    self.agent_id = agent_id
    self.current_environment = initial_environment
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Agent state
    self.state: Dict[str, Any] = {}
    self.status = AgentStatus.IDLE

    # Migration history
    self.migration_history: List[MigrationRecord] = []

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_migrations": 0,
        "successful_migrations": 0,
        "failed_migrations": 0,
        "total_execution_time_ms": 0.0
    }

Explanation: Initialize agent with ID, starting environment, LLM for decisions, state, and tracking.

Lines 162-180: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(MobileAgentState)

    workflow.add_node("analyze_requirements", self._analyze_requirements_node)
    workflow.add_node("decide_migration", self._decide_migration_node)
    workflow.add_node("serialize_state", self._serialize_state_node)
    workflow.add_node("migrate", self._migrate_node)
    workflow.add_node("execute_task", self._execute_task_node)

Explanation: Create 5-node workflow: 1. Analyze requirements: Check what task needs 2. Decide migration: Determine if/where to migrate 3. Serialize state: Package agent state for transfer 4. Migrate: Move agent to new environment 5. Execute task: Run the actual task

    workflow.set_entry_point("analyze_requirements")
    workflow.add_edge("analyze_requirements", "decide_migration")
    workflow.add_edge("decide_migration", "serialize_state")
    workflow.add_edge("serialize_state", "migrate")
    workflow.add_edge("migrate", "execute_task")
    workflow.add_edge("execute_task", END)

    return workflow.compile()

Explanation: Linear flow through migration and execution.

Lines 182-204: Analyze Requirements Node

def _analyze_requirements_node(self, state: MobileAgentState) -> MobileAgentState:
    """Analyze task requirements."""
    task = state["task"]

    state["insights"] = []
    state["insights"].append(f"Analyzing requirements for task {task.id}")

    if task.requirements:
        state["insights"].append(f"Task requires: {', '.join(task.requirements)}")
    else:
        state["insights"].append("Task has no specific requirements")

    return state

Explanation: Extract and document task requirements. This informs migration decision.

Lines 206-312: Decide Migration Node (LLM-Powered)

def _decide_migration_node(
    self,
    state: MobileAgentState,
    environments: Dict[str, Environment],
    strategy: MigrationStrategy = MigrationStrategy.INTELLIGENT
) -> MobileAgentState:
    """Decide whether and where to migrate."""
    current_env_id = state["current_environment"]
    task = state["task"]
    current_env = environments.get(current_env_id)

Explanation: Extract current environment and task information.

    if strategy == MigrationStrategy.LOAD_BALANCED:
        # Choose least loaded environment
        candidates = [(env_id, env) for env_id, env in environments.items()]
        candidates.sort(key=lambda x: x[1].load)

        if candidates and candidates[0][1].load < current_env.load:
            state["migration_decision"] = MigrationDecision(
                should_migrate=True,
                target_environment=candidates[0][0],
                reasoning=f"Target has lower load ({candidates[0][1].load:.2f} vs {current_env.load:.2f})",
                strategy=strategy,
                confidence=0.9
            )
        else:
            state["migration_decision"] = MigrationDecision(
                should_migrate=False,
                target_environment=None,
                reasoning="Current environment has best load",
                strategy=strategy,
                confidence=0.9
            )

Explanation: LOAD_BALANCED strategy – choose environment with lowest load. Migrate only if better option exists.

    elif strategy == MigrationStrategy.CAPABILITY:
        # Choose environment with required capabilities
        suitable = []
        for env_id, env in environments.items():
            if all(req in env.capabilities for req in task.requirements):
                suitable.append((env_id, env))

        if suitable:
            # Prefer environment with most capabilities
            suitable.sort(key=lambda x: len(x[1].capabilities), reverse=True)

            if suitable[0][0] != current_env_id:
                state["migration_decision"] = MigrationDecision(
                    should_migrate=True,
                    target_environment=suitable[0][0],
                    reasoning=f"Target has required capabilities: {suitable[0][1].capabilities}",
                    strategy=strategy,
                    confidence=0.95
                )
            else:
                state["migration_decision"] = MigrationDecision(
                    should_migrate=False,
                    target_environment=None,
                    reasoning="Current environment has required capabilities",
                    strategy=strategy,
                    confidence=0.95
                )
        else:
            state["migration_decision"] = MigrationDecision(
                should_migrate=False,
                target_environment=None,
                reasoning="No suitable environments found",
                strategy=strategy,
                confidence=0.5
            )

Explanation: CAPABILITY strategy – find environments that meet task requirements. Choose one with most capabilities.

    elif strategy == MigrationStrategy.INTELLIGENT:
        # Use LLM to decide
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in distributed systems and agent placement optimization."),
            ("user", """Analyze this mobile agent migration scenario:

Agent ID: {agent_id}
Current Environment: {current_env}
  - Capabilities: {current_capabilities}
  - Load: {current_load:.2f}
  - Cost: ${current_cost:.2f}/hour

Task Requirements: {requirements}

Available Environments:
{environments}

Provide migration decision as JSON with:
- should_migrate: true/false
- target_environment: ID of target (or null)
- reasoning: Detailed explanation
- confidence: 0.0-1.0 confidence score

Consider: capabilities, load, cost, and migration overhead.""")
        ])

        # Format environments
        env_list = []
        for env_id, env in environments.items():
            if env_id != current_env_id:
                env_list.append(
                    f"  - {env_id}: {env.name}\n"
                    f"    Capabilities: {env.capabilities}\n"
                    f"    Load: {env.load:.2f}\n"
                    f"    Cost: ${env.cost_per_hour:.2f}/hour"
                )

Explanation: INTELLIGENT strategy – prepare comprehensive context for LLM.

        chain = prompt | self.llm
        response = chain.invoke({
            "agent_id": state["agent_id"],
            "current_env": current_env.name,
            "current_capabilities": current_env.capabilities,
            "current_load": current_env.load,
            "current_cost": current_env.cost_per_hour,
            "requirements": task.requirements,
            "environments": "\n".join(env_list)
        })

Explanation: Invoke LLM with full context about current and available environments.

        try:
            decision_data = json.loads(response.content)

            state["migration_decision"] = MigrationDecision(
                should_migrate=decision_data.get("should_migrate", False),
                target_environment=decision_data.get("target_environment"),
                reasoning=decision_data.get("reasoning", "LLM decision"),
                strategy=strategy,
                confidence=decision_data.get("confidence", 0.5)
            )

        except Exception as e:
            # Fallback to staying
            state["migration_decision"] = MigrationDecision(
                should_migrate=False,
                target_environment=None,
                reasoning=f"LLM decision failed: {str(e)}",
                strategy=strategy,
                confidence=0.0
            )

Explanation: Parse LLM response. Fallback to staying if parsing fails.

    state["target_environment"] = state["migration_decision"].target_environment
    return state

Lines 314-355: Serialize State Node

def _serialize_state_node(self, state: MobileAgentState) -> MobileAgentState:
    """Serialize agent state for migration."""
    decision = state["migration_decision"]

    if not decision or not decision.should_migrate:
        state["serialized_state"] = None
        state["insights"].append("No migration needed, skipping serialization")
        return state

Explanation: Skip serialization if not migrating.

    try:
        # Serialize agent state
        agent_state = state["agent_state"]
        serialized = pickle.dumps(agent_state)
        encoded = base64.b64encode(serialized).decode('utf-8')

        state["serialized_state"] = encoded
        state["insights"].append(f"State serialized: {len(encoded)} bytes")

    except Exception as e:
        state["insights"].append(f"Serialization failed: {str(e)}")
        state["serialized_state"] = None
        # Cancel migration
        state["migration_decision"].should_migrate = False

    return state

Explanation: Serialize state using pickle, encode as base64 string for transfer. If serialization fails, cancel migration.

Why pickle + base64? – Pickle: Python object serialization (preserves types, nested structures) – Base64: Encode binary as text (safe for JSON, network transfer)

Lines 357-420: Migrate Node

def _migrate_node(
    self,
    state: MobileAgentState,
    environments: Dict[str, Environment]
) -> MobileAgentState:
    """Migrate agent to new environment."""
    decision = state["migration_decision"]

    if not decision or not decision.should_migrate:
        state["insights"].append("Staying in current environment")
        return state

Explanation: Skip migration if decision says stay.

    target_env_id = decision.target_environment
    current_env_id = state["current_environment"]

    if target_env_id not in environments:
        state["insights"].append(f"Target environment {target_env_id} not found")
        return state

Explanation: Validate target environment exists.

    try:
        start_time = datetime.now()

        # Update environment tracking
        current_env = environments[current_env_id]
        target_env = environments[target_env_id]

        # Remove from current
        if state["agent_id"] in current_env.active_agents:
            current_env.active_agents.remove(state["agent_id"])

        # Add to target
        target_env.active_agents.append(state["agent_id"])

        # Update state
        state["current_environment"] = target_env_id
        state["status"] = AgentStatus.IDLE

        # Deserialize state in new environment
        if state["serialized_state"]:
            decoded = base64.b64decode(state["serialized_state"])
            deserialized_state = pickle.loads(decoded)
            state["agent_state"] = deserialized_state

Explanation: Perform migration: 1. Remove agent from current environment’s active list 2. Add agent to target environment’s active list 3. Update agent’s current_environment 4. Deserialize state in new environment

        # Record migration
        end_time = datetime.now()
        migration_time = (end_time - start_time).total_seconds() * 1000

        record = MigrationRecord(
            agent_id=state["agent_id"],
            from_environment=current_env_id,
            to_environment=target_env_id,
            state_size_bytes=len(state["serialized_state"]) if state["serialized_state"] else 0,
            migration_time_ms=migration_time,
            success=True
        )

        self.migration_history.append(record)
        self.stats["total_migrations"] += 1
        self.stats["successful_migrations"] += 1

        state["insights"].append(
            f"Migrated from {current_env.name} to {target_env.name} in {migration_time:.2f}ms"
        )

    except Exception as e:
        state["insights"].append(f"Migration failed: {str(e)}")
        self.stats["failed_migrations"] += 1

    return state

Explanation: Record migration metrics, update statistics, handle failures.

Lines 422-457: Execute Task Node

def _execute_task_node(self, state: MobileAgentState) -> MobileAgentState:
    """Execute agent task in current environment."""
    task = state["task"]

    state["status"] = AgentStatus.EXECUTING

    try:
        start_time = datetime.now()

        # Execute task code
        result = task.code(task.data)

        end_time = datetime.now()
        execution_time = (end_time - start_time).total_seconds() * 1000

        state["execution_result"] = result
        state["status"] = AgentStatus.IDLE

        self.stats["total_execution_time_ms"] += execution_time

        state["insights"].append(f"Task executed in {execution_time:.2f}ms")

    except Exception as e:
        state["execution_result"] = {"error": str(e)}
        state["status"] = AgentStatus.IDLE
        state["insights"].append(f"Execution failed: {str(e)}")

    return state

Explanation: Execute task code with data, measure execution time, handle errors.

Lines 482-640: Demo Code

Scenario 1: Setup Environments

# Create environments
environments = {
    "env1": Environment(
        id="env1",
        name="Standard Compute",
        capabilities=["python", "basic"],
        load=0.3,
        cost_per_hour=0.05,
        location="us-east-1"
    ),
    "env2": Environment(
        id="env2",
        name="GPU Compute",
        capabilities=["python", "gpu", "tensorflow"],
        load=0.6,
        cost_per_hour=0.50,
        location="us-west-2"
    ),
    "env3": Environment(
        id="env3",
        name="High Memory",
        capabilities=["python", "high_memory"],
        load=0.2,
        cost_per_hour=0.15,
        location="europe-west"
    )
}

Explanation: Define three diverse environments with different capabilities, loads, costs, and locations.

Scenario 2: Load Balanced Migration

agent1 = pattern.create_agent("agent1", "env1")

def simple_task(data: Dict) -> Dict:
    return {"result": data.get("value", 0) * 2}

task1 = AgentTask(
    id="task1",
    code=simple_task,
    data={"value": 10},
    requirements=[]
)

result1 = pattern.execute_with_migration(
    "agent1",
    task1,
    strategy=MigrationStrategy.LOAD_BALANCED
)

Explanation: Execute task with load balancing. Agent migrates from env1 (load=0.3) to env3 (load=0.2) since it’s less loaded.

Scenario 3: Capability-Based Migration

agent2 = pattern.create_agent("agent2", "env1")

def gpu_task(data: Dict) -> Dict:
    # Simulated GPU computation
    return {"processed": True, "gpu_accelerated": True}

task2 = AgentTask(
    id="task2",
    code=gpu_task,
    data={"matrix": "large"},
    requirements=["gpu", "tensorflow"]
)

result2 = pattern.execute_with_migration(
    "agent2",
    task2,
    strategy=MigrationStrategy.CAPABILITY
)

Explanation: Task requires GPU. Agent migrates from env1 (no GPU) to env2 (has GPU + TensorFlow).

Scenario 4: Intelligent LLM-Based Migration

agent3 = pattern.create_agent("agent3", "env2")

def memory_intensive_task(data: Dict) -> Dict:
    return {"memory_used": "high", "completed": True}

task3 = AgentTask(
    id="task3",
    code=memory_intensive_task,
    data={"large_dataset": True},
    requirements=["high_memory"]
)

result3 = pattern.execute_with_migration(
    "agent3",
    task3,
    strategy=MigrationStrategy.INTELLIGENT
)

Explanation: LLM analyzes: – Current: env2 (GPU, expensive $0.50/hour, high load 0.6) – Task needs: high memory – env3 has high_memory, lower cost ($0.15/hour), lower load (0.2) – Decision: Migrate to env3 (better match, cheaper, less loaded)

Migration Strategy Comparison

Key Design Patterns

1. Strategy Pattern: Different migration strategies
2. State Pattern: Agent status management
3. Memento Pattern: State serialization for migration
4. Proxy Pattern: Agent acts as proxy in remote environment
5. Observer Pattern: Migration history tracking

Performance Considerations

• Serialization: O(state_size) – pickle is fast but grows with state
• Migration: O(network_latency + deserialization) – typically 10-100ms
• Decision (Rule-Based): O(environments) – scan all options
• Decision (LLM): O(LLM) – ~1-2 seconds
• Execution: O(task_complexity) – depends on task

Best Practices: – Minimize agent state size (only essentials) – Use compression for large states (gzip before base64) – Batch migrate multiple agents to same destination – Cache LLM decisions for similar scenarios – Consider migration cost vs execution time savings – Use load balancing for homogeneous tasks – Use capability matching for specialized tasks

When to Migrate: – Migration time << execution time savings – Task requires unavailable capabilities – Current environment overloaded – Cost savings justify overhead

Extension Points

1. Add checkpointing during long-running tasks
2. Implement secure migration with encryption
3. Add fault tolerance (retry, rollback)
4. Support partial migration (leave state, move code)
5. Add network-aware routing (bandwidth, latency)
6. Implement agent cloning for parallel execution
7. Add environment auto-scaling based on agent demand
8. Support heterogeneous code (not just Python)

    """
Pattern 84: Service Agent Pattern
Purpose: Agent that provides services to other agents
"""

from typing import Dict, Any, List, Callable
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 ServiceType(Enum):
    """Types of services."""
    COMPUTATION = "computation"
    INFORMATION = "information"
    TRANSLATION = "translation"
    VALIDATION = "validation"
    TRANSFORMATION = "transformation"


@dataclass
class ServiceRequest:
    """A request for service."""
    id: str
    service_type: ServiceType
    payload: Any
    requester: str
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class ServiceResponse:
    """A response from service."""
    request_id: str
    result: Any
    success: bool
    error: str = None
    completed_at: datetime = field(default_factory=datetime.now)


class ServiceAgent:
    """
    Service Agent - Provides services to other agents
    
    Offers specialized services that other agents can consume.
    Maintains service catalog and handles requests.
    """
    
    def __init__(self, name: str, services: List[ServiceType], model_name: str = "gpt-4"):
        self.name = name
        self.services = services
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.request_history: List[ServiceRequest] = []
        self.response_history: List[ServiceResponse] = []
        
    def provides_service(self, service_type: ServiceType) -> bool:
        """Check if agent provides this service."""
        return service_type in self.services
    
    def process_request(self, request: ServiceRequest) -> ServiceResponse:
        """Process a service request."""
        self.request_history.append(request)
        
        if not self.provides_service(request.service_type):
            response = ServiceResponse(
                request_id=request.id,
                result=None,
                success=False,
                error=f"Service {request.service_type.value} not provided"
            )
            self.response_history.append(response)
            return response
        
        try:
            # Process using LLM
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {self.name}, providing {request.service_type.value} service"),
                ("human", "Process this request: {payload}")
            ])
            
            chain = prompt | self.llm
            result = chain.invoke({"payload": str(request.payload)})
            
            response = ServiceResponse(
                request_id=request.id,
                result=str(result.content),
                success=True
            )
        except Exception as e:
            response = ServiceResponse(
                request_id=request.id,
                result=None,
                success=False,
                error=str(e)
            )
        
        self.response_history.append(response)
        return response
    
    def get_service_catalog(self) -> Dict[str, Any]:
        """Get catalog of provided services."""
        return {
            "agent": self.name,
            "services": [s.value for s in self.services],
            "requests_handled": len(self.request_history),
            "success_rate": sum(1 for r in self.response_history if r.success) / len(self.response_history) if self.response_history else 0
        }


class ServiceAgentPattern:
    """
    Service Agent Pattern - Manages service agents and routing
    """
    
    def __init__(self):
        self.agents: Dict[str, ServiceAgent] = {}
        self.service_registry: Dict[ServiceType, List[str]] = {}
        
    def register_agent(self, agent: ServiceAgent):
        """Register a service agent."""
        self.agents[agent.name] = agent
        
        # Update service registry
        for service in agent.services:
            if service not in self.service_registry:
                self.service_registry[service] = []
            self.service_registry[service].append(agent.name)
        
        print(f"✓ Registered: {agent.name} - Services: {[s.value for s in agent.services]}")
    
    def find_provider(self, service_type: ServiceType) -> ServiceAgent:
        """Find agent that provides service."""
        if service_type in self.service_registry:
            agent_name = self.service_registry[service_type][0]
            return self.agents[agent_name]
        return None
    
    def request_service(self, service_type: ServiceType, payload: Any, requester: str = "client") -> ServiceResponse:
        """Request a service."""
        provider = self.find_provider(service_type)
        
        if not provider:
            return ServiceResponse(
                request_id="no_provider",
                result=None,
                success=False,
                error=f"No provider for {service_type.value}"
            )
        
        request = ServiceRequest(
            id=f"req_{len(provider.request_history) + 1}",
            service_type=service_type,
            payload=payload,
            requester=requester
        )
        
        return provider.process_request(request)
    
    def run(self, requests: List[tuple]) -> Dict[str, Any]:
        """Process multiple service requests."""
        results = []
        
        for service_type, payload in requests:
            response = self.request_service(service_type, payload)
            results.append({
                "service": service_type.value,
                "success": response.success,
                "result": response.result[:100] if response.result else None,
                "error": response.error
            })
        
        return {
            "pattern": "ServiceAgent",
            "total_requests": len(requests),
            "successful": sum(1 for r in results if r['success']),
            "results": results
        }


if __name__ == "__main__":
    print("=== Service Agent Pattern Demo ===\n")
    
    pattern = ServiceAgentPattern()
    
    # Register service providers
    pattern.register_agent(ServiceAgent(
        "ComputeService", 
        [ServiceType.COMPUTATION, ServiceType.VALIDATION]
    ))
    pattern.register_agent(ServiceAgent(
        "DataService",
        [ServiceType.INFORMATION, ServiceType.TRANSFORMATION]
    ))
    pattern.register_agent(ServiceAgent(
        "LinguisticsService",
        [ServiceType.TRANSLATION]
    ))
    
    # Make service requests
    print("\nRequesting services...\n")
    requests = [
        (ServiceType.COMPUTATION, "Calculate fibonacci(10)"),
        (ServiceType.TRANSLATION, "Translate 'hello' to Spanish"),
        (ServiceType.INFORMATION, "Get user data for ID 12345"),
        (ServiceType.VALIDATION, "Validate email: user@example.com")
    ]
    
    summary = pattern.run(requests)
    
    print(f"\nPattern: {summary['pattern']}")
    print(f"Total Requests: {summary['total_requests']}")
    print(f"Successful: {summary['successful']}/{summary['total_requests']}")
    
    print("\nService Results:")
    for result in summary['results']:
        status = "✓" if result['success'] else "✗"
        print(f"  {status} {result['service']}: {result['result'] or result['error']}")

    

Agentic Architecture Pattern: Service Agent

Pattern 84: 84 Service Agent – Line-by-Line Explanation

Overview

Pattern 84: Service Agent Pattern Purpose: Agent that provides services to other agents

Code Structure Analysis

File Information

• Pattern Number: 84
• Pattern Name: 84 Service Agent
• File Path: patterns\specialized_agent\84_service_agent.py
• Total Lines: 220

Detailed Code Explanation

Module Docstring

"""
Pattern 84: Service Agent Pattern
Purpose: Agent that provides services to other agents
"""

Explanation: This module docstring describes the purpose and key features of the 84 Service Agent pattern.

Imports Section

from typing import Dict, Any, List, Callable
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, Callable: 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

ServiceType (Line 14)

class ServiceType(Enum):
    COMPUTATION
    INFORMATION
    TRANSLATION
    VALIDATION
    TRANSFORMATION

Purpose: Types of services.

Values: – COMPUTATION: Represents computation – INFORMATION: Represents information – TRANSLATION: Represents translation – VALIDATION: Represents validation – TRANSFORMATION: Represents transformation

Classes

ServiceType (Line 14)

Description: Types of services.

ServiceRequest (Line 24)

Description: A request for service.

ServiceResponse (Line 34)

Description: A response from service.

ServiceAgent (Line 43)

Description: Service Agent – Provides services to other agents

Offers specialized services that other agents can consume. Maintains service catalog and handles requests.

Methods:

__init__ (Line 51)

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

Purpose: No method docstring.

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

provides_service (Line 58)

def provides_service(self, service_type):
    """Check if agent provides this service."""

Purpose: Check if agent provides this service.

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

process_request (Line 62)

def process_request(self, request):
    """Process a service request."""

Purpose: Process a service request.

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

get_service_catalog (Line 102)

def get_service_catalog(self):
    """Get catalog of provided services."""

Purpose: Get catalog of provided services.

Parameters: – self: Instance reference

ServiceAgentPattern (Line 112)

Description: Service Agent Pattern – Manages service agents and routing

Methods:

__init__ (Line 117)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 121)

def register_agent(self, agent):
    """Register a service agent."""

Purpose: Register a service agent.

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

find_provider (Line 133)

def find_provider(self, service_type):
    """Find agent that provides service."""

Purpose: Find agent that provides service.

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

request_service (Line 140)

def request_service(self, service_type, payload, requester):
    """Request a service."""

Purpose: Request a service.

Parameters: – self: Instance reference – service_type: Parameter for service_type – payload: Parameter for payload – requester: Parameter for requester

run (Line 161)

def run(self, requests):
    """Process multiple service requests."""

Purpose: Process multiple service requests.

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

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

    """
Pattern 83: Task Agent Pattern
Purpose: Specialized agent for executing specific tasks
"""

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 TaskStatus(Enum):
    """Task execution status."""
    PENDING = "pending"
    ASSIGNED = "assigned"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    """A task to be executed."""
    id: str
    description: str
    priority: int = 1
    status: TaskStatus = TaskStatus.PENDING
    assigned_to: Optional[str] = None
    result: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None


class TaskAgent:
    """
    Task Agent - Executes specific tasks
    
    Specialized agent responsible for task execution.
    Can handle multiple tasks with priorities.
    """
    
    def __init__(self, name: str, capabilities: List[str], model_name: str = "gpt-4"):
        self.name = name
        self.capabilities = capabilities
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.assigned_tasks: Dict[str, Task] = {}
        self.completed_tasks: List[Task] = []
        
    def can_handle(self, task: Task) -> bool:
        """Check if agent can handle this task."""
        # Simple keyword matching
        task_lower = task.description.lower()
        return any(cap.lower() in task_lower for cap in self.capabilities)
    
    def assign_task(self, task: Task) -> bool:
        """Assign task to this agent."""
        if not self.can_handle(task):
            return False
            
        task.status = TaskStatus.ASSIGNED
        task.assigned_to = self.name
        self.assigned_tasks[task.id] = task
        print(f"   {self.name}: Assigned task '{task.id}' - {task.description}")
        return True
    
    def execute_task(self, task_id: str) -> Dict[str, Any]:
        """Execute assigned task."""
        if task_id not in self.assigned_tasks:
            return {"error": "Task not found"}
        
        task = self.assigned_tasks[task_id]
        task.status = TaskStatus.IN_PROGRESS
        
        # Execute using LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a task agent with capabilities: {', '.join(self.capabilities)}"),
            ("human", "Execute this task: {task_description}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task_description": task.description})
        
        task.status = TaskStatus.COMPLETED
        task.result = str(response.content)
        task.completed_at = datetime.now()
        
        # Move to completed
        del self.assigned_tasks[task_id]
        self.completed_tasks.append(task)
        
        return {
            "task_id": task_id,
            "agent": self.name,
            "status": task.status.value,
            "result": task.result[:100] + "..." if len(task.result) > 100 else task.result
        }
    
    def execute_all(self) -> List[Dict[str, Any]]:
        """Execute all assigned tasks."""
        results = []
        task_ids = list(self.assigned_tasks.keys())
        
        for task_id in task_ids:
            result = self.execute_task(task_id)
            results.append(result)
        
        return results


class TaskAgentPattern:
    """
    Task Agent Pattern - Manages task agents and task distribution
    """
    
    def __init__(self):
        self.agents: Dict[str, TaskAgent] = {}
        self.tasks: Dict[str, Task] = {}
        
    def register_agent(self, agent: TaskAgent):
        """Register a task agent."""
        self.agents[agent.name] = agent
        print(f"✓ Registered agent: {agent.name} (capabilities: {', '.join(agent.capabilities)})")
    
    def create_task(self, task: Task):
        """Create and queue a task."""
        self.tasks[task.id] = task
    
    def distribute_tasks(self) -> Dict[str, Any]:
        """Distribute tasks to capable agents."""
        assigned = 0
        unassigned = []
        
        for task_id, task in list(self.tasks.items()):
            if task.status == TaskStatus.PENDING:
                # Find capable agent
                for agent in self.agents.values():
                    if agent.assign_task(task):
                        del self.tasks[task_id]
                        assigned += 1
                        break
                else:
                    unassigned.append(task_id)
        
        return {
            "assigned": assigned,
            "unassigned": len(unassigned),
            "unassigned_tasks": unassigned
        }
    
    def execute_all_tasks(self) -> Dict[str, Any]:
        """Execute all tasks across all agents."""
        all_results = []
        
        for agent in self.agents.values():
            results = agent.execute_all()
            all_results.extend(results)
        
        return {
            "pattern": "TaskAgent",
            "total_executed": len(all_results),
            "results": all_results
        }
    
    def run(self, tasks: List[Task]) -> Dict[str, Any]:
        """Main execution: create, distribute, and execute tasks."""
        # Create tasks
        for task in tasks:
            self.create_task(task)
        
        # Distribute
        distribution = self.distribute_tasks()
        print(f"\nDistribution: {distribution['assigned']} assigned, {distribution['unassigned']} unassigned")
        
        # Execute
        print("\nExecuting tasks...\n")
        execution = self.execute_all_tasks()
        
        return {
            "pattern": "TaskAgent",
            "distribution": distribution,
            "execution": execution
        }


if __name__ == "__main__":
    print("=== Task Agent Pattern Demo ===\n")
    
    pattern = TaskAgentPattern()
    
    # Register specialized agents
    pattern.register_agent(TaskAgent("DataProcessor", ["data", "process", "analyze"]))
    pattern.register_agent(TaskAgent("ReportGenerator", ["report", "generate", "document"]))
    pattern.register_agent(TaskAgent("Notifier", ["notify", "alert", "send"]))
    
    # Create tasks
    tasks = [
        Task("t1", "Process customer data and extract insights", priority=1),
        Task("t2", "Generate monthly sales report", priority=2),
        Task("t3", "Send notification to stakeholders", priority=1),
        Task("t4", "Analyze market trends data", priority=2),
    ]
    
    print("\nTasks created:")
    for task in tasks:
        print(f"  {task.id}: {task.description}")
    
    # Run pattern
    summary = pattern.run(tasks)
    
    print("\n" + "="*70)
    print(f"Completed {summary['execution']['total_executed']} tasks!")
    
    print("\nTask Results:")
    for result in summary['execution']['results'][:2]:
        print(f"  {result['task_id']} ({result['agent']}): {result['result']}")

    

Agentic Architecture Pattern: Task Agent

Pattern 83: 83 Task Agent – Line-by-Line Explanation

Overview

Pattern 83: Task Agent Pattern Purpose: Specialized agent for executing specific tasks

Code Structure Analysis

File Information

• Pattern Number: 83
• Pattern Name: 83 Task Agent
• File Path: patterns\specialized_agent\83_task_agent.py
• Total Lines: 218

Detailed Code Explanation

Module Docstring

"""
Pattern 83: Task Agent Pattern
Purpose: Specialized agent for executing specific tasks
"""

Explanation: This module docstring describes the purpose and key features of the 83 Task Agent 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

TaskStatus (Line 14)

class TaskStatus(Enum):
    PENDING
    ASSIGNED
    IN_PROGRESS
    COMPLETED
    FAILED

Purpose: Task execution status.

Values: – PENDING: Initial state, not yet processed – ASSIGNED: Represents assigned – IN_PROGRESS: Represents in progress – COMPLETED: Successfully finished – FAILED: Execution failed

Classes

TaskStatus (Line 14)

Description: Task execution status.

Task (Line 24)

Description: A task to be executed.

TaskAgent (Line 36)

Description: Task Agent – Executes specific tasks

Specialized agent responsible for task execution. Can handle multiple tasks with priorities.

Methods:

__init__ (Line 44)

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

Purpose: No method docstring.

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

can_handle (Line 51)

def can_handle(self, task):
    """Check if agent can handle this task."""

Purpose: Check if agent can handle this task.

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

assign_task (Line 57)

def assign_task(self, task):
    """Assign task to this agent."""

Purpose: Assign task to this agent.

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

execute_task (Line 68)

def execute_task(self, task_id):
    """Execute assigned task."""

Purpose: Execute assigned task.

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

execute_all (Line 100)

def execute_all(self):
    """Execute all assigned tasks."""

Purpose: Execute all assigned tasks.

Parameters: – self: Instance reference

TaskAgentPattern (Line 112)

Description: Task Agent Pattern – Manages task agents and task distribution

Methods:

__init__ (Line 117)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 121)

def register_agent(self, agent):
    """Register a task agent."""

Purpose: Register a task agent.

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

create_task (Line 126)

def create_task(self, task):
    """Create and queue a task."""

Purpose: Create and queue a task.

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

distribute_tasks (Line 130)

def distribute_tasks(self):
    """Distribute tasks to capable agents."""

Purpose: Distribute tasks to capable agents.

Parameters: – self: Instance reference

execute_all_tasks (Line 152)

def execute_all_tasks(self):
    """Execute all tasks across all agents."""

Purpose: Execute all tasks across all agents.

Parameters: – self: Instance reference

run (Line 166)

def run(self, tasks):
    """Main execution: create, distribute, and execute tasks."""

Purpose: Main execution: create, distribute, and execute tasks.

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

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

    """
Pattern 82: Information Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class InformationAgentConfig:
    """Configuration for Information Agent."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class InformationAgentState:
    """State information for Information Agent."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class InformationAgentAgent:
    """Agent implementing Information Agent."""
    
    def __init__(self, name: str, config: InformationAgentConfig = None):
        self.name = name
        self.config = config or InformationAgentConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = InformationAgentState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Information Agent."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Information Agent. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class InformationAgentPattern:
    """
    Information Agent - Enhanced implementation
    
    This pattern provides comprehensive functionality for information agent
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: InformationAgentConfig = None):
        self.config = config or InformationAgentConfig()
        self.agents: Dict[str, InformationAgentAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Information Agent. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: InformationAgentAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Information Agent",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Information Agent",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Information Agent Demo ===\n")
    
    # Create configuration
    config = InformationAgentConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = InformationAgentPattern(config)
    
    # Create and register agents
    agent1 = InformationAgentAgent("Agent1", config)
    agent2 = InformationAgentAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate information agent functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

    

Agentic Architecture Pattern: Information Agent

Pattern 82: 82 Information Agent – Line-by-Line Explanation

Overview

Pattern 82: Information Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 82
• Pattern Name: 82 Information Agent
• File Path: patterns\specialized_agent\82_information_agent.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 82: Information Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 82 Information Agent 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
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

InformationAgentConfig (Line 18)

Description: Configuration for Information Agent.

InformationAgentState (Line 27)

Description: State information for Information Agent.

InformationAgentAgent (Line 36)

Description: Agent implementing Information Agent.

Methods:

__init__ (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

process (Line 49)

def process(self, input_data):
    """Process input according to Information Agent."""

Purpose: Process input according to Information Agent.

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

get_statistics (Line 75)

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

Purpose: Get agent statistics.

Parameters: – self: Instance reference

InformationAgentPattern (Line 85)

Description: Information Agent – Enhanced implementation

This pattern provides comprehensive functionality for information agent with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

__init__ (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

_build_workflow (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

register_agent (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

execute (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

run (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

get_statistics (Line 192)

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

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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

    """
Pattern 81: Interface Agent Pattern
Purpose: Agent that provides interface to external systems
"""

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


@dataclass
class InterfaceRequest:
    """A request through interface."""
    id: str
    system: str
    operation: str
    parameters: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class InterfaceResponse:
    """Response from interface."""
    request_id: str
    status: str
    data: Any
    error: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)


class InterfaceAgent:
    """
    Interface Agent - Bridges AI agents and external systems
    
    Provides standardized interface to external systems,
    handles protocol translation, and manages communication.
    """
    
    def __init__(self, name: str, systems: List[str], model_name: str = "gpt-4"):
        self.name = name
        self.systems = systems
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.requests: List[InterfaceRequest] = []
        self.responses: List[InterfaceResponse] = []
        
    def can_interface(self, system: str) -> bool:
        """Check if agent can interface with system."""
        return system in self.systems
    
    def send_request(self, request: InterfaceRequest) -> InterfaceResponse:
        """Send request to external system."""
        self.requests.append(request)
        
        if not self.can_interface(request.system):
            response = InterfaceResponse(
                request_id=request.id,
                status="error",
                data=None,
                error=f"Cannot interface with {request.system}"
            )
            self.responses.append(response)
            return response
        
        # Simulate interfacing with LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are interfacing with {request.system}. Translate and execute request."),
            ("human", "Operation: {operation}\nParameters: {parameters}")
        ])
        
        chain = prompt | self.llm
        result = chain.invoke({
            "operation": request.operation,
            "parameters": str(request.parameters)
        })
        
        response = InterfaceResponse(
            request_id=request.id,
            status="success",
            data=str(result.content)
        )
        
        self.responses.append(response)
        return response
    
    def get_stats(self) -> Dict[str, Any]:
        """Get interface statistics."""
        success_count = sum(1 for r in self.responses if r.status == "success")
        
        return {
            "agent": self.name,
            "systems": self.systems,
            "total_requests": len(self.requests),
            "successful": success_count,
            "success_rate": success_count / len(self.responses) if self.responses else 0
        }


class InterfaceAgentPattern:
    """Interface Agent Pattern - Manages interface agents."""
    
    def __init__(self):
        self.agents: Dict[str, InterfaceAgent] = {}
        
    def register_agent(self, agent: InterfaceAgent):
        """Register interface agent."""
        self.agents[agent.name] = agent
        print(f"✓ Registered: {agent.name} (systems: {', '.join(agent.systems)})")
    
    def find_agent_for_system(self, system: str) -> Optional[InterfaceAgent]:
        """Find agent that can interface with system."""
        for agent in self.agents.values():
            if agent.can_interface(system):
                return agent
        return None
    
    def execute_request(self, system: str, operation: str, parameters: Dict[str, Any]) -> InterfaceResponse:
        """Execute request on external system."""
        agent = self.find_agent_for_system(system)
        
        if not agent:
            return InterfaceResponse(
                request_id="no_agent",
                status="error",
                data=None,
                error=f"No agent for system {system}"
            )
        
        request = InterfaceRequest(
            id=f"req_{len(agent.requests) + 1}",
            system=system,
            operation=operation,
            parameters=parameters
        )
        
        return agent.send_request(request)
    
    def run(self, requests: List[tuple]) -> Dict[str, Any]:
        """Execute multiple interface requests."""
        results = []
        
        for system, operation, parameters in requests:
            response = self.execute_request(system, operation, parameters)
            results.append({
                "system": system,
                "operation": operation,
                "status": response.status,
                "data": response.data[:100] if response.data else None,
                "error": response.error
            })
        
        return {
            "pattern": "InterfaceAgent",
            "total_requests": len(requests),
            "results": results
        }


if __name__ == "__main__":
    print("=== Interface Agent Pattern Demo ===\n")
    
    pattern = InterfaceAgentPattern()
    
    # Register interface agents
    pattern.register_agent(InterfaceAgent("DatabaseAgent", ["PostgreSQL", "MongoDB"]))
    pattern.register_agent(InterfaceAgent("APIAgent", ["REST", "GraphQL"]))
    pattern.register_agent(InterfaceAgent("FileAgent", ["S3", "FileSystem"]))
    
    print("\nExecuting requests...\n")
    requests = [
        ("PostgreSQL", "query", {"sql": "SELECT * FROM users"}),
        ("REST", "get", {"endpoint": "/api/users"}),
        ("S3", "upload", {"bucket": "data", "file": "report.pdf"}),
    ]
    
    summary = pattern.run(requests)
    
    print(f"Pattern: {summary['pattern']}")
    print(f"Total Requests: {summary['total_requests']}\n")
    
    print("Results:")
    for result in summary['results']:
        status_icon = "✓" if result['status'] == "success" else "✗"
        print(f"  {status_icon} {result['system']}.{result['operation']}: {result['data'] or result['error']}")

    

Agentic Architecture Pattern: Interface Agent

Pattern 81: 81 Interface Agent – Line-by-Line Explanation

Overview

Pattern 81: Interface Agent Pattern Purpose: Agent that provides interface to external systems

Code Structure Analysis

File Information

• Pattern Number: 81
• Pattern Name: 81 Interface Agent
• File Path: patterns\specialized_agent\81_interface_agent.py
• Total Lines: 186

Detailed Code Explanation

Module Docstring

"""
Pattern 81: Interface Agent Pattern
Purpose: Agent that provides interface to external systems
"""

Explanation: This module docstring describes the purpose and key features of the 81 Interface Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
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 langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Classes

InterfaceRequest (Line 14)

Description: A request through interface.

InterfaceResponse (Line 24)

Description: Response from interface.

InterfaceAgent (Line 33)

Description: Interface Agent – Bridges AI agents and external systems

Provides standardized interface to external systems, handles protocol translation, and manages communication.

Methods:

__init__ (Line 41)

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

Purpose: No method docstring.

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

can_interface (Line 48)

def can_interface(self, system):
    """Check if agent can interface with system."""

Purpose: Check if agent can interface with system.

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

send_request (Line 52)

def send_request(self, request):
    """Send request to external system."""

Purpose: Send request to external system.

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

get_stats (Line 87)

def get_stats(self):
    """Get interface statistics."""

Purpose: Get interface statistics.

Parameters: – self: Instance reference

InterfaceAgentPattern (Line 100)

Description: Interface Agent Pattern – Manages interface agents.

Methods:

__init__ (Line 103)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 106)

def register_agent(self, agent):
    """Register interface agent."""

Purpose: Register interface agent.

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

find_agent_for_system (Line 111)

def find_agent_for_system(self, system):
    """Find agent that can interface with system."""

Purpose: Find agent that can interface with system.

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

execute_request (Line 118)

def execute_request(self, system, operation, parameters):
    """Execute request on external system."""

Purpose: Execute request on external system.

Parameters: – self: Instance reference – system: Parameter for system – operation: Parameter for operation – parameters: Parameter for parameters

run (Line 139)

def run(self, requests):
    """Execute multiple interface requests."""

Purpose: Execute multiple interface requests.

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

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