5.3 性能调优（终）

步骤4：性能监控与调优

创建完整的性能监控系统：

# monitoring/performance.py
import asyncio
import time
import threading
import psutil
import json
import os
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, asdict
from collections import defaultdict, deque
import logging

logger = logging.getLogger(__name__)

@dataclass
class PerformanceMetrics:
    """性能指标"""
    timestamp: float
    cpu_usage: float
    memory_usage: float
    disk_usage: float
    network_io: Dict[str, float]
    process_count: int
    thread_count: int
    checkpoint_ops: int
    tool_calls: int
    response_time: float

class PerformanceMonitor:
    """性能监控器"""
    
    def __init__(self, max_history: int = 1000):
        self.max_history = max_history
        self.metrics_history = deque(maxlen=max_history)
        self.alert_thresholds = {
            'cpu_usage': 80.0,
            'memory_usage': 85.0,
            'disk_usage': 90.0,
            'response_time': 5.0,
            'checkpoint_time': 10.0
        }
        self.alert_callbacks = []
        self.monitoring_active = False
        self.monitoring_thread = None
        
    def add_alert_callback(self, callback):
        """添加告警回调"""
        self.alert_callbacks.append(callback)
    
    def start_monitoring(self, interval: float = 1.0):
        """开始监控"""
        if self.monitoring_active:
            return
        
        self.monitoring_active = True
        self.monitoring_thread = threading.Thread(
            target=self._monitoring_loop,
            args=(interval,),
            daemon=True
        )
        self.monitoring_thread.start()
        logger.info("Performance monitoring started")
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring_active = False
        if self.monitoring_thread:
            self.monitoring_thread.join()
        logger.info("Performance monitoring stopped")
    
    def _monitoring_loop(self, interval: float):
        """监控循环"""
        while self.monitoring_active:
            try:
                # 收集指标
                metrics = self._collect_metrics()
                self.metrics_history.append(metrics)
                
                # 检查告警
                self._check_alerts(metrics)
                
                # 等待间隔
                time.sleep(interval)
                
            except Exception as e:
                logger.error(f"Monitoring error: {str(e)}")
                time.sleep(interval)
    
    def _collect_metrics(self) -> PerformanceMetrics:
        """收集性能指标"""
        try:
            # CPU使用率
            cpu_usage = psutil.cpu_percent(interval=1)
            
            # 内存使用率
            memory = psutil.virtual_memory()
            memory_usage = memory.percent
            
            # 磁盘使用率
            disk = psutil.disk_usage('/')
            disk_usage = disk.percent
            
            # 网络I/O
            network = psutil.net_io_counters()
            network_io = {
                'bytes_sent': network.bytes_sent,
                'bytes_recv': network.bytes_recv,
                'packets_sent': network.packets_sent,
                'packets_recv': network.packets_recv
            }
            
            # 进程和线程数
            process_count = len(psutil.pids())
            thread_count = sum(p.num_threads() for p in psutil.process_iter(['num_threads']) if p.info['num_threads'])
            
            # 模拟检查点操作和工具调用
            checkpoint_ops = len([m for m in self.metrics_history if hasattr(m, 'checkpoint_ops')])
            tool_calls = len([m for m in self.metrics_history if hasattr(m, 'tool_calls')])
            
            # 响应时间（最近请求的平均响应时间）
            recent_metrics = list(self.metrics_history)[-10:]
            response_time = sum(m.response_time for m in recent_metrics) / len(recent_metrics) if recent_metrics else 0.0
            
            return PerformanceMetrics(
                timestamp=time.time(),
                cpu_usage=cpu_usage,
                memory_usage=memory_usage,
                disk_usage=disk_usage,
                network_io=network_io,
                process_count=process_count,
                thread_count=thread_count,
                checkpoint_ops=checkpoint_ops,
                tool_calls=tool_calls,
                response_time=response_time
            )
            
        except Exception as e:
            logger.error(f"Metrics collection failed: {str(e)}")
            # 返回默认值
            return PerformanceMetrics(
                timestamp=time.time(),
                cpu_usage=0.0,
                memory_usage=0.0,
                disk_usage=0.0,
                network_io={},
                process_count=0,
                thread_count=0,
                checkpoint_ops=0,
                tool_calls=0,
                response_time=0.0
            )
    
    def _check_alerts(self, metrics: PerformanceMetrics):
        """检查告警"""
        alerts = []
        
        # CPU使用率告警
        if metrics.cpu_usage > self.alert_thresholds['cpu_usage']:
            alerts.append({
                'type': 'cpu_usage',
                'value': metrics.cpu_usage,
                'threshold': self.alert_thresholds['cpu_usage'],
                'message': f'CPU usage is high: {metrics.cpu_usage:.1f}%'
            })
        
        # 内存使用率告警
        if metrics.memory_usage > self.alert_thresholds['memory_usage']:
            alerts.append({
                'type': 'memory_usage',
                'value': metrics.memory_usage,
                'threshold': self.alert_thresholds['memory_usage'],
                'message': f'Memory usage is high: {metrics.memory_usage:.1f}%'
            })
        
        # 磁盘使用率告警
        if metrics.disk_usage > self.alert_thresholds['disk_usage']:
            alerts.append({
                'type': 'disk_usage',
                'value': metrics.disk_usage,
                'threshold': self.alert_thresholds['disk_usage'],
                'message': f'Disk usage is high: {metrics.disk_usage:.1f}%'
            })
        
        # 响应时间告警
        if metrics.response_time > self.alert_thresholds['response_time']:
            alerts.append({
                'type': 'response_time',
                'value': metrics.response_time,
                'threshold': self.alert_thresholds['response_time'],
                'message': f'Response time is high: {metrics.response_time:.2f}s'
            })
        
        # 触发告警回调
        for alert in alerts:
            for callback in self.alert_callbacks:
                try:
                    callback(alert)
                except Exception as e:
                    logger.error(f"Alert callback failed: {str(e)}")
    
    def get_metrics(self, limit: int = None) -> List[Dict[str, Any]]:
        """获取指标历史"""
        if limit is None:
            return [asdict(m) for m in self.metrics_history]
        
        return [asdict(m) for m in list(self.metrics_history)[-limit:]]
    
    def get_average_metrics(self, time_window: int = 300) -> Dict[str, float]:
        """获取指定时间窗口内的平均指标"""
        current_time = time.time()
        window_start = current_time - time_window
        
        # 筛选时间窗口内的指标
        window_metrics = [
            m for m in self.metrics_history
            if m.timestamp >= window_start
        ]
        
        if not window_metrics:
            return {}
        
        # 计算平均值
        avg_metrics = {
            'cpu_usage': sum(m.cpu_usage for m in window_metrics) / len(window_metrics),
            'memory_usage': sum(m.memory_usage for m in window_metrics) / len(window_metrics),
            'disk_usage': sum(m.disk_usage for m in window_metrics) / len(window_metrics),
            'response_time': sum(m.response_time for m in window_metrics) / len(window_metrics),
            'checkpoint_ops': sum(m.checkpoint_ops for m in window_metrics) / len(window_metrics),
            'tool_calls': sum(m.tool_calls for m in window_metrics) / len(window_metrics)
        }
        
        return avg_metrics
    
    def export_metrics(self, filepath: str):
        """导出指标到文件"""
        try:
            metrics_data = [asdict(m) for m in self.metrics_history]
            with open(filepath, 'w') as f:
                json.dump(metrics_data, f, indent=2)
            logger.info(f"Metrics exported to {filepath}")
        except Exception as e:
            logger.error(f"Metrics export failed: {str(e)}")

# 性能分析器
class PerformanceAnalyzer:
    """性能分析器"""
    
    def __init__(self, monitor: PerformanceMonitor):
        self.monitor = monitor
        self.analysis_history = []
    
    def analyze_performance(self) -> Dict[str, Any]:
        """分析性能数据"""
        try:
            # 获取最近1小时的平均指标
            avg_metrics = self.monitor.get_average_metrics(3600)
            
            # 分析趋势
            trends = self._analyze_trends()
            
            # 识别瓶颈
            bottlenecks = self._identify_bottlenecks()
            
            # 生成建议
            recommendations = self._generate_recommendations(avg_metrics, trends, bottlenecks)
            
            analysis = {
                'timestamp': time.time(),
                'average_metrics': avg_metrics,
                'trends': trends,
                'bottlenecks': bottlenecks,
                'recommendations': recommendations
            }
            
            self.analysis_history.append(analysis)
            return analysis
            
        except Exception as e:
            logger.error(f"Performance analysis failed: {str(e)}")
            return {}
    
    def _analyze_trends(self) -> Dict[str, str]:
        """分析趋势"""
        if len(self.monitor.metrics_history) < 2:
            return {}
        
        # 计算最近的变化趋势
        recent_metrics = list(self.monitor.metrics_history)[-10:]
        earlier_metrics = list(self.monitor.metrics_history)[-20:-10]
        
        if not earlier_metrics:
            return {}
        
        trends = {}
        for metric in ['cpu_usage', 'memory_usage', 'disk_usage']:
            recent_avg = sum(getattr(m, metric) for m in recent_metrics) / len(recent_metrics)
            earlier_avg = sum(getattr(m, metric) for m in earlier_metrics) / len(earlier_metrics)
            
            change = recent_avg - earlier_avg
            if abs(change) < 1.0:
                trend = 'stable'
            elif change > 0:
                trend = 'increasing'
            else:
                trend = 'decreasing'
            
            trends[metric] = trend
        
        return trends
    
    def _identify_bottlenecks(self) -> List[Dict[str, Any]]:
        """识别性能瓶颈"""
        bottlenecks = []
        
        # 获取当前指标
        if self.monitor.metrics_history:
            current_metrics = self.monitor.metrics_history[-1]
            
            # 检查CPU瓶颈
            if current_metrics.cpu_usage > 80:
                bottlenecks.append({
                    'type': 'cpu',
                    'severity': 'high' if current_metrics.cpu_usage > 90 else 'medium',
                    'value': current_metrics.cpu_usage,
                    'description': 'CPU usage is high, consider scaling or optimizing CPU-intensive tasks'
                })
            
            # 检查内存瓶颈
            if current_metrics.memory_usage > 85:
                bottlenecks.append({
                    'type': 'memory',
                    'severity': 'high' if current_metrics.memory_usage > 95 else 'medium',
                    'value': current_metrics.memory_usage,
                    'description': 'Memory usage is high, consider memory optimization or increasing memory'
                })
            
            # 检查响应时间瓶颈
            if current_metrics.response_time > 5.0:
                bottlenecks.append({
                    'type': 'response_time',
                    'severity': 'high' if current_metrics.response_time > 10.0 else 'medium',
                    'value': current_metrics.response_time,
                    'description': 'Response time is high, consider optimizing the application or scaling resources'
                })
        
        return bottlenecks
    
    def _generate_recommendations(self, avg_metrics: Dict[str, float], 
                                trends: Dict[str, str], bottlenecks: List[Dict[str, Any]]) -> List[str]:
        """生成优化建议"""
        recommendations = []
        
        # 基于瓶颈生成建议
        for bottleneck in bottlenecks:
            if bottleneck['type'] == 'cpu':
                recommendations.append('增加CPU资源或优化CPU密集型任务')
            elif bottleneck['type'] == 'memory':
                recommendations.append('增加内存资源或优化内存使用')
            elif bottleneck['type'] == 'response_time':
                recommendations.append('优化应用程序性能或增加服务器资源')
        
        # 基于趋势生成建议
        if trends.get('cpu_usage') == 'increasing':
            recommendations.append('CPU使用率呈上升趋势，建议提前规划资源扩展')
        
        if trends.get('memory_usage') == 'increasing':
            recommendations.append('内存使用率呈上升趋势，建议检查内存泄漏')
        
        # 添加一般性建议
        if not recommendations:
            recommendations.append('性能良好，继续监控系统状态')
        
        return recommendations
​

完整示例

# monitoring/example.py
"""
性能监控应用示例
展示如何在实际LangGraph应用中应用性能监控策略
"""
import asyncio
import time
import logging
from monitoring.performance import PerformanceMonitor, PerformanceAnalyzer

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PerformanceMonitoringSystem:
    """性能监控系统"""
    
    def __init__(self):
        # 初始化性能监控
        self.performance_monitor = PerformanceMonitor(max_history=1000)
        self.performance_analyzer = PerformanceAnalyzer(self.performance_monitor)
        
        # 添加告警回调
        self.performance_monitor.add_alert_callback(self._handle_alert)
        
        # 启动性能监控
        self.performance_monitor.start_monitoring(interval=5.0)
        
        # 统计信息
        self.stats = {
            'monitoring_start_time': time.time(),
            'alerts_count': 0,
            'analysis_count': 0
        }
    
    async def start(self):
        """启动系统"""
        logger.info("Starting performance monitoring system...")
        
        # 启动性能分析任务
        await self._analysis_task()
    
    async def _analysis_task(self):
        """性能分析任务"""
        while True:
            try:
                # 每5分钟进行一次性能分析
                analysis = self.performance_analyzer.analyze_performance()
                if analysis:
                    self.stats['analysis_count'] += 1
                    logger.info(f"Performance analysis #{self.stats['analysis_count']}: {analysis['bottlenecks']}")
                
                # 等待5分钟
                await asyncio.sleep(300)
                
            except Exception as e:
                logger.error(f"Performance analysis failed: {str(e)}")
                await asyncio.sleep(300)
    
    def _handle_alert(self, alert):
        """处理告警"""
        self.stats['alerts_count'] += 1
        logger.warning(f"Performance Alert #{self.stats['alerts_count']}: {alert['message']}")
        
        # 根据告警类型采取行动
        if alert['type'] == 'cpu_usage':
            logger.info("建议: 增加CPU资源或优化CPU密集型任务")
        elif alert['type'] == 'memory_usage':
            logger.info("建议: 增加内存资源或优化内存使用")
        elif alert['type'] == 'response_time':
            logger.info("建议: 优化应用程序性能或增加服务器资源")
    
    def get_stats(self) -> Dict[str, Any]:
        """获取统计信息"""
        return {
            **self.stats,
            'monitoring_duration': time.time() - self.stats['monitoring_start_time'],
            'metrics_count': len(self.performance_monitor.metrics_history)
        }
    
    def export_metrics(self, filepath: str):
        """导出指标到文件"""
        try:
            self.performance_monitor.export_metrics(filepath)
            logger.info(f"Metrics exported to {filepath}")
        except Exception as e:
            logger.error(f"Metrics export failed: {str(e)}")
    
    def stop_monitoring(self):
        """停止监控"""
        self.performance_monitor.stop_monitoring()
        logger.info("Performance monitoring system stopped")

# 使用示例
async def main():
    """主函数"""
    system = PerformanceMonitoringSystem()
    
    try:
        # 启动系统
        await system.start()
        
        # 模拟一些工作负载
        for i in range(10):
            # 模拟CPU密集型工作
            start_time = time.time()
            while time.time() - start_time < 0.1:  # 100ms
                _ = sum(range(1000))
            
            # 模拟I/O密集型工作
            await asyncio.sleep(0.05)
            
            print(f"Completed iteration {i}")
        
        # 等待一段时间观察性能
        await asyncio.sleep(60)
        
    finally:
        system.stop_monitoring()
        # 导出性能数据
        system.export_metrics("/tmp/monitoring_metrics.json")
        print(f"Monitoring system stats: {system.get_stats()}")

if __name__ == "__main__":
    asyncio.run(main())
​

常见问题 FAQ

Q4：如何优化检查点性能？

A：检查点优化的关键点：

• 压缩存储：使用压缩算法减少存储空间
• 增量检查点：只保存变化的部分
• 异步写入：不阻塞主线程的I/O操作
• 清理旧检查点：定期清理不需要的检查点
• 缓存策略：合理设置检查点缓存大小

优化示例：

# 使用优化的检查点管理器
checkpoint_manager = OptimizedCheckpointManager(
    storage_path="/optimized_checkpoints",
    compression_enabled=True,
    compression_level=6
)

# 创建检查点
await checkpoint_manager.create_checkpoint("graph_1", state)
​

Q5：如何设置合理的告警阈值？

A：告警阈值的设置应该基于实际业务需求：

• CPU使用率：生产环境建议80%，测试环境70%
• 内存使用率：生产环境85%，测试环境80%
• 响应时间：根据SLA要求，通常5-10秒
• 错误率：生产环境1%，测试环境5%
• 队列长度：根据系统容量设置，通常100-1000

阈值设置示例：

# 设置告警阈值
performance_monitor.alert_thresholds = {
    'cpu_usage': 80.0,
    'memory_usage': 85.0,
    'response_time': 5.0,
    'error_rate': 0.01  # 1%
}
​

最佳实践与避坑

• 全面监控：确保所有关键组件都有性能监控
• 分层优化：从应用层、系统层、基础设施层分别优化
• 渐进优化：先识别瓶颈，再有针对性地优化
• 基准测试：优化前后进行基准测试，验证效果
• 持续监控：优化后继续监控，确保性能稳定
• 文档记录：记录优化过程和结果，便于后续参考
• 团队协作：DevOps团队和开发团队协作进行性能优化

本节小结

本节详细介绍了LangGraph应用的检查点优化和性能监控系统的构建方法。通过这些优化策略，可以显著提升LangGraph应用在生产环境中的性能表现。

关键要点：

1. 检查点优化：压缩存储、异步写入、定期清理等策略优化检查点性能
2. 性能监控：全面的性能监控和及时的告警机制确保系统稳定运行
3. 数据分析：通过趋势分析和瓶颈识别生成优化建议
4. 完整示例：提供了实际可用的优化代码示例，涵盖检查点系统和性能监控

至此，我们完成了LangGraph性能调优的完整学习路径，从内存管理到并发处理，从检查点优化到性能监控，构建了完整的性能优化体系。

关键词：性能优化, 检查点优化, 性能监控, LangGraph, 生产环境, 高可用
难度：高级
预计阅读：30 分钟