""" 对话总结服务（SQLAlchemy ORM 版本） """ import uuid from typing import Awaitable, Callable, List, Dict, Optional import logging from sqlalchemy import select import redis.asyncio as aioredis logger = logging.getLogger(__name__) class SummaryService: """对话总结服务""" def __init__(self, db_manager): """ 初始化总结服务 Args: db_manager: 数据库管理器实例 """ self.db = db_manager self.redis = None # 🔥 延迟获取Redis self._redis_provider: Optional[Callable[[], Awaitable[aioredis.Redis]]] = ( db_manager.get_redis if db_manager else None ) async def _ensure_redis(self): """确保Redis连接可用""" if not self._redis_provider: if not self.db: raise RuntimeError("Redis 未初始化，缺少数据库管理器") self._redis_provider = self.db.get_redis redis_client = await self._redis_provider() if not redis_client: raise RuntimeError("Redis 客户端不可用") self.redis = redis_client return redis_client async def generate_summary( self, user_id: str, start_round: int, end_round: int ): """ 生成对话总结（异步执行） 逻辑： - 如果是第一次总结（start_round=1）：只总结当前15轮消息 - 如果是后续总结：获取之前的总结 + 当前15轮消息，进行累积总结 Args: user_id: 用户ID start_round: 起始轮次 end_round: 结束轮次 """ try: # 🔥 确保Redis连接可用 await self._ensure_redis() logger.info(f"🔄 开始生成总结: user={user_id}, rounds={start_round}-{end_round}") # 0. 参数验证 if end_round <= 0: logger.warning(f"⚠️ 无效的总结范围: user={user_id}, start_round={start_round}, end_round={end_round}") return if start_round > end_round: logger.warning(f"⚠️ 总结范围错误: user={user_id}, start_round={start_round}, end_round={end_round}") return # 使用 ORM 查询 from database import db_manager from models_db import ConversationSummary, Message # 1. 检查是否已存在总结 async with db_manager.get_session() as session: existing_query = await session.execute( select(ConversationSummary) .where(ConversationSummary.user_id == user_id) .where(ConversationSummary.start_round == start_round) .where(ConversationSummary.end_round == end_round) ) existing = existing_query.scalar_one_or_none() if existing: logger.info(f"⏭️ 总结已存在，跳过: user={user_id}, rounds={start_round}-{end_round}") return # 2. 获取之前的总结（如果有） previous_summaries = [] if start_round > 1: previous_query = await session.execute( select(ConversationSummary) .where(ConversationSummary.user_id == user_id) .where(ConversationSummary.end_round < start_round) .order_by(ConversationSummary.end_round.asc()) ) previous_summaries_list = previous_query.scalars().all() previous_summaries = [s.summary_text for s in previous_summaries_list] logger.info(f"📚 获取到 {len(previous_summaries)} 个历史总结") # 3. 获取待总结的消息 logger.info(f"🔍 [总结查询] user={user_id}, start_round={start_round}, end_round={end_round}") messages_query = await session.execute( select(Message) .where(Message.user_id == user_id) .where(Message.round_number >= start_round) .where(Message.round_number <= end_round) .order_by(Message.round_number.asc(), Message.id.asc()) ) messages_list = messages_query.scalars().all() messages = [ { "role": msg.role, "content": msg.content, "round_number": msg.round_number, } for msg in messages_list ] logger.info(f"📊 [总结查询结果] user={user_id}, 查询到 {len(messages)} 条消息") # 如果查询不到消息，检查数据库中实际有哪些轮次 if not messages: # 诊断：查询该用户的所有消息轮次 async with db_manager.get_session() as session: from sqlalchemy import func all_rounds_query = await session.execute( select(Message.round_number, func.count(Message.id).label('count')) .where(Message.user_id == user_id) .group_by(Message.round_number) .order_by(Message.round_number.asc()) ) all_rounds_list = all_rounds_query.all() all_rounds = [ {"round_number": r.round_number, "count": r.count} for r in all_rounds_list ] # 过滤掉 round_number = 0 的记录（可能是历史错误数据） valid_rounds = [r['round_number'] for r in all_rounds if r['round_number'] > 0] if all_rounds else [] logger.warning( f"⚠️ 没有消息可总结: user={user_id}, " f"查询范围={start_round}-{end_round}, " f"数据库中实际轮次={valid_rounds if valid_rounds else '无消息'}, " f"总记录数={len(all_rounds) if all_rounds else 0}" ) return # 记录查询到的消息轮次范围 if messages: actual_start = min(m['round_number'] for m in messages) actual_end = max(m['round_number'] for m in messages) logger.info(f"✅ [总结查询] 实际消息轮次范围: {actual_start}-{actual_end}, 共 {len(messages)} 条") # 4. 调用LLM生成总结（包含之前的总结） summary_text = await self._generate_summary_with_llm(messages, previous_summaries) if not summary_text: logger.warning(f"⚠️ 总结生成失败: user={user_id}") return # 4. 保存总结到MySQL（使用 ORM） summary_id = str(uuid.uuid4()) async with db_manager.get_session() as session: summary = ConversationSummary( summary_id=summary_id, user_id=user_id, start_round=start_round, end_round=end_round, summary_text=summary_text, summary_type="auto", message_count=len(messages), token_count=len(summary_text) // 2 ) session.add(summary) await session.commit() logger.info(f"✅ 总结保存成功: user={user_id}, messages={len(messages)}") # 5. 更新Redis缓存 await self._update_summary_cache(user_id, summary_text) # 6. 更新元数据 meta_key = f"user:{user_id}:meta" await self.redis.hset(meta_key, 'last_summary_round', end_round) await self.redis.expire(meta_key, 86400) logger.info(f"🎉 总结完成: user={user_id}, rounds={start_round}-{end_round}") except Exception as e: logger.error(f"❌ 生成总结失败: user={user_id}, error={e}") import traceback traceback.print_exc() async def _generate_summary_with_llm( self, messages: List[Dict], previous_summaries: List[str] = None ) -> str: """ 调用LLM生成总结 总结内容聚焦于： 1. 用户个人信息（姓名、年龄、家庭住址等）- 用于后续精准回答用户问题 2. 用户当前英语水平评估 - 用于逐步引导用户提高英语水平 Args: messages: 消息列表 previous_summaries: 之前的总结列表（用于累积总结） Returns: 总结文本（包含用户信息和英语水平评估） """ try: # 构建对话内容 conversation_lines = [] for msg in messages: role = msg['role'] content = msg['content'] round_num = msg.get('round_number', '') role_label = "用户" if role == "user" else "助手" conversation_lines.append(f"[第{round_num}轮] {role_label}: {content}") conversation = "\n".join(conversation_lines) # 构建总结Prompt if previous_summaries: # 有历史总结：进行累积总结 previous_summaries_text = "\n\n".join([ f"【历史总结{i+1}】\n{summary}" for i, summary in enumerate(previous_summaries) ]) prompt = f"""请基于以下历史总结和新的对话内容，生成一个更新的累积总结。 **历史总结：** {previous_summaries_text} **新的对话内容（第{messages[0].get('round_number', 1)}-{messages[-1].get('round_number', 15)}轮）：** {conversation} 请整合历史总结和新对话内容，更新以下两类关键信息： **第一类：用户个人信息** 请提取并整理用户在对话中透露的个人信息，包括但不限于： - 姓名、昵称 - 年龄 - 家庭住址、所在城市 - 职业、工作 - 兴趣爱好 - 家庭成员信息 - 其他个人背景信息 如果某些信息未在对话中提及，请标注"未提及"。 **第二类：用户英语水平评估** 请评估用户当前的英语水平，包括： - 词汇量水平（基础/中级/高级） - 语法掌握程度（能使用哪些时态、句型） - 口语流利度（是否流畅、是否有较多停顿） - 发音准确性（是否有明显发音问题） - 表达能力（能否清晰表达想法） - 当前薄弱环节（需要重点提升的方面） 请按照以下格式输出总结（每部分100-150字）： **用户个人信息：** [整合后的用户个人信息，未提及的标注"未提及"] **英语水平评估：** [对用户当前英语水平的详细评估]""" else: # 第一次总结：只总结当前对话 prompt = f"""请从以下英语口语练习对话中提炼两类关键信息： **第一类：用户个人信息** 请提取并整理用户在对话中透露的个人信息，包括但不限于： - 姓名、昵称 - 年龄 - 家庭住址、所在城市 - 职业、工作 - 兴趣爱好 - 家庭成员信息 - 其他个人背景信息 如果某些信息未在对话中提及，请标注"未提及"。 **第二类：用户英语水平评估** 请评估用户当前的英语水平，包括： - 词汇量水平（基础/中级/高级） - 语法掌握程度（能使用哪些时态、句型） - 口语流利度（是否流畅、是否有较多停顿） - 发音准确性（是否有明显发音问题） - 表达能力（能否清晰表达想法） - 当前薄弱环节（需要重点提升的方面） 对话内容： {conversation} 请按照以下格式输出总结（每部分100-150字）： **用户个人信息：** [提取的用户个人信息，未提及的标注"未提及"] **英语水平评估：** [对用户当前英语水平的详细评估]""" # 调用AI服务生成总结 from ai_service import AIService ai_service = AIService() # 使用非流式调用 summary = await ai_service.generate_summary(prompt) return summary.strip() except Exception as e: logger.error(f"❌ LLM总结生成失败: {e}") # 降级方案：简单拼接 return f"**用户个人信息：**\n未提及\n\n**英语水平评估：**\n对话总结（第{messages[0].get('round_number', 1)}-{messages[-1].get('round_number', 15)}轮）：共{len(messages)}条消息的英语口语练习对话。" async def _update_summary_cache(self, user_id: str, summary_text: str): """ 更新Redis中的总结缓存 Args: user_id: 用户ID summary_text: 总结文本 """ try: redis = await self._ensure_redis() cache_key = f"user:{user_id}:latest_summary" await redis.setex(cache_key, 86400, summary_text) logger.debug(f"✅ 总结已缓存: user={user_id}") except Exception as e: logger.warning(f"⚠️ 缓存总结失败: user={user_id}, error={e}") async def get_latest_summary(self, user_id: str) -> str: """ 获取用户最新的总结 Args: user_id: 用户ID Returns: 总结文本或空字符串 """ try: redis = await self._ensure_redis() # 先查Redis cache_key = f"user:{user_id}:latest_summary" cached = await redis.get(cache_key) if cached: return cached # 再查MySQL（使用 ORM） from database import db_manager from models_db import ConversationSummary async with db_manager.get_session() as session: result_query = await session.execute( select(ConversationSummary) .where(ConversationSummary.user_id == user_id) .order_by(ConversationSummary.end_round.desc()) .limit(1) ) summary = result_query.scalar_one_or_none() if summary: summary_text = summary.summary_text # 回填缓存 await redis.setex(cache_key, 86400, summary_text) return summary_text return "" except Exception as e: logger.error(f"❌ 获取最新总结失败: user={user_id}, error={e}") return ""