
    """
Pattern 171: Monitoring Pattern with LangChain/LangGraph

Purpose: Track agent performance, health, and metrics in real-time with LangGraph integration.
"""

from typing import Dict, Any, List, Optional, TypedDict
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
from langgraph.graph import StateGraph, END
import statistics


class HealthStatus(Enum):
    """Health status levels."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"


class MonitoringStrategy(Enum):
    """Monitoring strategies."""
    REAL_TIME = "real_time"
    PERIODIC = "periodic"
    THRESHOLD_BASED = "threshold_based"
    ANOMALY_DETECTION = "anomaly_detection"


class AlertLevel(Enum):
    """Alert severity levels."""
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"


class ThresholdType(Enum):
    """Threshold comparison types."""
    GREATER_THAN = "greater_than"
    LESS_THAN = "less_than"
    EQUALS = "equals"
    RANGE = "range"


@dataclass
class Metric:
    """Performance metric."""
    name: str
    value: float
    unit: str
    tags: Dict[str, str] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class HealthCheck:
    """Health check result."""
    component: str
    status: HealthStatus
    message: str
    metrics: List[Metric] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Alert:
    """Monitoring alert."""
    alert_id: str
    level: AlertLevel
    component: str
    message: str
    metric_name: str
    metric_value: float
    threshold: float
    triggered_at: datetime = field(default_factory=datetime.now)
    acknowledged: bool = False


@dataclass
class Threshold:
    """Metric threshold configuration."""
    metric_name: str
    threshold_type: ThresholdType
    value: float
    alert_level: AlertLevel
    max_value: Optional[float] = None  # For range thresholds


@dataclass
class TimeSeriesData:
    """Time-series data for a metric."""
    metric_name: str
    values: List[float]
    timestamps: List[datetime]
    window_size: int = 10


@dataclass
class Dashboard:
    """Monitoring dashboard data."""
    components: Dict[str, HealthStatus]
    total_metrics: int
    active_alerts: int
    alert_distribution: Dict[AlertLevel, int]
    recent_anomalies: List[str]
    system_health: str


class MonitoringState(TypedDict):
    """State for monitoring workflow."""
    component: str
    metrics: List[Dict[str, Any]]
    health_status: str
    alerts: List[str]
    analysis: str
    recommendations: List[str]


class MonitoringEngine:
    """
    Core monitoring engine with LangChain/LangGraph integration.
    """
    
    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.3):
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.metrics: Dict[str, List[Metric]] = {}
        self.health_checks: Dict[str, HealthCheck] = {}
        self.alerts: Dict[str, Alert] = {}
        self.thresholds: Dict[str, Threshold] = {}
        self.time_series: Dict[str, TimeSeriesData] = {}
        self.monitoring_strategies: Dict[str, MonitoringStrategy] = {}
        self.anomaly_history: List[Dict[str, Any]] = []
        self.workflow = self._build_monitoring_workflow()
        
        # Statistics
        self.stats = {
            "total_metrics_collected": 0,
            "alerts_triggered": 0,
            "anomalies_detected": 0,
            "health_checks_performed": 0,
            "alerts_by_level": {level: 0 for level in AlertLevel},
            "components_monitored": set()
        }
    
    def _build_monitoring_workflow(self) -> StateGraph:
        """Build LangGraph workflow for intelligent monitoring."""
        workflow = StateGraph(MonitoringState)
        
        # Add nodes
        workflow.add_node("collect_metrics", self._collect_metrics_node)
        workflow.add_node("analyze_health", self._analyze_health_node)
        workflow.add_node("detect_anomalies", self._detect_anomalies_node)
        workflow.add_node("generate_alerts", self._generate_alerts_node)
        workflow.add_node("provide_recommendations", self._provide_recommendations_node)
        
        # Define edges
        workflow.set_entry_point("collect_metrics")
        workflow.add_edge("collect_metrics", "analyze_health")
        workflow.add_edge("analyze_health", "detect_anomalies")
        workflow.add_edge("detect_anomalies", "generate_alerts")
        workflow.add_edge("generate_alerts", "provide_recommendations")
        workflow.add_edge("provide_recommendations", END)
        
        return workflow.compile()
    
    def _collect_metrics_node(self, state: MonitoringState) -> MonitoringState:
        """Collect and organize metrics."""
        component = state["component"]
        
        # Get recent metrics for component
        metrics_data = []
        if component in self.metrics:
            recent_metrics = self.metrics[component][-10:]
            metrics_data = [
                {
                    "name": m.name,
                    "value": m.value,
                    "unit": m.unit,
                    "timestamp": m.timestamp.isoformat()
                }
                for m in recent_metrics
            ]
        
        state["metrics"] = metrics_data
        return state
    
    def _analyze_health_node(self, state: MonitoringState) -> MonitoringState:
        """Analyze component health using LLM."""
        metrics = state["metrics"]
        
        if not metrics:
            state["health_status"] = HealthStatus.HEALTHY.value
            state["analysis"] = "No metrics available for analysis"
            return state
        
        # Use LLM to analyze metrics
        prompt = ChatPromptTemplate.from_template(
            "Analyze the following metrics and determine system health:\n"
            "Metrics: {metrics}\n\n"
            "Provide a health status (HEALTHY, DEGRADED, or UNHEALTHY) and brief analysis."
        )
        
        chain = prompt | self.llm
        response = chain.invoke({"metrics": str(metrics)})
        analysis = response.content
        
        # Determine health status
        analysis_lower = analysis.lower()
        if "unhealthy" in analysis_lower:
            health_status = HealthStatus.UNHEALTHY.value
        elif "degraded" in analysis_lower:
            health_status = HealthStatus.DEGRADED.value
        else:
            health_status = HealthStatus.HEALTHY.value
        
        state["health_status"] = health_status
        state["analysis"] = analysis
        
        return state
    
    def _detect_anomalies_node(self, state: MonitoringState) -> MonitoringState:
        """Detect anomalies in metrics using statistical analysis."""
        component = state["component"]
        alerts = []
        
        # Check thresholds
        if component in self.metrics:
            recent_metrics = self.metrics[component][-5:]
            for metric in recent_metrics:
                if metric.name in self.thresholds:
                    threshold = self.thresholds[metric.name]
                    if self._check_threshold(metric.value, threshold):
                        alerts.append(
                            f"Threshold exceeded: {metric.name} = {metric.value} "
                            f"(threshold: {threshold.value})"
                        )
        
        # Statistical anomaly detection
        if component in self.time_series:
            ts_data = self.time_series[component]
            if len(ts_data.values) >= 5:
                mean = statistics.mean(ts_data.values)
                stdev = statistics.stdev(ts_data.values)
                latest = ts_data.values[-1]
                
                # Check if latest value is an outlier (> 2 standard deviations)
                if abs(latest - mean) > 2 * stdev:
                    alerts.append(
                        f"Anomaly detected: {ts_data.metric_name} = {latest:.2f} "
                        f"(mean: {mean:.2f}, stdev: {stdev:.2f})"
                    )
        
        state["alerts"] = alerts
        return state
    
    def _generate_alerts_node(self, state: MonitoringState) -> MonitoringState:
        """Generate and categorize alerts."""
        alerts = state.get("alerts", [])
        
        if alerts:
            # Use LLM to categorize alert severity
            prompt = ChatPromptTemplate.from_template(
                "Categorize these monitoring alerts by severity (INFO, WARNING, ERROR, CRITICAL):\n"
                "Alerts: {alerts}\n\n"
                "Provide severity level for each alert."
            )
            
            chain = prompt | self.llm
            response = chain.invoke({"alerts": str(alerts)})
            
            # Update alerts with LLM categorization
            categorized_alerts = []
            for alert in alerts:
                severity = "WARNING"  # Default
                if "critical" in response.content.lower():
                    severity = "CRITICAL"
                elif "error" in response.content.lower():
                    severity = "ERROR"
                categorized_alerts.append(f"[{severity}] {alert}")
            
            state["alerts"] = categorized_alerts
        
        return state
    
    def _provide_recommendations_node(self, state: MonitoringState) -> MonitoringState:
        """Provide recommendations based on analysis."""
        health_status = state.get("health_status", "HEALTHY")
        alerts = state.get("alerts", [])
        analysis = state.get("analysis", "")
        
        recommendations = []
        
        if health_status != HealthStatus.HEALTHY.value or alerts:
            # Use LLM to generate recommendations
            prompt = ChatPromptTemplate.from_template(
                "Based on the following system analysis, provide 3-5 actionable recommendations:\n"
                "Health Status: {health_status}\n"
                "Analysis: {analysis}\n"
                "Alerts: {alerts}\n\n"
                "Provide specific, actionable recommendations."
            )
            
            chain = prompt | self.llm
            response = chain.invoke({
                "health_status": health_status,
                "analysis": analysis,
                "alerts": str(alerts)
            })
            
            # Parse recommendations
            recommendations = [
                line.strip() 
                for line in response.content.split('\n') 
                if line.strip() and len(line.strip()) > 10
            ][:5]
        
        state["recommendations"] = recommendations
        return state
    
    def _check_threshold(self, value: float, threshold: Threshold) -> bool:
        """Check if value breaches threshold."""
        if threshold.threshold_type == ThresholdType.GREATER_THAN:
            return value > threshold.value
        elif threshold.threshold_type == ThresholdType.LESS_THAN:
            return value < threshold.value
        elif threshold.threshold_type == ThresholdType.EQUALS:
            return abs(value - threshold.value) < 0.01
        elif threshold.threshold_type == ThresholdType.RANGE:
            return not (threshold.value <= value  ts_data.window_size:
            ts_data.values = ts_data.values[-ts_data.window_size:]
            ts_data.timestamps = ts_data.timestamps[-ts_data.window_size:]
        
        # Update statistics
        self.stats["total_metrics_collected"] += 1
        self.stats["components_monitored"].add(component)
    
    def set_threshold(
        self,
        metric_name: str,
        value: float,
        threshold_type: ThresholdType = ThresholdType.GREATER_THAN,
        alert_level: AlertLevel = AlertLevel.WARNING,
        max_value: Optional[float] = None
    ):
        """Set threshold for a metric."""
        self.thresholds[metric_name] = Threshold(
            metric_name=metric_name,
            threshold_type=threshold_type,
            value=value,
            alert_level=alert_level,
            max_value=max_value
        )
    
    def check_health(self, component: str) -> HealthCheck:
        """Perform comprehensive health check using workflow."""
        initial_state: MonitoringState = {
            "component": component,
            "metrics": [],
            "health_status": HealthStatus.HEALTHY.value,
            "alerts": [],
            "analysis": "",
            "recommendations": []
        }
        
        # Run workflow
        result = self.workflow.invoke(initial_state)
        
        # Create health check
        status = HealthStatus(result["health_status"])
        health_check = HealthCheck(
            component=component,
            status=status,
            message=result["analysis"],
            metrics=self.metrics.get(component, [])[-5:]
        )
        
        self.health_checks[component] = health_check
        self.stats["health_checks_performed"] += 1
        
        # Create alerts if needed
        for alert_msg in result["alerts"]:
            alert = Alert(
                alert_id=f"alert_{len(self.alerts)}",
                level=AlertLevel.WARNING,
                component=component,
                message=alert_msg,
                metric_name=component,
                metric_value=0.0,
                threshold=0.0
            )
            self.alerts[alert.alert_id] = alert
            self.stats["alerts_triggered"] += 1
            self.stats["alerts_by_level"][AlertLevel.WARNING] += 1
        
        return health_check
    
    def get_metric_stats(self, component: str, metric_name: str) -> Dict[str, Any]:
        """Get statistics for a metric."""
        if component not in self.metrics:
            return {}
        
        values = [
            m.value for m in self.metrics[component]
            if m.name == metric_name
        ]
        
        if not values:
            return {}
        
        return {
            "count": len(values),
            "min": min(values),
            "max": max(values),
            "avg": statistics.mean(values),
            "median": statistics.median(values),
            "stdev": statistics.stdev(values) if len(values) > 1 else 0,
            "latest": values[-1]
        }
    
    def get_dashboard(self) -> Dashboard:
        """Get comprehensive monitoring dashboard."""
        components = {
            component: check.status
            for component, check in self.health_checks.items()
        }
        
        total_metrics = sum(len(metrics) for metrics in self.metrics.values())
        active_alerts = len([a for a in self.alerts.values() if not a.acknowledged])
        
        alert_distribution = {
            level: sum(
                1 for a in self.alerts.values()
                if a.level == level and not a.acknowledged
            )
            for level in AlertLevel
        }
        
        recent_anomalies = [
            a["message"] for a in self.anomaly_history[-5:]
        ]
        
        # Determine overall system health
        unhealthy_count = sum(
            1 for status in components.values()
            if status == HealthStatus.UNHEALTHY
        )
        degraded_count = sum(
            1 for status in components.values()
            if status == HealthStatus.DEGRADED
        )
        
        if unhealthy_count > 0:
            system_health = "UNHEALTHY"
        elif degraded_count > 0:
            system_health = "DEGRADED"
        else:
            system_health = "HEALTHY"
        
        return Dashboard(
            components=components,
            total_metrics=total_metrics,
            active_alerts=active_alerts,
            alert_distribution=alert_distribution,
            recent_anomalies=recent_anomalies,
            system_health=system_health
        )
    
    def acknowledge_alert(self, alert_id: str):
        """Acknowledge an alert."""
        if alert_id in self.alerts:
            self.alerts[alert_id].acknowledged = True
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get monitoring statistics."""
        return {
            "total_metrics_collected": self.stats["total_metrics_collected"],
            "alerts_triggered": self.stats["alerts_triggered"],
            "anomalies_detected": self.stats["anomalies_detected"],
            "health_checks_performed": self.stats["health_checks_performed"],
            "alerts_by_level": {
                level.value: count
                for level, count in self.stats["alerts_by_level"].items()
            },
            "components_monitored": len(self.stats["components_monitored"]),
            "active_alerts": len([a for a in self.alerts.values() if not a.acknowledged]),
            "thresholds_configured": len(self.thresholds)
        }


class MonitoringPattern:
    """
    Comprehensive monitoring pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.engine = MonitoringEngine(model_name=model_name)
    
    def monitor(
        self,
        component: str,
        metric_name: str,
        value: float,
        unit: str = ""
    ) -> Dict[str, Any]:
        """Monitor a component with automated analysis."""
        # Record metric
        self.engine.record_metric(component, metric_name, value, unit)
        
        # Perform health check
        health_check = self.engine.check_health(component)
        
        return {
            "component": component,
            "health_status": health_check.status.value,
            "message": health_check.message,
            "alerts": [
                alert.message
                for alert in self.engine.alerts.values()
                if alert.component == component and not alert.acknowledged
            ]
        }
    
    def configure_monitoring(
        self,
        metric_name: str,
        threshold: float,
        threshold_type: ThresholdType = ThresholdType.GREATER_THAN
    ):
        """Configure monitoring thresholds."""
        self.engine.set_threshold(metric_name, threshold, threshold_type)
    
    def get_dashboard(self) -> Dict[str, Any]:
        """Get monitoring dashboard."""
        dashboard = self.engine.get_dashboard()
        return {
            "system_health": dashboard.system_health,
            "components": {
                comp: status.value
                for comp, status in dashboard.components.items()
            },
            "total_metrics": dashboard.total_metrics,
            "active_alerts": dashboard.active_alerts,
            "recent_anomalies": dashboard.recent_anomalies
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get monitoring statistics."""
        return self.engine.get_statistics()


if __name__ == "__main__":
    print("=== Monitoring Pattern with LangChain/LangGraph Demo ===\n")
    
    # Note: This demo shows the structure. In production, you'd need:
    # - export OPENAI_API_KEY="your-key-here"
    # - Actual LLM calls would be made
    
    print("Scenario 1: Basic Monitoring")
    print("-" * 50)
    monitor = MonitoringPattern(model_name="gpt-4")
    
    # Configure thresholds
    monitor.configure_monitoring("cpu_usage", 80.0, ThresholdType.GREATER_THAN)
    monitor.configure_monitoring("memory_usage", 90.0, ThresholdType.GREATER_THAN)
    monitor.configure_monitoring("response_time", 1000.0, ThresholdType.GREATER_THAN)
    
    print("✓ Configured monitoring thresholds")
    
    # Monitor different components
    components = [
        ("api_server", "cpu_usage", 65.0, "%"),
        ("api_server", "memory_usage", 78.0, "%"),
        ("api_server", "response_time", 250.0, "ms"),
        ("database", "cpu_usage", 45.0, "%"),
        ("database", "query_time", 150.0, "ms"),
        ("cache", "hit_rate", 92.0, "%"),
    ]
    
    print("\nMonitoring components:")
    for component, metric, value, unit in components:
        result = monitor.monitor(component, metric, value, unit)
        print(f"  {component}.{metric}: {value}{unit} -> {result['health_status']}")
    
    print("\nScenario 2: Threshold Violations")
    print("-" * 50)
    
    # Simulate threshold violations
    violations = [
        ("api_server", "cpu_usage", 85.0, "%"),
        ("database", "cpu_usage", 92.0, "%"),
        ("api_server", "response_time", 1500.0, "ms"),
    ]
    
    print("Simulating threshold violations:")
    for component, metric, value, unit in violations:
        result = monitor.monitor(component, metric, value, unit)
        print(f"\n  {component}.{metric}: {value}{unit}")
        print(f"    Status: {result['health_status']}")
        print(f"    Message: {result['message'][:100]}...")
        if result['alerts']:
            print(f"    Alerts: {len(result['alerts'])} active")
    
    print("\nScenario 3: Monitoring Dashboard")
    print("-" * 50)
    
    dashboard = monitor.get_dashboard()
    print(f"System Health: {dashboard['system_health']}")
    print(f"Total Metrics: {dashboard['total_metrics']}")
    print(f"Active Alerts: {dashboard['active_alerts']}")
    
    print("\nComponent Health:")
    for component, status in dashboard['components'].items():
        print(f"  {component}: {status}")
    
    if dashboard['recent_anomalies']:
        print("\nRecent Anomalies:")
        for anomaly in dashboard['recent_anomalies']:
            print(f"  - {anomaly}")
    
    print("\nScenario 4: Statistics")
    print("-" * 50)
    
    stats = monitor.get_statistics()
    print(f"Total Metrics Collected: {stats['total_metrics_collected']}")
    print(f"Alerts Triggered: {stats['alerts_triggered']}")
    print(f"Health Checks Performed: {stats['health_checks_performed']}")
    print(f"Components Monitored: {stats['components_monitored']}")
    print(f"Thresholds Configured: {stats['thresholds_configured']}")
    
    print("\nAlerts by Level:")
    for level, count in stats['alerts_by_level'].items():
        print(f"  {level}: {count}")
    
    print("\n" + "=" * 50)
    print("Monitoring pattern demonstration complete!")
    print("=" * 50)

Agentic Architecture Pattern: Monitoring

Pattern 171: 171 Monitoring – Line-by-Line Explanation

Overview

Pattern 171: Monitoring Pattern with LangChain/LangGraph

Purpose: Track agent performance, health, and metrics in real-time with LangGraph integration.

Code Structure Analysis

File Information

Pattern Number: 171
Pattern Name: 171 Monitoring
File Path: patterns\quality_performance\171_monitoring.py
Total Lines: 677

Detailed Code Explanation

Module Docstring

"""
Pattern 171: Monitoring Pattern with LangChain/LangGraph

Purpose: Track agent performance, health, and metrics in real-time with LangGraph integration.
"""

Explanation: This module docstring describes the purpose and key features of the 171 Monitoring pattern.

Imports Section

from typing import Dict, Any, List, Optional, TypedDict
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
from langgraph.graph import StateGraph, END
import statistics  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, TypedDict: 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 – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration

Enumerations

HealthStatus (Line 17)

class HealthStatus(Enum):
    HEALTHY
    DEGRADED
    UNHEALTHY

Purpose: Health status levels.

Values: – HEALTHY: Represents healthy – DEGRADED: Represents degraded – UNHEALTHY: Represents unhealthy

MonitoringStrategy (Line 24)

class MonitoringStrategy(Enum):
    REAL_TIME
    PERIODIC
    THRESHOLD_BASED
    ANOMALY_DETECTION

Purpose: Monitoring strategies.

Values: – REAL_TIME: Represents real time – PERIODIC: Represents periodic – THRESHOLD_BASED: Represents threshold based – ANOMALY_DETECTION: Represents anomaly detection

AlertLevel (Line 32)

class AlertLevel(Enum):
    INFO
    WARNING
    ERROR
    CRITICAL

Purpose: Alert severity levels.

Values: – INFO: Informational severity level – WARNING: Warning severity level – ERROR: Error severity level – CRITICAL: Represents critical

ThresholdType (Line 40)

class ThresholdType(Enum):
    GREATER_THAN
    LESS_THAN
    EQUALS
    RANGE

Purpose: Threshold comparison types.

Values: – GREATER_THAN: Represents greater than – LESS_THAN: Represents less than – EQUALS: Represents equals – RANGE: Represents range

Classes

HealthStatus (Line 17)

Description: Health status levels.

MonitoringStrategy (Line 24)

Description: Monitoring strategies.

AlertLevel (Line 32)

Description: Alert severity levels.

ThresholdType (Line 40)

Description: Threshold comparison types.

Metric (Line 49)

Description: Performance metric.

HealthCheck (Line 59)

Description: Health check result.

Alert (Line 69)

Description: Monitoring alert.

Threshold (Line 83)

Description: Metric threshold configuration.

TimeSeriesData (Line 93)

Description: Time-series data for a metric.

Dashboard (Line 102)

Description: Monitoring dashboard data.

MonitoringState (Line 112)

Description: State for monitoring workflow.

MonitoringEngine (Line 122)

Description: Core monitoring engine with LangChain/LangGraph integration.

Methods:

`init` (Line 127)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”) – temperature: Randomness in LLM responses (0.0-1.0)

`_build_monitoring_workflow` (Line 148)

def _build_monitoring_workflow(self):
    """Build LangGraph workflow for intelligent monitoring."""

Purpose: Build LangGraph workflow for intelligent monitoring.

Parameters: – self: Instance reference