5.3 监控与维护

5.3 监控与维护 — AI知识库搭建全攻略

本节导读：建立完善的监控体系，实现系统状态的实时监控和故障预警，掌握AI知识库的日常维护和故障排除方法，确保系统长期稳定运行。

学习目标

• 掌握AI知识库监控系统的设计和实现
• 学会关键性能指标的监控和分析
• 理解故障预警和自愈机制的设计
• 掌握日常维护流程和最佳实践
• 能够快速定位和解决常见问题

监控体系架构

监控层次设计

建立多层次监控体系，覆盖系统各个层面：

• 基础设施层：服务器硬件、网络设备、存储系统
• 平台层：容器编排、数据库集群、缓存集群
• 应用层：知识库服务、API服务、用户界面
• 业务层：搜索性能、数据质量、用户体验

监控数据收集

Prometheus + Grafana 监控栈

# 监控配置管理
import yaml
from prometheus_client import Counter, Gauge, Histogram
import logging

class MonitoringConfig:
    def __init__(self):
        self.prometheus_config = self._get_prometheus_config()
        
    def _get_prometheus_config(self):
        """获取Prometheus配置"""
        return {
            'global': {
                'scrape_interval': '15s',
                'evaluation_interval': '15s'
            },
            'scrape_configs': [
                {
                    'job_name': 'knowledge-base',
                    'static_configs': [
                        {
                            'targets': ['localhost:8080', 'localhost:8081', 'localhost:8082']
                        }
                    ]
                },
                {
                    'job_name': 'milvus',
                    'static_configs': [
                        {
                            'targets': ['localhost:19530']
                        }
                    ]
                },
                {
                    'job_name': 'redis',
                    'static_configs': [
                        {
                            'targets': ['localhost:6379']
                        }
                    ]
                }
            ]
        }
​

关键监控指标

系统指标监控

资源使用监控

import psutil
import time
import threading
from datetime import datetime

class SystemMonitor:
    def __init__(self, interval=30):
        self.interval = interval
        self.is_running = False
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.metrics_history = []
        
    def start(self):
        """启动监控"""
        self.is_running = True
        self.monitor_thread.start()
        
    def stop(self):
        """停止监控"""
        self.is_running = False
        self.monitor_thread.join()
        
    def _monitor_loop(self):
        """监控循环"""
        while self.is_running:
            try:
                metrics = self._collect_system_metrics()
                self._record_metrics(metrics)
                time.sleep(self.interval)
            except Exception as e:
                logging.error(f"监控收集失败: {e}")
                
    def _collect_system_metrics(self):
        """收集系统指标"""
        timestamp = datetime.now().isoformat()
        
        # CPU指标
        cpu_percent = psutil.cpu_percent(interval=1)
        cpu_count = psutil.cpu_count()
        
        # 内存指标
        memory = psutil.virtual_memory()
        
        # 磁盘指标
        disk_usage = psutil.disk_usage('/')
        
        metrics = {
            'timestamp': timestamp,
            'cpu': {
                'percent': cpu_percent,
                'count': cpu_count
            },
            'memory': {
                'total': memory.total,
                'used': memory.used,
                'percent': memory.percent,
                'available': memory.available
            },
            'disk': {
                'total': disk_usage.total,
                'used': disk_usage.used,
                'free': disk_usage.free,
                'percent': disk_usage.percent
            }
        }
        
        return metrics
        
    def get_metrics_summary(self, hours=24):
        """获取指标摘要"""
        current_time = datetime.now()
        start_time = current_time - timedelta(hours=hours)
        
        # 筛选指定时间范围内的指标
        filtered_metrics = [
            m for m in self.metrics_history
            if datetime.fromisoformat(m['timestamp']) >= start_time
        ]
        
        if not filtered_metrics:
            return {}
            
        # 计算统计信息
        cpu_percentages = [m['cpu']['percent'] for m in filtered_metrics]
        memory_percentages = [m['memory']['percent'] for m in filtered_metrics]
        disk_percentages = [m['disk']['percent'] for m in filtered_metrics]
        
        summary = {
            'time_range': {
                'start': start_time.isoformat(),
                'end': current_time.isoformat()
            },
            'cpu': {
                'avg': sum(cpu_percentages) / len(cpu_percentages),
                'max': max(cpu_percentages),
                'min': min(cpu_percentages)
            },
            'memory': {
                'avg': sum(memory_percentages) / len(memory_percentages),
                'max': max(memory_percentages),
                'min': min(memory_percentages)
            },
            'disk': {
                'avg': sum(disk_percentages) / len(disk_percentages),
                'max': max(disk_percentages),
                'min': min(disk_percentages)
            },
            'sample_count': len(filtered_metrics)
        }
        
        return summary
​

业务指标监控

搜索性能监控

import time
import threading
from collections import defaultdict, deque
import statistics

class SearchPerformanceMonitor:
    def __init__(self, window_size=1000):
        self.window_size = window_size
        self.search_times = deque(maxlen=window_size)
        self.search_types = defaultdict(list)
        self.error_counts = defaultdict(int)
        
    def record_search(self, search_type, duration, success=True):
        """记录搜索性能数据"""
        timestamp = time.time()
        search_data = {
            'timestamp': timestamp,
            'type': search_type,
            'duration': duration,
            'success': success
        }
        
        self.search_times.append(search_data)
        self.search_types[search_type].append(search_data)
        
        if not success:
            self.error_counts[search_type] += 1
            
    def get_performance_metrics(self):
        """获取性能指标"""
        recent_searches = list(self.search_times)[-100:]  # 最近100次搜索
        
        if not recent_searches:
            return {}
            
        # 计算基础指标
        durations = [s['duration'] for s in recent_searches if s['success']]
        
        metrics = {
            'total_searches': len(recent_searches),
            'successful_searches': len(durations),
            'failed_searches': len(recent_searches) - len(durations),
            'error_rate': (len(recent_searches) - len(durations)) / len(recent_searches),
            'avg_duration': statistics.mean(durations) if durations else 0,
            'p95_duration': statistics.quantiles(durations, n=20)[18] if durations else 0,
            'max_duration': max(durations) if durations else 0
        }
        
        return metrics
​

故障预警机制

告警规则配置

class AlertManager:
    def __init__(self, config):
        self.config = config
        self.alert_history = []
        self.alert_cooldown = {}  # 告警冷却机制
        
    def check_alert_conditions(self, metrics):
        """检查告警条件"""
        alerts = []
        
        # 检查系统指标
        if metrics.get('cpu', {}).get('percent', 0) > 90:
            alerts.append(self._create_alert(
                'CPU使用率过高',
                'cpu_high',
                f"CPU使用率达到{metrics['cpu']['percent']}%",
                'critical'
            ))
            
        if metrics.get('memory', {}).get('percent', 0) > 85:
            alerts.append(self._create_alert(
                '内存使用率过高',
                'memory_high',
                f"内存使用率达到{metrics['memory']['percent']}%",
                'critical'
            ))
            
        if metrics.get('disk', {}).get('percent', 0) > 95:
            alerts.append(self._create_alert(
                '磁盘空间不足',
                'disk_full',
                f"磁盘使用率达到{metrics['disk']['percent']}%",
                'critical'
            ))
            
        # 检查业务指标
        search_metrics = metrics.get('search', {})
        if search_metrics.get('error_rate', 0) > 0.05:  # 5%错误率
            alerts.append(self._create_alert(
                '搜索错误率过高',
                'search_error_rate',
                f"搜索错误率达到{search_metrics['error_rate']*100:.2f}%",
                'warning'
            ))
            
        # 发送告警
        for alert in alerts:
            self._send_alert(alert)
            
        return alerts
​

日常维护流程

运维管理规范

定期检查清单

import yaml
from datetime import datetime, timedelta

class MaintenanceManager:
    def __init__(self):
        self.maintenance_tasks = self._load_maintenance_tasks()
        self.task_history = []
        
    def _load_maintenance_tasks(self):
        """加载维护任务清单"""
        tasks = {
            'daily': [
                {
                    'id': 'check_system_health',
                    'name': '系统健康检查',
                    'description': '检查CPU、内存、磁盘使用率',
                    'command': 'system_health_check',
                    'schedule': '06:00',
                    'enabled': True
                },
                {
                    'id': 'backup_daily',
                    'name': '日常数据备份',
                    'description': '备份重要数据和配置',
                    'command': 'backup_data',
                    'schedule': '02:00',
                    'enabled': True
                },
                {
                    'id': 'clear_logs',
                    'name': '清理日志文件',
                    'description': '清理过期日志文件',
                    'command': 'clear_logs',
                    'schedule': '03:00',
                    'enabled': True
                }
            ],
            'weekly': [
                {
                    'id': 'performance_analysis',
                    'name': '性能分析',
                    'description': '分析系统性能趋势',
                    'command': 'analyze_performance',
                    'schedule': 'monday 10:00',
                    'enabled': True
                },
                {
                    'id': 'security_scan',
                    'name': '安全扫描',
                    'description': '系统安全漏洞扫描',
                    'command': 'security_scan',
                    'schedule': 'tuesday 14:00',
                    'enabled': True
                }
            ],
            'monthly': [
                {
                    'id': 'full_backup',
                    'name': '完整备份',
                    'description': '创建系统完整备份',
                    'command': 'full_backup',
                    'schedule': '1 02:00',
                    'enabled': True
                },
                {
                    'id': 'update_patches',
                    'name': '安全补丁更新',
                    'description': '安装安全补丁',
                    'command': 'update_patches',
                    'schedule': '15 10:00',
                    'enabled': True
                }
            ]
        }
        
        return tasks
​

配置管理

版本化配置管理

import yaml
import json
import os
import shutil
from datetime import datetime

class ConfigManager:
    def __init__(self, config_dir='/etc/knowledge-base'):
        self.config_dir = config_dir
        self.backup_dir = os.path.join(config_dir, 'backups')
        
        # 创建必要的目录
        os.makedirs(self.backup_dir, exist_ok=True)
        
    def load_config(self, config_name):
        """加载配置文件"""
        config_path = os.path.join(self.config_dir, f'{config_name}.yaml')
        
        if not os.path.exists(config_path):
            config_path = os.path.join(self.config_dir, f'{config_name}.json')
            
        if not os.path.exists(config_path):
            raise FileNotFoundError(f'配置文件不存在: {config_name}')
            
        with open(config_path, 'r', encoding='utf-8') as f:
            if config_path.endswith('.yaml'):
                return yaml.safe_load(f)
            else:
                return json.load(f)
                
    def save_config(self, config_name, config_data, backup=True):
        """保存配置文件"""
        # 确保是字典类型
        if not isinstance(config_data, dict):
            raise ValueError('配置数据必须是字典类型')
            
        # 备份当前配置
        if backup and os.path.exists(os.path.join(self.config_dir, f'{config_name}.yaml')):
            self._backup_config(config_name)
            
        # 保存新配置
        config_path = os.path.join(self.config_dir, f'{config_name}.yaml')
        with open(config_path, 'w', encoding='utf-8') as f:
            yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True)
            
    def _backup_config(self, config_name):
        """备份配置文件"""
        source_path = os.path.join(self.config_dir, f'{config_name}.yaml')
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_path = os.path.join(self.backup_dir, f'{config_name}_{timestamp}.yaml')
        
        shutil.copy2(source_path, backup_path)
​

故障排除指南

常见问题诊断

向量检索问题

class Troubleshooter:
    def __init__(self, vector_store):
        self.vector_store = vector_store
        
    def diagnose_vector_search_issues(self, symptoms):
        """诊断向量检索问题"""
        diagnosis = {
            'symptoms': symptoms,
            'possible_causes': [],
            'recommendations': []
        }
        
        # 根据症状分析可能原因
        if 'slow_search' in symptoms:
            diagnosis['possible_causes'].extend([
                '索引损坏',
                '内存不足',
                '网络延迟',
                '配置不当'
            ])
            diagnosis['recommendations'].extend([
                '重建索引',
                '增加内存',
                '检查网络连接',
                '优化搜索参数'
            ])
            
        if 'low_accuracy' in symptoms:
            diagnosis['possible_causes'].extend([
                '向量质量差',
                '索引类型不合适',
                '参数设置错误',
                '数据预处理不当'
            ])
            diagnosis['recommendations'].extend([
                '优化嵌入模型',
                '更改索引类型',
                '调整相似度阈值',
                '改进数据预处理'
            ])
            
        if 'high_error_rate' in symptoms:
            diagnosis['possible_causes'].extend([
                '连接超时',
                '服务崩溃',
                '权限问题',
                '资源不足'
            ])
            diagnosis['recommendations'].extend([
                '检查服务状态',
                '调整超时设置',
                '验证权限配置',
                '扩展资源'
            ])
            
        return diagnosis
​

系统性能问题

    def diagnose_performance_issues(self, symptoms):
        """诊断系统性能问题"""
        diagnosis = {
            'symptoms': symptoms,
            'possible_causes': [],
            'recommendations': []
        }
        
        if 'high_cpu_usage' in symptoms:
            diagnosis['possible_causes'].extend([
                '查询负载过高',
                '索引重建',
                '垃圾回收',
                '算法复杂度过高'
            ])
            diagnosis['recommendations'].extend([
                '优化查询算法',
                '分批处理索引重建',
                '调整JVM参数',
                '使用更高效的算法'
            ])
            
        if 'high_memory_usage' in symptoms:
            diagnosis['possible_causes'].extend([
                '缓存配置不当',
                '内存泄漏',
                '数据量过大',
                '并发过高'
            ])
            diagnosis['recommendations'].extend([
                '优化缓存策略',
                '检查内存泄漏',
                '分片处理数据',
                '控制并发数'
            ])
            
        if 'slow_response' in symptoms:
            diagnosis['possible_causes'].extend([
                '数据库查询慢',
                '网络延迟',
                '磁盘IO瓶颈',
                '资源竞争'
            ])
            diagnosis['recommendations'].extend([
                '优化数据库索引',
                '优化网络配置',
                '使用SSD存储',
                '减少资源竞争'
            ])
            
        return diagnosis
​

本节小结

本节详细介绍了AI知识库的监控与维护策略：

1. 监控体系架构：建立了多层次监控体系，包括基础设施、平台、应用和业务层面的监控
2. 关键监控指标：详细介绍了系统指标和业务指标的设计与实现方法
3. 故障预警机制：实现了告警规则配置和自动恢复机制
4. 日常维护流程：包括定期检查清单、版本化配置管理和最佳实践
5. 故障排除指南：提供了向量检索问题和系统性能问题的诊断方法

通过完善的监控和维护体系，可以确保AI知识库系统长期稳定运行，及时发现和解决问题。

延伸阅读

• 监控系统最佳实践
• Prometheus监控手册
• 故障处理方法论
• 相关章节：本教程第5章其他小节

关键词：AI知识库搭建全攻略, 监控系统, 故障预警, 自动恢复, 日常维护, 故障排除
难度：高级
预计阅读：20分钟