"""
Pattern 130: Cloud Agent Pattern
Cloud-based agent with scalability
Implements distributed agent coordination using LangChain/LangGraph.
"""
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
import random
class AgentStatus(Enum):
"""Status of distributed agent."""
ACTIVE = "active"
IDLE = "idle"
BUSY = "busy"
FAILED = "failed"
@dataclass
class DistributedAgent:
"""Represents a distributed agent node."""
agent_id: str
status: AgentStatus = AgentStatus.IDLE
load: float = 0.0 # 0.0 to 1.0
capabilities: List[str] = field(default_factory=list)
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class Task:
"""Task to be executed."""
task_id: str
description: str
assigned_to: Optional[str] = None
status: str = "pending"
result: Optional[str] = None
class CloudagentPattern:
"""
Cloud Agent Pattern - Cloud-based agent with scalability
Coordinates multiple distributed agents for scalable and fault-tolerant
execution of tasks across a distributed system.
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.3)
self.agents: Dict[str, DistributedAgent] = {}
self.tasks: List[Task] = []
def register_agent(self, agent_id: str, capabilities: List[str] = None):
"""Register a new distributed agent."""
agent = DistributedAgent(
agent_id=agent_id,
capabilities=capabilities or ["general"],
status=AgentStatus.IDLE
)
self.agents[agent_id] = agent
return agent
def select_agent(self, task: Task) -> Optional[str]:
"""Select best agent for task."""
available = [
(aid, agent) for aid, agent in self.agents.items()
if agent.status in [AgentStatus.IDLE, AgentStatus.ACTIVE] and agent.load bool:
"""Assign task to an agent."""
agent_id = self.select_agent(task)
if not agent_id:
return False
task.assigned_to = agent_id
task.status = "assigned"
self.agents[agent_id].load += 0.2
self.agents[agent_id].status = AgentStatus.BUSY
return True
def execute_task(self, task: Task) -> Dict[str, Any]:
"""Execute task on assigned agent."""
if not task.assigned_to:
if not self.assign_task(task):
return {"status": "failed", "reason": "no_agent_available"}
agent = self.agents[task.assigned_to]
# Simulate task execution with LLM
prompt = ChatPromptTemplate.from_messages([
("system", f"You are agent {agent.agent_id} in a Cloud Agent system."),
("human", "{task_description}")
])
chain = prompt | self.llm
response = chain.invoke({"task_description": task.description})
task.result = str(response.content)
task.status = "completed"
# Update agent status
agent.load = max(0.0, agent.load - 0.2)
if agent.load Dict[str, Any]:
"""Main execution method."""
task = Task(
task_id=f"task_{len(self.tasks) + 1}",
description=task_description
)
self.tasks.append(task)
result = self.execute_task(task)
return {
"pattern": "Cloud Agent",
"task_id": task.task_id,
"assigned_to": task.assigned_to,
"status": result["status"],
"result": result.get("result", ""),
"total_agents": len(self.agents),
"active_agents": sum(1 for a in self.agents.values() if a.status == AgentStatus.ACTIVE or a.status == AgentStatus.BUSY)
}
# Example usage
if __name__ == "__main__":
print("=== Cloud Agent Pattern Demo ===\n")
pattern = CloudagentPattern()
# Register agents
for i in range(3):
pattern.register_agent(f"agent_{i+1}", ["compute", "process"])
print(f"Registered {len(pattern.agents)} agents")
# Execute tasks
tasks = [
"Process data batch 1",
"Analyze results",
"Generate report"
]
for task_desc in tasks:
result = pattern.run(task_desc)
print(f"\nTask: {task_desc}")
print(f" Assigned to: {result['assigned_to']}")
print(f" Status: {result['status']}")
print(f" Result: {result['result'][:100]}...")
print(f"\nTotal tasks executed: {len(pattern.tasks)}")
print(f"Active agents: {result['active_agents']}/{result['total_agents']}")
Agentic Architecture Pattern: Cloud Agent
Pattern 130: 130 Cloud Agent Pattern – Line-by-Line Explanation
Overview
Pattern 130: Cloud Agent Pattern
Cloud-based agent with scalability Implements distributed agent coordination using LangChain/LangGraph.
Code Structure Analysis
File Information
- Pattern Number: 130
- Pattern Name: 130 Cloud Agent Pattern
- File Path:
patterns\distributed_agent\130_cloud_agent_pattern.py - Total Lines: 178
Detailed Code Explanation
Module Docstring
"""
Pattern 130: Cloud Agent Pattern
Cloud-based agent with scalability
Implements distributed agent coordination using LangChain/LangGraph.
"""
Explanation: This module docstring describes the purpose and key features of the 130 Cloud Agent Pattern 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
import random # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: Type hints for better code documentation
– from 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
Enumerations
AgentStatus (Line 19)
class AgentStatus(Enum):
ACTIVE
IDLE
BUSY
FAILED
Purpose: Status of distributed agent.
Values:
– ACTIVE: Currently operational
– IDLE: Not currently active
– BUSY: Represents busy
– FAILED: Execution failed
Classes
AgentStatus (Line 19)
Description: Status of distributed agent.
DistributedAgent (Line 28)
Description: Represents a distributed agent node.
Task (Line 38)
Description: Task to be executed.
CloudagentPattern (Line 47)
Description: Cloud Agent Pattern – Cloud-based agent with scalability
Coordinates multiple distributed agents for scalable and fault-tolerant execution of tasks across a distributed system.
Methods:
__init__ (Line 55)
def __init__(self, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– model_name: LLM model to use (e.g., “gpt-4”)
register_agent (Line 60)
def register_agent(self, agent_id, capabilities):
"""Register a new distributed agent."""
Purpose: Register a new distributed agent.
Parameters:
– self: Instance reference
– agent_id: Parameter for agent_id
– capabilities: Parameter for capabilities
select_agent (Line 70)
def select_agent(self, task):
"""Select best agent for task."""
Purpose: Select best agent for task.
Parameters:
– self: Instance reference
– task: Parameter for task
assign_task (Line 84)
def assign_task(self, task):
"""Assign task to an agent."""
Purpose: Assign task to an agent.
Parameters:
– self: Instance reference
– task: Parameter for task
execute_task (Line 97)
def execute_task(self, task):
"""Execute task on assigned agent."""
Purpose: Execute task on assigned agent.
Parameters:
– self: Instance reference
– task: Parameter for task
run (Line 129)
def run(self, task_description):
"""Main execution method."""
Purpose: Main execution method.
Parameters:
– self: Instance reference
– task_description: Parameter for task_description
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 = AgentStatus(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