"""
Pattern 102: Monitor Agent Pattern
Agent that continuously watches specific conditions and triggers actions based on rules.
Implements reactive monitoring with LangChain-based rule evaluation.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import time
import threading
class MonitorStatus(Enum):
"""Status of monitor."""
HEALTHY = "healthy"
WARNING = "warning"
CRITICAL = "critical"
UNKNOWN = "unknown"
@dataclass
class MonitorRule:
"""Rule for monitoring."""
rule_id: str
name: str
condition: Callable[[Any], bool]
action: Callable[[Any], None]
priority: int = 1 # 1=low, 5=critical
enabled: bool = True
trigger_count: int = 0
last_triggered: Optional[datetime] = None
@dataclass
class MonitorEvent:
"""Event detected by monitor."""
event_id: str
monitor_id: str
rule_id: str
status: MonitorStatus
data: Any
message: str
timestamp: datetime = field(default_factory=datetime.now)
class MonitorAgent:
"""
Agent that monitors a specific target (system, service, metric) and takes action
when conditions are met.
"""
def __init__(
self,
monitor_id: str,
name: str,
check_function: Callable[[], Any],
check_interval: int = 5,
model_name: str = "gpt-4"
):
self.monitor_id = monitor_id
self.name = name
self.check_function = check_function
self.check_interval = check_interval
self.llm = ChatOpenAI(model=model_name, temperature=0.3)
self.rules: Dict[str, MonitorRule] = {}
self.events: List[MonitorEvent] = []
self.status = MonitorStatus.UNKNOWN
self.is_monitoring = False
self._monitor_thread: Optional[threading.Thread] = None
def add_rule(self, rule: MonitorRule):
"""Add a monitoring rule."""
self.rules[rule.rule_id] = rule
def remove_rule(self, rule_id: str):
"""Remove a monitoring rule."""
if rule_id in self.rules:
del self.rules[rule_id]
def check(self) -> Dict[str, Any]:
"""Perform a single check."""
data = self.check_function()
# Evaluate all rules
triggered_rules = []
for rule in self.rules.values():
if rule.enabled and rule.condition(data):
rule.trigger_count += 1
rule.last_triggered = datetime.now()
triggered_rules.append(rule)
# Execute action
try:
rule.action(data)
except Exception as e:
print(f"Error executing action for rule {rule.rule_id}: {e}")
# Determine status using LLM
if triggered_rules:
status = self._determine_status(data, triggered_rules)
message = f"{len(triggered_rules)} rule(s) triggered"
# Create event
for rule in triggered_rules:
event = MonitorEvent(
event_id=f"evt-{len(self.events)+1}",
monitor_id=self.monitor_id,
rule_id=rule.rule_id,
status=status,
data=data,
message=f"{rule.name}: {message}"
)
self.events.append(event)
else:
status = MonitorStatus.HEALTHY
message = "All checks passed"
self.status = status
return {
"monitor_id": self.monitor_id,
"status": status.value,
"data": data,
"triggered_rules": len(triggered_rules),
"message": message
}
def _determine_status(self, data: Any, triggered_rules: List[MonitorRule]) -> MonitorStatus:
"""Determine overall status based on triggered rules."""
max_priority = max(rule.priority for rule in triggered_rules)
if max_priority >= 4:
return MonitorStatus.CRITICAL
elif max_priority >= 2:
return MonitorStatus.WARNING
else:
return MonitorStatus.HEALTHY
def start_monitoring(self):
"""Start continuous monitoring in background thread."""
if self.is_monitoring:
return
self.is_monitoring = True
self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._monitor_thread.start()
def stop_monitoring(self):
"""Stop continuous monitoring."""
self.is_monitoring = False
if self._monitor_thread:
self._monitor_thread.join(timeout=self.check_interval + 1)
def _monitor_loop(self):
"""Background monitoring loop."""
while self.is_monitoring:
self.check()
time.sleep(self.check_interval)
def get_statistics(self) -> Dict[str, Any]:
"""Get monitoring statistics."""
return {
"monitor_id": self.monitor_id,
"name": self.name,
"status": self.status.value,
"total_events": len(self.events),
"critical_events": len([e for e in self.events if e.status == MonitorStatus.CRITICAL]),
"warning_events": len([e for e in self.events if e.status == MonitorStatus.WARNING]),
"rules": len(self.rules),
"active_rules": len([r for r in self.rules.values() if r.enabled])
}
class MonitorAgentPattern:
"""
Coordinates multiple monitor agents for comprehensive system monitoring.
Provides centralized alerting and dashboard capabilities.
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.5)
self.monitors: Dict[str, MonitorAgent] = {}
def register_monitor(self, monitor: MonitorAgent):
"""Register a monitor agent."""
self.monitors[monitor.monitor_id] = monitor
def check_all(self) -> Dict[str, Dict[str, Any]]:
"""Check all monitors."""
results = {}
for monitor_id, monitor in self.monitors.items():
results[monitor_id] = monitor.check()
return results
def start_all(self):
"""Start all monitors."""
for monitor in self.monitors.values():
monitor.start_monitoring()
def stop_all(self):
"""Stop all monitors."""
for monitor in self.monitors.values():
monitor.stop_monitoring()
def get_system_status(self) -> str:
"""Get overall system status using LLM analysis."""
stats = {mid: m.get_statistics() for mid, m in self.monitors.items()}
prompt = ChatPromptTemplate.from_messages([
("system", "You are a system monitoring AI. Analyze monitor data and provide status."),
("human", """Monitor Statistics:
{stats}
Provide:
1. Overall system health (HEALTHY/WARNING/CRITICAL)
2. Summary of issues
3. Recommended actions""")
])
chain = prompt | self.llm
response = chain.invoke({
"stats": "\n".join([f"{k}: {v}" for k, v in stats.items()])
})
return str(response.content)
def get_recent_events(self, limit: int = 10) -> List[MonitorEvent]:
"""Get recent events from all monitors."""
all_events = []
for monitor in self.monitors.values():
all_events.extend(monitor.events)
return sorted(all_events, key=lambda e: e.timestamp, reverse=True)[:limit]
def main():
"""Example usage of Monitor Agent Pattern."""
print("=" * 80)
print("Monitor Agent Pattern - System Monitoring Demo")
print("=" * 80)
# Create monitoring pattern
pattern = MonitorAgentPattern()
# Simulate monitored systems
import random
cpu_usage = [20.0]
memory_usage = [50.0]
def check_cpu():
cpu_usage[0] += random.uniform(-5, 10)
cpu_usage[0] = max(0, min(100, cpu_usage[0]))
return {"cpu": cpu_usage[0], "cores": 4}
def check_memory():
memory_usage[0] += random.uniform(-3, 8)
memory_usage[0] = max(0, min(100, memory_usage[0]))
return {"memory": memory_usage[0], "total_gb": 16}
def check_disk():
return {"used_gb": random.randint(100, 200), "total_gb": 500}
# Create CPU monitor
cpu_monitor = MonitorAgent(
monitor_id="cpu-monitor",
name="CPU Monitor",
check_function=check_cpu,
check_interval=2
)
cpu_monitor.add_rule(MonitorRule(
rule_id="cpu-high",
name="High CPU Usage",
condition=lambda data: data["cpu"] > 70,
action=lambda data: print(f"ā ļø CPU at {data['cpu']:.1f}%"),
priority=3
))
cpu_monitor.add_rule(MonitorRule(
rule_id="cpu-critical",
name="Critical CPU Usage",
condition=lambda data: data["cpu"] > 90,
action=lambda data: print(f"šØ CRITICAL: CPU at {data['cpu']:.1f}%"),
priority=5
))
# Create Memory monitor
memory_monitor = MonitorAgent(
monitor_id="mem-monitor",
name="Memory Monitor",
check_function=check_memory,
check_interval=2
)
memory_monitor.add_rule(MonitorRule(
rule_id="mem-high",
name="High Memory Usage",
condition=lambda data: data["memory"] > 75,
action=lambda data: print(f"ā ļø Memory at {data['memory']:.1f}%"),
priority=3
))
# Create Disk monitor
disk_monitor = MonitorAgent(
monitor_id="disk-monitor",
name="Disk Monitor",
check_function=check_disk,
check_interval=3
)
disk_monitor.add_rule(MonitorRule(
rule_id="disk-full",
name="Disk Almost Full",
condition=lambda data: (data["used_gb"] / data["total_gb"]) > 0.8,
action=lambda data: print(f"ā ļø Disk at {(data['used_gb']/data['total_gb'])*100:.1f}%"),
priority=4
))
# Register monitors
pattern.register_monitor(cpu_monitor)
pattern.register_monitor(memory_monitor)
pattern.register_monitor(disk_monitor)
# Perform manual checks
print("\nš Manual Check Results:")
results = pattern.check_all()
for monitor_id, result in results.items():
print(f"\n{monitor_id}:")
print(f" Status: {result['status']}")
print(f" Data: {result['data']}")
print(f" Message: {result['message']}")
# Start continuous monitoring
print("\n" + "=" * 80)
print("š Starting continuous monitoring for 10 seconds...")
print("=" * 80)
pattern.start_all()
time.sleep(10)
pattern.stop_all()
# Show statistics
print("\n" + "=" * 80)
print("š Monitoring Statistics")
print("=" * 80)
for monitor_id, monitor in pattern.monitors.items():
stats = monitor.get_statistics()
print(f"\n{monitor_id}:")
for key, value in stats.items():
print(f" {key}: {value}")
# Show recent events
print("\n" + "=" * 80)
print("š Recent Events")
print("=" * 80)
events = pattern.get_recent_events(limit=10)
for event in events:
print(f"\n[{event.status.value.upper()}] {event.monitor_id}")
print(f" Rule: {event.rule_id}")
print(f" Message: {event.message}")
print(f" Time: {event.timestamp.strftime('%H:%M:%S')}")
# Get system status
print("\n" + "=" * 80)
print("š„ļø System Status Analysis")
print("=" * 80)
status = pattern.get_system_status()
print(status)
if __name__ == "__main__":
main()
Agentic Architecture Pattern: Monitor Agent
Pattern 102: 102 Monitor Agent Pattern – Line-by-Line Explanation
Overview
Pattern 102: Monitor Agent Pattern Continuous monitoring and alerting based on thresholds.
Code Structure Analysis
File Information
- Pattern Number: 102
- Pattern Name: 102 Monitor Agent Pattern
- File Path:
patterns\perception_sensing\102_monitor_agent_pattern.py - Total Lines: 224
Detailed Code Explanation
Module Docstring
"""
Pattern 102: Monitor Agent Pattern
Continuous monitoring and alerting based on thresholds.
"""
Explanation: This module docstring describes the purpose and key features of the 102 Monitor Agent Pattern 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_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, 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
AlertLevel (Line 14)
class AlertLevel(Enum):
INFO
WARNING
CRITICAL
Purpose: Alert severity levels.
Values:
– INFO: Informational severity level
– WARNING: Warning severity level
– CRITICAL: Represents critical
Classes
AlertLevel (Line 14)
Description: Alert severity levels.
MonitoringMetric (Line 22)
Description: A metric to monitor.
Methods:
get_alert_level (Line 31)
def get_alert_level(self):
"""Get alert level based on thresholds."""
Purpose: Get alert level based on thresholds.
Parameters:
– self: Instance reference
Alert (Line 41)
Description: An alert generated by monitoring.
MonitorAgent (Line 51)
Description: Agent that monitors metrics and generates alerts.
Methods:
__init__ (Line 54)
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_metric (Line 65)
def register_metric(self, metric):
"""Register a metric to monitor."""
Purpose: Register a metric to monitor.
Parameters:
– self: Instance reference
– metric: Parameter for metric
update_metric (Line 69)
def update_metric(self, name, value):
"""Update a metric value and check for alerts."""
Purpose: Update a metric value and check for alerts.
Parameters:
– self: Instance reference
– name: Identifier or label
– value: Parameter for value
_generate_alert (Line 86)
def _generate_alert(self, metric, level):
"""Generate an alert with LLM-enhanced message."""
Purpose: Generate an alert with LLM-enhanced message.
Parameters:
– self: Instance reference
– metric: Parameter for metric
– level: Parameter for level
register_handler (Line 112)
def register_handler(self, level, handler):
"""Register a handler for specific alert level."""
Purpose: Register a handler for specific alert level.
Parameters:
– self: Instance reference
– level: Parameter for level
– handler: Parameter for handler
_handle_alert (Line 116)
def _handle_alert(self, alert):
"""Execute handlers for an alert."""
Purpose: Execute handlers for an alert.
Parameters:
– self: Instance reference
– alert: Parameter for alert
get_status (Line 121)
def get_status(self):
"""Get monitoring status."""
Purpose: Get monitoring status.
Parameters:
– self: Instance reference
MonitorAgentPattern (Line 134)
Description: Monitor Agent Pattern – Continuous monitoring and alerting.
Methods:
__init__ (Line 137)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
create_agent (Line 140)
def create_agent(self, name):
"""Create a monitoring agent."""
Purpose: Create a monitoring agent.
Parameters:
– self: Instance reference
– name: Identifier or label
run (Line 146)
def run(self, scenario):
"""Execute monitoring scenario."""
Purpose: Execute monitoring scenario.
Parameters:
– self: Instance reference
– scenario: Parameter for scenario
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = AlertLevel(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
Leave a Reply