第19讲：数据持久化与状态管理

掌握 Skill 的数据持久化和状态管理技巧，实现用户会话保持、数据存储和历史记录管理。

一、为什么需要数据持久化

1.1 典型场景

场景	需求	解决方案
多轮对话	记住用户之前提供的信息	会话状态管理
用户偏好	记住用户的设置和偏好	用户配置存储
历史记录	查看之前的操作记录	历史数据存储
数据分析	统计 Skill 使用情况	日志和指标存储
断点续传	长时间任务的中断恢复	任务状态持久化

1.2 数据类型分类

Skill 数据类型
├── 会话数据（临时）
│   ├── 当前对话上下文
│   ├── 临时变量
│   └── 中间计算结果
├── 用户数据（持久）
│   ├── 用户配置
│   ├── 使用偏好
│   └── 个人数据
├── 全局数据（共享）
│   ├── Skill 配置
│   ├── 公共知识库
│   └── 统计数据
└── 日志数据（审计）
    ├── 操作日志
    ├── 错误日志
    └── 性能指标

二、会话状态管理

2.1 会话生命周期

会话生命周期
┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐
│  开始   │ → │  进行中  │ → │  暂停   │ → │  结束   │
└─────────┘    └─────────┘    └─────────┘    └─────────┘
     │              │              │              │
     ▼              ▼              ▼              ▼
  创建会话      更新状态       保存状态       清理资源
  初始化数据    记录交互       等待恢复       归档数据

2.2 会话管理实现

import time
from typing import Dict, Any, Optional
from dataclasses import dataclass, field

@dataclass
class Session:
    """会话对象"""
    session_id: str
    user_id: str
    created_at: float = field(default_factory=time.time)
    last_active: float = field(default_factory=time.time)
    context: Dict[str, Any] = field(default_factory=dict)
    history: list = field(default_factory=list)
    
    def update(self, key: str, value: Any):
        """更新会话数据"""
        self.context[key] = value
        self.last_active = time.time()
    
    def get(self, key: str, default=None) -> Any:
        """获取会话数据"""
        return self.context.get(key, default)
    
    def add_history(self, role: str, content: str):
        """添加对话历史"""
        self.history.append({
            'role': role,
            'content': content,
            'timestamp': time.time()
        })
        self.last_active = time.time()
    
    def is_expired(self, timeout: int = 3600) -> bool:
        """检查会话是否过期"""
        return time.time() - self.last_active > timeout

class SessionManager:
    """会话管理器"""
    
    def __init__(self, storage=None, timeout=3600):
        self.sessions: Dict[str, Session] = {}
        self.storage = storage  # 持久化存储
        self.timeout = timeout
    
    def create_session(self, user_id: str) -> Session:
        """创建新会话"""
        import uuid
        session_id = str(uuid.uuid4())
        
        session = Session(
            session_id=session_id,
            user_id=user_id
        )
        
        self.sessions[session_id] = session
        
        # 持久化
        if self.storage:
            self.storage.save(f"session:{session_id}", {
                'session_id': session_id,
                'user_id': user_id,
                'created_at': session.created_at,
                'context': {}
            })
        
        return session
    
    def get_session(self, session_id: str) -> Optional[Session]:
        """获取会话"""
        session = self.sessions.get(session_id)
        
        if session and session.is_expired(self.timeout):
            # 会话过期，清理
            self.close_session(session_id)
            return None
        
        # 从存储恢复
        if not session and self.storage:
            data = self.storage.load(f"session:{session_id}")
            if data:
                session = Session(
                    session_id=data['session_id'],
                    user_id=data['user_id'],
                    created_at=data['created_at'],
                    context=data.get('context', {})
                )
                self.sessions[session_id] = session
        
        return session
    
    def close_session(self, session_id: str):
        """关闭会话"""
        if session_id in self.sessions:
            session = self.sessions[session_id]
            
            # 归档历史记录
            if self.storage and session.history:
                self.storage.save(
                    f"history:{session_id}",
                    session.history
                )
            
            del self.sessions[session_id]
            
            # 清理存储
            if self.storage:
                self.storage.delete(f"session:{session_id}")
    
    def cleanup_expired(self):
        """清理过期会话"""
        expired = [
            sid for sid, session in self.sessions.items()
            if session.is_expired(self.timeout)
        ]
        for sid in expired:
            self.close_session(sid)

2.3 多轮对话示例

class MultiTurnSkill:
    """支持多轮对话的 Skill"""
    
    def __init__(self):
        self.session_manager = SessionManager()
        self.intent_handlers = {
            'book_flight': self.handle_book_flight,
            'query_order': self.handle_query_order
        }
    
    def process(self, user_input: str, session_id: str = None, user_id: str = None) -> Dict:
        """处理用户输入"""
        # 获取或创建会话
        session = self.get_or_create_session(session_id, user_id)
        
        # 记录用户输入
        session.add_history('user', user_input)
        
        # 检查是否有进行中的任务
        current_task = session.get('current_task')
        
        if current_task:
            # 继续之前的任务
            response = self.continue_task(session, user_input)
        else:
            # 识别新意图
            intent = self.classify_intent(user_input)
            handler = self.intent_handlers.get(intent)
            
            if handler:
                response = handler(session, user_input)
            else:
                response = "抱歉，我不理解您的意思。"
        
        # 记录助手回复
        session.add_history('assistant', response)
        
        return {
            'response': response,
            'session_id': session.session_id
        }
    
    def handle_book_flight(self, session: Session, user_input: str) -> str:
        """处理机票预订（多轮对话示例）"""
        step = session.get('booking_step', 'start')
        
        if step == 'start':
            session.update('current_task', 'book_flight')
            session.update('booking_step', 'ask_origin')
            return "请问您要从哪个城市出发？"
        
        elif step == 'ask_origin':
            session.update('origin', user_input)
            session.update('booking_step', 'ask_destination')
            return f"好的，从{user_input}出发。请问您要去哪个城市？"
        
        elif step == 'ask_destination':
            session.update('destination', user_input)
            session.update('booking_step', 'ask_date')
            origin = session.get('origin')
            return f"好的，从{origin}到{user_input}。请问您计划什么时候出发？"
        
        elif step == 'ask_date':
            session.update('date', user_input)
            # 完成信息收集，执行预订
            result = self.book_flight(
                session.get('origin'),
                session.get('destination'),
                user_input
            )
            # 清理任务状态
            session.update('current_task', None)
            session.update('booking_step', None)
            return f"预订成功！{result}"

三、用户数据存储

3.1 用户配置管理

class UserPreferenceManager:
    """用户偏好管理"""
    
    def __init__(self, storage):
        self.storage = storage
        self.default_preferences = {
            'language': 'zh-CN',
            'timezone': 'Asia/Shanghai',
            'notification': True,
            'theme': 'light',
            'excel_format': {
                'font': '微软雅黑',
                'font_size': 11,
                'header_color': '4472C4'
            }
        }
    
    def get_preferences(self, user_id: str) -> Dict:
        """获取用户偏好"""
        key = f"user:{user_id}:preferences"
        prefs = self.storage.load(key)
        
        if not prefs:
            prefs = self.default_preferences.copy()
            self.storage.save(key, prefs)
        
        return prefs
    
    def update_preference(self, user_id: str, key: str, value: Any):
        """更新用户偏好"""
        prefs = self.get_preferences(user_id)
        
        # 支持嵌套更新
        keys = key.split('.')
        target = prefs
        for k in keys[:-1]:
            if k not in target:
                target[k] = {}
            target = target[k]
        target[keys[-1]] = value
        
        self.storage.save(f"user:{user_id}:preferences", prefs)
    
    def reset_preferences(self, user_id: str):
        """重置用户偏好"""
        self.storage.save(
            f"user:{user_id}:preferences",
            self.default_preferences.copy()
        )

3.2 用户数据安全

import hashlib
from cryptography.fernet import Fernet

class SecureStorage:
    """安全存储"""
    
    def __init__(self, encryption_key: str):
        self.cipher = Fernet(encryption_key.encode())
    
    def encrypt(self, data: str) -> str:
        """加密数据"""
        return self.cipher.encrypt(data.encode()).decode()
    
    def decrypt(self, encrypted: str) -> str:
        """解密数据"""
        return self.cipher.decrypt(encrypted.encode()).decode()
    
    def hash_id(self, user_id: str) -> str:
        """哈希用户 ID"""
        return hashlib.sha256(user_id.encode()).hexdigest()[:16]
    
    def save_sensitive(self, user_id: str, key: str, data: str):
        """保存敏感数据"""
        hashed_id = self.hash_id(user_id)
        encrypted = self.encrypt(data)
        # 存储加密后的数据
        storage.save(f"sensitive:{hashed_id}:{key}", encrypted)
    
    def load_sensitive(self, user_id: str, key: str) -> str:
        """加载敏感数据"""
        hashed_id = self.hash_id(user_id)
        encrypted = storage.load(f"sensitive:{hashed_id}:{key}")
        return self.decrypt(encrypted) if encrypted else None

四、日志与监控

4.1 操作日志

import logging
import json
from datetime import datetime

class SkillLogger:
    """Skill 日志记录器"""
    
    def __init__(self, skill_name: str, storage=None):
        self.skill_name = skill_name
        self.storage = storage
        
        # 配置日志
        self.logger = logging.getLogger(skill_name)
        self.logger.setLevel(logging.INFO)
    
    def log_interaction(self, user_id: str, session_id: str, 
                       input_text: str, output_text: str, 
                       intent: str = None, duration_ms: int = None):
        """记录交互日志"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'skill': self.skill_name,
            'user_id': user_id,
            'session_id': session_id,
            'input': input_text[:1000],  # 限制长度
            'output': output_text[:1000],
            'intent': intent,
            'duration_ms': duration_ms
        }
        
        self.logger.info(json.dumps(log_entry, ensure_ascii=False))
        
        # 持久化存储
        if self.storage:
            self._save_to_storage(log_entry)
    
    def log_error(self, user_id: str, error: Exception, context: Dict = None):
        """记录错误日志"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'skill': self.skill_name,
            'user_id': user_id,
            'error_type': type(error).__name__,
            'error_message': str(error),
            'context': context
        }
        
        self.logger.error(json.dumps(log_entry, ensure_ascii=False))
    
    def _save_to_storage(self, log_entry: Dict):
        """保存到存储"""
        date = datetime.now().strftime('%Y-%m-%d')
        key = f"logs:{self.skill_name}:{date}"
        
        # 追加到当日日志
        existing = self.storage.load(key) or []
        existing.append(log_entry)
        self.storage.save(key, existing)

4.2 使用统计

class UsageAnalytics:
    """使用统计分析"""
    
    def __init__(self, storage):
        self.storage = storage
    
    def record_usage(self, skill_name: str, user_id: str, 
                    intent: str, success: bool):
        """记录使用情况"""
        today = datetime.now().strftime('%Y-%m-%d')
        
        # 日统计
        daily_key = f"stats:{skill_name}:daily:{today}"
        daily_stats = self.storage.load(daily_key) or {
            'total_calls': 0,
            'unique_users': set(),
            'intents': {},
            'success_count': 0
        }
        
        daily_stats['total_calls'] += 1
        daily_stats['unique_users'].add(user_id)
        daily_stats['intents'][intent] = daily_stats['intents'].get(intent, 0) + 1
        if success:
            daily_stats['success_count'] += 1
        
        self.storage.save(daily_key, daily_stats)
    
    def get_daily_report(self, skill_name: str, date: str = None) -> Dict:
        """获取日报"""
        if not date:
            date = datetime.now().strftime('%Y-%m-%d')
        
        key = f"stats:{skill_name}:daily:{date}"
        stats = self.storage.load(key)
        
        if not stats:
            return None
        
        return {
            'date': date,
            'total_calls': stats['total_calls'],
            'unique_users': len(stats['unique_users']),
            'success_rate': stats['success_count'] / stats['total_calls'],
            'intent_distribution': stats['intents']
        }