""" Redis缓存管理服务 用于高速缓存对话上下文,减少MySQL查询 """ import json import time import asyncio from typing import Dict, List, Optional import logging logger = logging.getLogger(__name__) class RedisCacheService: """Redis缓存服务""" def __init__(self, redis_client): """ 初始化Redis缓存服务 Args: redis_client: redis.asyncio.Redis 实例 """ self.redis = redis_client # 缓存Key设计 self.KEY_USER_CONTEXT = "context:{user_id}" # 用户上下文 self.KEY_USER_MESSAGES = "messages:{user_id}" # 用户消息列表 self.KEY_USER_ROUND = "round:{user_id}" # 当前轮次 self.KEY_USER_SUMMARY = "summary:{user_id}:{end_round}" # 历史总结 # TTL设置 self.TTL_CONTEXT = 86400 # 24小时 self.TTL_MESSAGES = 86400 # 24小时 self.TTL_SUMMARY = 604800 # 7天 async def get_context(self, user_id: str) -> Optional[Dict]: """ 从Redis获取用户上下文(零延迟) Returns: { "summary": str | None, # 最新的一个总结(已包含所有历史信息) "recent_messages": List[Dict], "current_round": int, "total_tokens": int } """ try: key = self.KEY_USER_CONTEXT.format(user_id=user_id) cached = await self.redis.get(key) if cached: context = json.loads(cached) logger.info(f"⚡ Redis命中: user={user_id}, round={context.get('current_round')}") return context logger.info(f"⚠️ Redis未命中: user={user_id}") return None except Exception as e: logger.error(f"❌ Redis读取失败: {e}") return None async def set_context( self, user_id: str, context: Dict, ttl: int = None ) -> bool: """ 更新Redis上下文缓存(超快写入) Args: user_id: 用户ID context: 上下文数据 ttl: 过期时间(秒),默认24小时 """ try: key = self.KEY_USER_CONTEXT.format(user_id=user_id) ttl = ttl or self.TTL_CONTEXT await self.redis.setex( key, ttl, json.dumps(context, ensure_ascii=False) ) logger.info(f"✅ Redis更新: user={user_id}") return True except Exception as e: logger.error(f"❌ Redis写入失败: {e}") return False async def append_message( self, user_id: str, role: str, content: str, round_number: int ) -> bool: """ 追加消息到Redis(实时更新) 使用Redis List结构,支持高效追加和范围查询 """ try: key = self.KEY_USER_MESSAGES.format(user_id=user_id) message = { 'role': role, 'content': content, 'round_number': round_number, 'timestamp': time.time() } # 追加到List尾部 await self.redis.rpush(key, json.dumps(message, ensure_ascii=False)) # 设置过期时间 await self.redis.expire(key, self.TTL_MESSAGES) # 限制List长度(最多保留200条消息) await self.redis.ltrim(key, -200, -1) logger.debug(f"✅ Redis追加消息: user={user_id}, round={round_number}, role={role}") return True except Exception as e: logger.error(f"❌ Redis追加失败: {e}") return False async def get_recent_messages( self, user_id: str, count: int = 30 ) -> List[Dict]: """ 获取最近N条消息(从Redis List) Args: user_id: 用户ID count: 消息数量(默认30条,即15轮) """ try: key = self.KEY_USER_MESSAGES.format(user_id=user_id) # 获取最后N条消息 messages_raw = await self.redis.lrange(key, -count, -1) messages = [] for msg_bytes in messages_raw: try: # 解码并解析JSON msg_str = msg_bytes.decode('utf-8') if isinstance(msg_bytes, bytes) else msg_bytes messages.append(json.loads(msg_str)) except Exception as e: logger.warning(f"⚠️ 解析消息失败: {e}") continue logger.debug(f"⚡ Redis获取消息: user={user_id}, count={len(messages)}") return messages except Exception as e: logger.error(f"❌ Redis读取消息失败: {e}") return [] async def get_current_round(self, user_id: str) -> int: """获取当前轮次""" try: key = self.KEY_USER_ROUND.format(user_id=user_id) round_bytes = await self.redis.get(key) if round_bytes: round_str = round_bytes.decode('utf-8') if isinstance(round_bytes, bytes) else round_bytes return int(round_str) return 0 except Exception as e: logger.error(f"❌ Redis读取轮次失败: {e}") return 0 async def set_current_round(self, user_id: str, round_number: int) -> bool: """设置当前轮次""" try: key = self.KEY_USER_ROUND.format(user_id=user_id) await self.redis.setex(key, self.TTL_CONTEXT, str(round_number)) return True except Exception as e: logger.error(f"❌ Redis设置轮次失败: {e}") return False async def increment_round(self, user_id: str) -> int: """轮次自增""" try: key = self.KEY_USER_ROUND.format(user_id=user_id) new_round = await self.redis.incr(key) await self.redis.expire(key, self.TTL_CONTEXT) # 同时更新 user:{user_id}:meta 中的 current_round(保持一致性) meta_key = f"user:{user_id}:meta" await self.redis.hset(meta_key, 'current_round', new_round) await self.redis.expire(meta_key, self.TTL_CONTEXT) logger.debug(f"✅ Redis轮次自增: user={user_id}, new_round={new_round}") return new_round except Exception as e: logger.error(f"❌ Redis轮次自增失败: {e}") return 0 async def get_summary(self, user_id: str, end_round: int) -> Optional[str]: """获取历史总结""" try: key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round) summary_bytes = await self.redis.get(key) if summary_bytes: summary = summary_bytes.decode('utf-8') if isinstance(summary_bytes, bytes) else summary_bytes logger.info(f"⚡ Redis获取总结: user={user_id}, end_round={end_round}") return summary return None except Exception as e: logger.error(f"❌ Redis读取总结失败: {e}") return None async def set_summary( self, user_id: str, end_round: int, summary: str ) -> bool: """保存历史总结""" try: key = self.KEY_USER_SUMMARY.format(user_id=user_id, end_round=end_round) await self.redis.setex(key, self.TTL_SUMMARY, summary) logger.info(f"✅ Redis保存总结: user={user_id}, end_round={end_round}, 长度={len(summary)}字") return True except Exception as e: logger.error(f"❌ Redis保存总结失败: {e}") return False async def build_context_from_cache( self, user_id: str, current_round: int, db_manager=None ) -> Dict: """ 从Redis构建完整上下文(超快) 逻辑: - 如果 current_round <= 15: 返回所有消息 - 如果 current_round > 15: 返回所有历史总结 + 最近15轮 Args: user_id: 用户ID current_round: 当前轮次 db_manager: 数据库管理器(用于获取所有总结) """ try: if current_round <= 15: # 前15轮:直接返回所有消息 messages = await self.get_recent_messages(user_id, count=current_round * 2) return { "summary": None, "recent_messages": messages, "current_round": current_round, "total_tokens": self._estimate_tokens(messages=messages) } else: # 超过15轮:最新总结 + 最近15轮 # 🔥 修复:查找 end_round < current_round 的最新总结 # 因为总结是累积的,最新的总结包含了所有历史信息 summary_end_round = current_round - 1 logger.info(f"🔍 [Redis] 查询总结: user={user_id}, current_round={current_round}, summary_end_round={summary_end_round}") # 获取消息 messages = await self.get_recent_messages(user_id, count=30) # 15轮=30条消息 # 从数据库获取最新的一个总结(已包含所有历史信息) summary = None if db_manager: try: # 先查询所有总结,用于调试 all_summaries = await db_manager.execute_query( """ SELECT end_round, start_round, created_at FROM conversation_summaries WHERE user_id = %(user_id)s ORDER BY end_round DESC LIMIT 10 """, {"user_id": user_id} ) if all_summaries: logger.info(f"📊 [Redis] 数据库中的总结: user={user_id}, 总数={len(all_summaries)}, 列表={[(s['start_round'], s['end_round']) for s in all_summaries]}") else: logger.warning(f"⚠️ [Redis] 数据库中没有任何总结: user={user_id}") # 调试:查询数据库中所有不同的 user_id try: all_user_ids = await db_manager.execute_query( """ SELECT DISTINCT user_id, COUNT(*) as summary_count FROM conversation_summaries GROUP BY user_id LIMIT 10 """ ) if all_user_ids: logger.info(f"🔍 [Redis] 数据库中存在的其他 user_id: {[(u['user_id'], u['summary_count']) for u in all_user_ids]}") except Exception as e: logger.debug(f"查询所有 user_id 失败: {e}") # 查询符合条件的总结 summary_result = await db_manager.execute_query( """ SELECT summary_text, end_round FROM conversation_summaries WHERE user_id = %(user_id)s AND end_round <= %(up_to_round)s ORDER BY end_round DESC LIMIT 1 """, {"user_id": user_id, "up_to_round": summary_end_round} ) if summary_result: summary = summary_result[0]['summary_text'] end_round = summary_result[0]['end_round'] logger.info(f"✅ [Redis] 获取最新总结: user={user_id}, end_round={end_round}, summary_length={len(summary)}") else: logger.warning(f"⚠️ [Redis] 未找到总结: user={user_id}, up_to_round={summary_end_round}, 数据库中总结数={len(all_summaries) if all_summaries else 0}") except Exception as e: logger.warning(f"⚠️ 从数据库获取总结失败: {e}") import traceback logger.error(traceback.format_exc()) return { "summary": summary, "recent_messages": messages, "current_round": current_round, "total_tokens": self._estimate_tokens(summary=summary, messages=messages) } except Exception as e: logger.error(f"❌ 从Redis构建上下文失败: {e}") return { "summary": None, "recent_messages": [], "current_round": current_round, "total_tokens": 0 } def _estimate_tokens(self, summary: str = None, messages: List[Dict] = None) -> int: """估算token数量""" total = 0 if summary: total += len(summary) // 2 if messages: for msg in messages: total += len(msg.get('content', '')) // 2 return total async def invalidate_context(self, user_id: str) -> bool: """清除用户上下文缓存(强制刷新)""" try: key = self.KEY_USER_CONTEXT.format(user_id=user_id) await self.redis.delete(key) logger.info(f"🗑️ Redis清除缓存: user={user_id}") return True except Exception as e: logger.error(f"❌ Redis清除失败: {e}") return False async def sync_from_db(self, user_id: str, messages: List[Dict], current_round: int) -> bool: """ 从数据库同步数据到Redis(初始化缓存) Args: user_id: 用户ID messages: 消息列表 current_round: 当前轮次 """ try: # 清空旧数据 messages_key = self.KEY_USER_MESSAGES.format(user_id=user_id) await self.redis.delete(messages_key) # 批量写入消息 if messages: pipeline = self.redis.pipeline() for msg in messages: pipeline.rpush( messages_key, json.dumps(msg, ensure_ascii=False) ) pipeline.expire(messages_key, self.TTL_MESSAGES) await pipeline.execute() # 设置轮次 await self.set_current_round(user_id, current_round) logger.info(f"✅ Redis同步完成: user={user_id}, messages={len(messages)}, round={current_round}") return True except Exception as e: logger.error(f"❌ Redis同步失败: {e}") return False async def health_check(self) -> bool: """健康检查""" try: await self.redis.ping() return True except Exception as e: logger.error(f"❌ Redis健康检查失败: {e}") return False