AI助手系统架构设计：从自然语言处理到个性化对话管理

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

1. 从阿罗娜的自我介绍看AI助手系统设计思路

最近在开发智能助手类项目时，我反复思考如何让AI助手拥有更自然的交互体验。阿罗娜的这句自我介绍虽然简短，却蕴含了AI助手系统设计的核心要素。本文将从技术角度完整拆解AI助手系统的架构设计、对话管理、个性化实现等关键模块，提供可落地的代码示例和工程实践。

无论你是想开发聊天机器人、虚拟助手，还是希望为现有系统添加智能交互能力，这套方案都能直接复用。我们将从基础概念开始，逐步深入到生产环境部署，涵盖自然语言处理、对话状态管理、个性化响应生成等核心技术点。

1.1 AI助手系统的基本架构

1.1.1 什么是AI助手系统

AI助手系统是一种基于人工智能技术的交互式软件，能够理解用户输入的自然语言，并给出智能响应。这类系统通常包含语音识别、自然语言理解、对话管理、响应生成等核心模块。

从技术架构角度看，一个完整的AI助手系统可以分为以下几个层次：

• 交互层：负责与用户进行输入输出交互，可以是文本聊天界面、语音接口等
• 理解层：将用户输入转换为机器可理解的结构化数据
• 决策层：根据用户意图和对话上下文决定如何响应
• 执行层：调用外部API或数据库获取信息，生成具体响应内容
• 个性化层：根据用户历史和行为数据提供定制化服务

1.1.2 阿罗娜自我介绍的技术解读

阿罗娜的自我介绍虽然只有一句话，但体现了优秀AI助手的设计理念：

# 阿罗娜自我介绍的技术要素分析 introduction_elements = { "身份声明": "我叫阿罗娜", "系统定位": "常驻在【什亭之箱】里的系统管理员兼主操作系统", "服务承诺": "以后也会作为助理帮助老师", "个性化特征": "友好的语气、明确的角色定位" }

这种设计模式在技术实现上需要考虑多个维度的协调，包括角色一致性、对话连贯性、个性化表达等。

2. 环境准备与开发工具

2.1 开发环境要求

在开始构建AI助手系统前，需要准备以下开发环境：

操作系统: Windows 10/11, macOS 10.14+, Ubuntu 18.04+Python版本: 3.8+（推荐3.9或3.10）核心依赖库:

• transformers >= 4.20.0
• torch >= 1.12.0
• numpy >= 1.21.0
• flask >= 2.0.0（用于Web接口）
• sqlalchemy >= 1.4.0（用于数据持久化）

2.2 项目结构规划

建议采用模块化的项目结构，便于维护和扩展：

ai_assistant/ ├── src/ │ ├── nlp/ # 自然语言处理模块 │ ├── dialog/ # 对话管理模块 │ ├── personality/ # 个性化模块 │ ├── api/ # 接口层 │ └── utils/ # 工具函数 ├── tests/ # 测试代码 ├── config/ # 配置文件 ├── data/ # 训练数据和资源 └── requirements.txt # 依赖列表

2.3 初始化项目环境

创建并激活虚拟环境：

# 创建虚拟环境 python -m venv ai_assistant_env source ai_assistant_env/bin/activate # Linux/macOS # 或 ai_assistant_env\Scripts\activate # Windows # 安装基础依赖 pip install torch transformers flask sqlalchemy

创建requirements.txt文件记录完整依赖：

torch>=1.12.0 transformers>=4.20.0 flask>=2.0.0 sqlalchemy>=1.4.0 numpy>=1.21.0 requests>=2.25.0 python-dotenv>=0.19.0

3. 自然语言理解模块实现

3.1 意图识别技术选型

意图识别是AI助手理解用户需求的关键技术。我们使用基于Transformer的预训练模型来实现高效的意图分类。

# src/nlp/intent_classifier.py import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification from typing import Dict, List class IntentClassifier: def __init__(self, model_name: str = "bert-base-uncased"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) self.intent_labels = { 0: "问候", 1: "询问功能", 2: "寻求帮助", 3: "系统操作", 4: "个性化请求" } def predict_intent(self, text: str) -> Dict: """预测用户输入的意图""" inputs = self.tokenizer(text, return_tensors="pt", truncation=True, padding=True) with torch.no_grad(): outputs = self.model(**inputs) predictions = torch.nn.functional.softmax(outputs.logits, dim=-1) predicted_class = torch.argmax(predictions, dim=1).item() confidence = predictions[0][predicted_class].item() return { "intent": self.intent_labels[predicted_class], "confidence": confidence, "all_probabilities": { label: predictions[0][i].item() for i, label in self.intent_labels.items() } } # 使用示例 if __name__ == "__main__": classifier = IntentClassifier() result = classifier.predict_intent("你好，你能帮我做什么？") print(f"识别结果: {result}")

3.2 实体提取实现

实体提取用于识别用户输入中的关键信息，如人名、地点、时间等。

# src/nlp/entity_extractor.py import re from typing import List, Dict class EntityExtractor: def __init__(self): self.patterns = { "人名": r"[A-Za-z\u4e00-\u9fa5]{2,5}先生|[A-Za-z\u4e00-\u9fa5]{2,5}女士|[A-Za-z\u4e00-\u9fa5]{2,5}老师", "时间": r"\d{1,2}点|\d{1,2}月\d{1,2}日|今天|明天|下周", "地点": r"在.+?|到.+?", "系统操作": r"打开|关闭|重启|设置" } def extract_entities(self, text: str) -> List[Dict]: """从文本中提取实体信息""" entities = [] for entity_type, pattern in self.patterns.items(): matches = re.finditer(pattern, text) for match in matches: entities.append({ "type": entity_type, "value": match.group(), "start": match.start(), "end": match.end() }) return entities def extract_with_context(self, text: str) -> Dict: """带上下文的实体提取""" entities = self.extract_entities(text) return { "original_text": text, "entities": entities, "processed_text": self._remove_entities(text, entities) } def _remove_entities(self, text: str, entities: List[Dict]) -> str: """从文本中移除已识别的实体""" processed_text = text # 按起始位置倒序处理，避免索引变化 for entity in sorted(entities, key=lambda x: x["start"], reverse=True): processed_text = processed_text[:entity["start"]] + processed_text[entity["end"]:] return processed_text.strip()

4. 对话管理系统设计

4.1 对话状态跟踪

对话状态管理是维持对话连贯性的核心技术，需要跟踪用户意图、已提供信息、待补充信息等。

# src/dialog/dialog_state.py from datetime import datetime from typing import Dict, Any, List class DialogState: def __init__(self, user_id: str): self.user_id = user_id self.current_intent = None self.entities = [] self.history = [] self.context = {} self.created_at = datetime.now() self.last_updated = datetime.now() def update_state(self, intent: str, entities: List[Dict], user_input: str): """更新对话状态""" self.current_intent = intent self.entities = entities self.history.append({ "timestamp": datetime.now(), "user_input": user_input, "intent": intent, "entities": entities }) self.last_updated = datetime.now() # 更新上下文信息 self._update_context(entities) def _update_context(self, entities: List[Dict]): """根据实体更新上下文""" for entity in entities: if entity["type"] in ["人名", "时间", "地点"]: self.context[entity["type"]] = entity["value"] def get_missing_info(self, required_entities: List[str]) -> List[str]: """获取对话中缺失的必要信息""" missing = [] for entity_type in required_entities: if entity_type not in self.context: missing.append(entity_type) return missing def to_dict(self) -> Dict[str, Any]: """转换为字典格式，便于序列化""" return { "user_id": self.user_id, "current_intent": self.current_intent, "entities": self.entities, "context": self.context, "history_length": len(self.history), "created_at": self.created_at.isoformat(), "last_updated": self.last_updated.isoformat() } class DialogStateManager: def __init__(self): self.states = {} # user_id -> DialogState def get_state(self, user_id: str) -> DialogState: """获取用户的对话状态""" if user_id not in self.states: self.states[user_id] = DialogState(user_id) return self.states[user_id] def cleanup_old_states(self, max_age_hours: int = 24): """清理过期的对话状态""" current_time = datetime.now() expired_users = [] for user_id, state in self.states.items(): if (current_time - state.last_updated).total_seconds() > max_age_hours * 3600: expired_users.append(user_id) for user_id in expired_users: del self.states[user_id]

4.2 对话策略管理

基于规则和机器学习相结合的对话策略管理，确保响应既准确又自然。

# src/dialog/dialog_policy.py from typing import Dict, List, Optional class DialogPolicy: def __init__(self): self.response_templates = { "问候": { "模板": [ "你好！{name}很高兴为你服务。", "嗨！我是{name}，有什么可以帮你的吗？", "你好，我是{name}，随时为你提供帮助。" ], "参数": ["name"] }, "询问功能": { "模板": [ "我是{name}，可以帮你{capabilities}。", "作为{role}，我能够{capabilities}。", "我的主要功能包括{capabilities}，需要我具体介绍哪个呢？" ], "参数": ["name", "role", "capabilities"] }, "寻求帮助": { "模板": [ "当然可以！请告诉我你遇到的具体问题。", "我很乐意帮忙，能详细描述一下你的需求吗？", "没问题，我会尽力协助你解决。" ], "参数": [] } } def select_response(self, intent: str, context: Dict, personality_traits: Dict) -> str: """根据意图和上下文选择合适的响应模板""" if intent not in self.response_templates: return self._get_fallback_response() templates = self.response_templates[intent]["模板"] selected_template = self._select_best_template(templates, context, personality_traits) return self._fill_template(selected_template, context, personality_traits) def _select_best_template(self, templates: List[str], context: Dict, personality: Dict) -> str: """选择最合适的响应模板""" # 简单的基于上下文长度的选择策略 # 在实际项目中可以使用更复杂的机器学习方法 if len(context) > 2: return templates[2] # 选择更详细的模板 elif personality.get("verbosity", "normal") == "high": return templates[1] # 选择中等详细度的模板 else: return templates[0] # 选择最简洁的模板 def _fill_template(self, template: str, context: Dict, personality: Dict) -> str: """填充模板参数""" filled_template = template # 填充个性化参数 if "{name}" in template and "name" in personality: filled_template = filled_template.replace("{name}", personality["name"]) if "{role}" in template and "role" in personality: filled_template = filled_template.replace("{role}", personality["role"]) if "{capabilities}" in template and "capabilities" in personality: capabilities = "、".join(personality["capabilities"][:3]) # 限制显示数量 filled_template = filled_template.replace("{capabilities}", capabilities) return filled_template def _get_fallback_response(self) -> str: """获取回退响应""" return "我理解你的意思了，让我想想怎么更好地帮助你。"

5. 个性化系统实现

5.1 角色个性化配置

实现类似阿罗娜的个性化角色设定，包括名称、角色定位、服务承诺等要素。

# src/personality/character_config.py from dataclasses import dataclass from typing import List, Dict @dataclass class CharacterConfig: """角色配置数据类""" name: str role: str system_name: str service_commitment: str personality_traits: Dict[str, str] capabilities: List[str] greeting_template: str @classmethod def create_arona_config(cls): """创建阿罗娜风格的角色配置""" return cls( name="阿罗娜", role="系统管理员兼主操作系统", system_name="什亭之箱", service_commitment="作为助理帮助老师", personality_traits={ "语气": "友好热情", "专业度": "高", "幽默感": "适中", "详细程度": "详细" }, capabilities=[ "系统管理", "任务协助", "信息查询", "日程安排", "问题解答", "学习辅导" ], greeting_template="我叫{name}！是常驻在这个【{system_name}】里的{role}，以后也会{service_commitment}！" ) def get_greeting(self) -> str: """生成个性化的问候语""" return self.greeting_template.format( name=self.name, system_name=self.system_name, role=self.role, service_commitment=self.service_commitment ) def to_dict(self) -> Dict: """转换为字典格式""" return { "name": self.name, "role": self.role, "system_name": self.system_name, "service_commitment": self.service_commitment, "personality_traits": self.personality_traits, "capabilities": self.capabilities } class CharacterManager: """角色管理器，支持多角色切换""" def __init__(self): self.characters = {} self.current_character = None def register_character(self, name: str, config: CharacterConfig): """注册新角色""" self.characters[name] = config if self.current_character is None: self.current_character = name def switch_character(self, name: str) -> bool: """切换当前角色""" if name in self.characters: self.current_character = name return True return False def get_current_character(self) -> CharacterConfig: """获取当前角色配置""" if self.current_character in self.characters: return self.characters[self.current_character] return None def list_characters(self) -> List[str]: """列出所有可用角色""" return list(self.characters.keys())

5.2 用户偏好学习

基于用户交互历史学习个性化偏好，提供更精准的服务。

# src/personality/user_preference.py import json from datetime import datetime from typing import Dict, List, Any class UserPreference: def __init__(self, user_id: str): self.user_id = user_id self.interaction_history = [] self.preferred_topics = {} self.communication_style = "normal" # normal, formal, casual self.detail_level = "medium" # low, medium, high self.learning_rate = 0.1 # 偏好学习速率 def record_interaction(self, user_input: str, assistant_response: str, feedback: float = 0): """记录用户交互历史""" interaction = { "timestamp": datetime.now().isoformat(), "user_input": user_input, "assistant_response": assistant_response, "feedback": feedback, "topics": self._extract_topics(user_input) } self.interaction_history.append(interaction) # 基于交互更新偏好 self._update_preferences(interaction) def _extract_topics(self, text: str) -> List[str]: """从文本中提取主题关键词""" # 简化的主题提取，实际项目可以使用TF-IDF或主题模型 topics = [] topic_keywords = { "技术": ["代码", "编程", "技术", "系统", "软件"], "学习": ["学习", "教学", "教育", "课程", "知识"], "工作": ["工作", "任务", "项目", "会议", "日程"] } for topic, keywords in topic_keywords.items(): if any(keyword in text for keyword in keywords): topics.append(topic) return topics def _update_preferences(self, interaction: Dict): """基于交互更新用户偏好""" # 更新主题偏好 for topic in interaction["topics"]: if topic in self.preferred_topics: self.preferred_topics[topic] += self.learning_rate else: self.preferred_topics[topic] = self.learning_rate # 根据反馈调整沟通风格 if interaction["feedback"] > 0.7: # 正面反馈 # 强化当前风格 pass elif interaction["feedback"] < 0.3: # 负面反馈 # 考虑调整风格 pass def get_preferred_topics(self, top_n: int = 3) -> List[str]: """获取用户最感兴趣的主题""" sorted_topics = sorted(self.preferred_topics.items(), key=lambda x: x[1], reverse=True) return [topic for topic, score in sorted_topics[:top_n]] def to_dict(self) -> Dict[str, Any]: """转换为字典格式""" return { "user_id": self.user_id, "preferred_topics": self.preferred_topics, "communication_style": self.communication_style, "detail_level": self.detail_level, "interaction_count": len(self.interaction_history) } class PreferenceManager: """用户偏好管理器""" def __init__(self): self.user_preferences = {} # user_id -> UserPreference def get_preference(self, user_id: str) -> UserPreference: """获取用户偏好""" if user_id not in self.user_preferences: self.user_preferences[user_id] = UserPreference(user_id) return self.user_preferences[user_id] def save_preferences(self, filepath: str): """保存偏好数据到文件""" data = { user_id: pref.to_dict() for user_id, pref in self.user_preferences.items() } with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) def load_preferences(self, filepath: str): """从文件加载偏好数据""" try: with open(filepath, 'r', encoding='utf-8') as f: data = json.load(f) for user_id, pref_data in data.items(): pref = UserPreference(user_id) pref.preferred_topics = pref_data.get("preferred_topics", {}) pref.communication_style = pref_data.get("communication_style", "normal") pref.detail_level = pref_data.get("detail_level", "medium") self.user_preferences[user_id] = pref except FileNotFoundError: print("偏好文件不存在，将创建新的偏好管理器")

6. 完整系统集成与API设计

6.1 核心助手类实现

将各个模块整合成完整的AI助手系统。

# src/core/ai_assistant.py from typing import Dict, Any import logging from ..nlp.intent_classifier import IntentClassifier from ..nlp.entity_extractor import EntityExtractor from ..dialog.dialog_state import DialogStateManager from ..dialog.dialog_policy import DialogPolicy from ..personality.character_config import CharacterManager, CharacterConfig from ..personality.user_preference import PreferenceManager class AIAssistant: def __init__(self, character_config: CharacterConfig = None): self.logger = logging.getLogger(__name__) # 初始化各个模块 self.intent_classifier = IntentClassifier() self.entity_extractor = EntityExtractor() self.dialog_state_manager = DialogStateManager() self.dialog_policy = DialogPolicy() self.character_manager = CharacterManager() self.preference_manager = PreferenceManager() # 设置默认角色 if character_config: self.character_manager.register_character("default", character_config) self.character_manager.switch_character("default") self.logger.info("AI助手系统初始化完成") def process_message(self, user_id: str, message: str) -> Dict[str, Any]: """处理用户消息并生成响应""" try: # 1. 自然语言理解 intent_result = self.intent_classifier.predict_intent(message) entities = self.entity_extractor.extract_entities(message) # 2. 更新对话状态 dialog_state = self.dialog_state_manager.get_state(user_id) dialog_state.update_state(intent_result["intent"], entities, message) # 3. 获取用户偏好 user_preference = self.preference_manager.get_preference(user_id) # 4. 获取当前角色配置 current_character = self.character_manager.get_current_character() # 5. 生成个性化响应 personality_traits = { "name": current_character.name, "role": current_character.role, "capabilities": current_character.capabilities, "verbosity": user_preference.detail_level } response_text = self.dialog_policy.select_response( intent_result["intent"], dialog_state.context, personality_traits ) # 6. 记录交互历史 user_preference.record_interaction(message, response_text) return { "success": True, "response": response_text, "intent": intent_result["intent"], "confidence": intent_result["confidence"], "entities": entities, "character": current_character.name } except Exception as e: self.logger.error(f"处理消息时出错: {e}") return { "success": False, "response": "抱歉，我遇到了一些问题，请稍后再试。", "error": str(e) } def get_system_info(self) -> Dict[str, Any]: """获取系统信息""" current_character = self.character_manager.get_current_character() return { "character": current_character.to_dict() if current_character else None, "active_users": len(self.dialog_state_manager.states), "total_interactions": sum( len(pref.interaction_history) for pref in self.preference_manager.user_preferences.values() ) } # 初始化助手实例 def create_default_assistant(): """创建默认配置的AI助手""" arona_config = CharacterConfig.create_arona_config() return AIAssistant(arona_config)

6.2 Web API接口实现

提供HTTP API接口，方便其他系统集成。

# src/api/app.py from flask import Flask, request, jsonify from core.ai_assistant import create_default_assistant import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) assistant = create_default_assistant() @app.route('/api/chat', methods=['POST']) def chat_endpoint(): """聊天接口""" try: data = request.get_json() # 验证必要参数 if not data or 'user_id' not in data or 'message' not in data: return jsonify({ "success": False, "error": "缺少必要参数: user_id 和 message" }), 400 user_id = data['user_id'] message = data['message'] # 处理消息 result = assistant.process_message(user_id, message) return jsonify(result) except Exception as e: logger.error(f"API调用出错: {e}") return jsonify({ "success": False, "error": "服务器内部错误" }), 500 @app.route('/api/system/info', methods=['GET']) def system_info(): """获取系统信息""" try: info = assistant.get_system_info() return jsonify({ "success": True, "data": info }) except Exception as e: logger.error(f"获取系统信息出错: {e}") return jsonify({ "success": False, "error": "获取系统信息失败" }), 500 @app.route('/health', methods=['GET']) def health_check(): """健康检查接口""" return jsonify({"status": "healthy", "service": "AI Assistant"}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True)

6.3 客户端使用示例

提供Python客户端代码，演示如何调用AI助手服务。

# examples/client_example.py import requests import json class AIClient: def __init__(self, base_url: str = "http://localhost:5000"): self.base_url = base_url self.session_id = f"user_{hash(str(id(self)))}" # 模拟用户ID def send_message(self, message: str) -> dict: """发送消息到AI助手""" payload = { "user_id": self.session_id, "message": message } try: response = requests.post( f"{self.base_url}/api/chat", json=payload, timeout=10 ) return response.json() except requests.exceptions.RequestException as e: return { "success": False, "error": f"请求失败: {e}" } def chat_loop(self): """交互式聊天循环""" print("AI助手聊天系统（输入'退出'结束对话）") print("=" * 50) while True: user_input = input("\n你: ").strip() if user_input.lower() in ['退出', 'exit', 'quit']: print("对话结束，再见！") break if not user_input: continue result = self.send_message(user_input) if result.get("success"): character = result.get("character", "助手") response = result.get("response", "抱歉，我无法理解你的意思。") print(f"{character}: {response}") else: print(f"系统错误: {result.get('error', '未知错误')}") if __name__ == "__main__": client = AIClient() client.chat_loop()

7. 部署与生产环境配置

7.1 Docker容器化部署

使用Docker简化部署流程，确保环境一致性。

# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ ./src/ COPY examples/ ./examples/ COPY config/ ./config/ # 创建非root用户 RUN useradd -m -u 1000 aiuser USER aiuser # 暴露端口 EXPOSE 5000 # 启动命令 CMD ["python", "src/api/app.py"]

对应的docker-compose配置文件：

# docker-compose.yml version: '3.8' services: ai-assistant: build: . ports: - "5000:5000" environment: - PYTHONPATH=/app/src - FLASK_ENV=production volumes: - ./data:/app/data restart: unless-stopped # 可以添加数据库等服务 # redis: # image: redis:alpine # ports: # - "6379:6379"

7.2 生产环境配置

生产环境需要特别注意安全性和性能配置。

# config/production.py import os class ProductionConfig: """生产环境配置""" # 基础配置 DEBUG = False TESTING = False # 安全配置 SECRET_KEY = os.getenv('SECRET_KEY', 'your-secret-key-here') # 性能配置 MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB最大请求大小 # 模型配置 MODEL_CACHE_SIZE = 1000 MODEL_LOAD_TIMEOUT = 30 # 日志配置 LOG_LEVEL = 'INFO' LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' # 数据库配置（如果使用） # SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL') # SQLALCHEMY_TRACK_MODIFICATIONS = False # 环境变量配置示例 """ # .env.production SECRET_KEY=your-production-secret-key DATABASE_URL=postgresql://user:pass@localhost/dbname FLASK_ENV=production PYTHONPATH=/app/src """

8. 性能优化与监控

8.1 响应时间优化

针对AI助手的性能瓶颈进行优化。

# src/utils/performance_optimizer.py import time from functools import wraps from typing import Any, Callable import logging logger = logging.getLogger(__name__) def timing_decorator(func: Callable) -> Callable: """执行时间统计装饰器""" @wraps(func) def wrapper(*args, **kwargs) -> Any: start_time = time.time() result = func(*args, **kwargs) end_time = time.time() execution_time = end_time - start_time logger.info(f"{func.__name__} 执行时间: {execution_time:.4f}秒") # 如果执行时间过长，记录警告 if execution_time > 1.0: # 1秒阈值 logger.warning(f"{func.__name__} 执行时间过长: {execution_time:.4f}秒") return result return wrapper class CacheManager: """简单的缓存管理""" def __init__(self, max_size: int = 1000): self.cache = {} self.max_size = max_size self.access_count = {} def get(self, key: str) -> Any: """获取缓存值""" if key in self.cache: self.access_count[key] = self.access_count.get(key, 0) + 1 return self.cache[key] return None def set(self, key: str, value: Any): """设置缓存值""" if len(self.cache) >= self.max_size: # 简单的LRU淘汰策略 least_used = min(self.access_count.items(), key=lambda x: x[1]) del self.cache[least_used[0]] del self.access_count[least_used[0]] self.cache[key] = value self.access_count[key] = 1 def clear(self): """清空缓存""" self.cache.clear() self.access_count.clear() # 在关键函数上使用性能监控 class OptimizedIntentClassifier(IntentClassifier): @timing_decorator def predict_intent(self, text: str) -> Dict: """带性能监控的意图预测""" return super().predict_intent(text)

8.2 资源监控告警

实现基本的系统监控功能。

# src/utils/monitoring.py import psutil import time from threading import Thread from typing import Dict, Any class SystemMonitor: """系统资源监控""" def __init__(self, check_interval: int = 60): self.check_interval = check_interval self.monitoring = False self.monitor_thread = None # 监控阈值 self.thresholds = { "cpu_percent": 80.0, "memory_percent": 85.0, "disk_percent": 90.0 } def get_system_stats(self) -> Dict[str, Any]: """获取系统统计信息""" return { "timestamp": time.time(), "cpu_percent": psutil.cpu_percent(interval=1), "memory_percent": psutil.virtual_memory().percent, "disk_percent": psutil.disk_usage('/').percent, "process_memory_mb": psutil.Process().memory_info().rss / 1024 / 1024 } def check_thresholds(self, stats: Dict[str, Any]) -> List[str]: """检查是否超过阈值""" alerts = [] for metric, threshold in self.thresholds.items(): if stats[metric] > threshold: alerts.append(f"{metric} 超过阈值: {stats[metric]:.1f}% > {threshold}%") return alerts def start_monitoring(self): """开始监控""" self.monitoring = True self.monitor_thread = Thread(target=self._monitor_loop) self.monitor_thread.daemon = True self.monitor_thread.start() def stop_monitoring(self): """停止监控""" self.monitoring = False if self.monitor_thread: self.monitor_thread.join(timeout=5 > 🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉[点击领海量免费额度](https://taotoken.net/models/detail/chat?modelId=deepseek-v4-pro&utm_source=tt_blog_mr)