Skip to main content

最佳实践

代码管理​

代码规范​

Python代码风格​

# 良好的命名和注释示例
class DialogueManager:
    """对话管理器，负责处理用户输入和生成回复
    
    Attributes:
        history (List[Dict]): 对话历史记录
        character (Character): 当前选择的角色
        max_history_length (int): 最大历史记录长度
    """
    
    def __init__(self, max_history_length: int = 100):
        self.history = []
        self.character = None
        self.max_history_length = max_history_length

Git提交规范​

# 提交信息格式
<type>(<scope>): <description>

# 示例
feat(dialogue): 添加方言处理功能
fix(character): 修复角色切换bug
docs(readme): 更新部署说明

项目结构​

推荐的目录结构​

project/
├── app/
│   ├── __init__.py
│   ├── core/
│   │   ├── dialogue.py
│   │   └── character.py
│   ├── utils/
│   │   ├── logger.py
│   │   └── helpers.py
│   └── api/
│       └── routes.py
├── tests/
│   ├── test_dialogue.py
│   └── test_character.py
├── docs/
│   └── api.md
└── scripts/
    └── deploy.sh

模块划分​

# app/core/dialogue.py
from typing import List, Dict, Optional
from .character import Character
from ..utils.logger import get_logger

logger = get_logger(__name__)

class DialogueCore:
    """核心对话处理模块"""
    pass

性能优化​

内存管理​

缓存优化​

from functools import lru_cache
from typing import Dict, Any

@lru_cache(maxsize=1000)
def get_character_profile(character_id: str) -> Dict[str, Any]:
    """获取角色配置信息（使用缓存）"""
    return load_character_profile(character_id)

def clear_expired_cache():
    """清理过期缓存"""
    get_character_profile.cache_clear()

内存监控​

import psutil
import os

def monitor_memory_usage():
    """监控内存使用情况"""
    process = psutil.Process(os.getpid())
    memory_info = process.memory_info()
    
    return {
        'rss': memory_info.rss / 1024 / 1024,  # MB
        'vms': memory_info.vms / 1024 / 1024,  # MB
        'percent': process.memory_percent()
    }

响应优化​

异步处理​

import asyncio
from typing import List

async def process_messages(messages: List[str]) -> List[str]:
    """并行处理多条消息"""
    async def process_single(msg: str) -> str:
        # 模拟耗时操作
        await asyncio.sleep(0.1)
        return f"Processed: {msg}"
    
    tasks = [process_single(msg) for msg in messages]
    return await asyncio.gather(*tasks)

批处理优化​

class BatchProcessor:
    def __init__(self, batch_size: int = 32):
        self.batch_size = batch_size
        self.batch = []
        
    async def add_item(self, item: Any):
        self.batch.append(item)
        
        if len(self.batch) >= self.batch_size:
            await self.process_batch()
            
    async def process_batch(self):
        if not self.batch:
            return
            
        # 批量处理
        results = await process_messages(self.batch)
        self.batch = []
        return results

安全实践​

数据安全​

敏感信息处理​

import os
from cryptography.fernet import Fernet

class SecureStorage:
    def __init__(self):
        self.key = os.getenv('ENCRYPTION_KEY')
        self.cipher_suite = Fernet(self.key)
        
    def encrypt_data(self, data: str) -> bytes:
        """加密敏感数据"""
        return self.cipher_suite.encrypt(data.encode())
        
    def decrypt_data(self, encrypted_data: bytes) -> str:
        """解密数据"""
        return self.cipher_suite.decrypt(encrypted_data).decode()

输入验证​

from pydantic import BaseModel, validator
from typing import Optional

class UserInput(BaseModel):
    message: str
    character_id: Optional[str]
    
    @validator('message')
    def validate_message(cls, v):
        if len(v) > 1000:
            raise ValueError('消息长度不能超过1000字符')
        if not v.strip():
            raise ValueError('消息不能为空')
        return v.strip()

错误处理​

全局异常处理​

from functools import wraps
from typing import Type, Union

def handle_exceptions(*exceptions: Type[Exception]):
    """异常处理装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except exceptions as e:
                logger.error(f"Error in {func.__name__}: {str(e)}")
                return {
                    'status': 'error',
                    'message': str(e),
                    'error_type': type(e).__name__
                }
        return wrapper
    return decorator

优雅降级​

class GracefulDegradation:
    """服务降级管理器"""
    
    def __init__(self):
        self.fallback_responses = {
            'timeout': "服务响应超时，请稍后重试",
            'overload': "系统繁忙，请稍后重试",
            'error': "服务暂时不可用，请稍后重试"
        }
    
    async def handle_failure(self, failure_type: str) -> str:
        """根据失败类型返回降级响应"""
        return self.fallback_responses.get(
            failure_type,
            "系统遇到问题，请稍后重试"
        )