""" 聊天总结服务 负责调用LLM对历史消息进行总结 """ import json from typing import List, Dict, Any, Optional from datetime import datetime from models.internal import ChatMessage, SummaryConfig from core.config import settings class ChatSummaryService: """聊天总结服务类""" @staticmethod async def summarize_messages( messages: List[ChatMessage], start_floor: int, end_floor: int, summary_config: SummaryConfig, api_config: Dict[str, str] ) -> str: """ 对指定范围的消息进行总结 Args: messages: 完整的消息列表 start_floor: 总结起始楼层(1-based) end_floor: 总结结束楼层(1-based) summary_config: 总结配置 api_config: API配置 {api_url, api_key, model} Returns: 总结文本 """ # 1. 提取需要总结的消息 messages_to_summarize = ChatSummaryService._extract_messages( messages, start_floor, end_floor, summary_config.includeUserInput ) if not messages_to_summarize: return "" # 2. 构建总结提示词 prompt = ChatSummaryService._build_summary_prompt( messages_to_summarize, summary_config ) # 3. 调用LLM生成总结 summary_text = await ChatSummaryService._call_llm_for_summary( prompt, api_config, summary_config.maxSummaryLength ) return summary_text @staticmethod def _extract_messages( messages: List[ChatMessage], start_floor: int, end_floor: int, include_user_input: bool ) -> List[ChatMessage]: """ 提取需要总结的消息 ✅ 根据用户需求:后端不筛选,全部输入给LLM Args: messages: 完整消息列表 start_floor: 起始楼层 end_floor: 结束楼层 include_user_input: 是否包含用户输入(此参数目前不使用,但保留以保持接口兼容) Returns: 需要总结的消息列表(全部消息,不做筛选) """ # 转换为0-based索引 start_idx = start_floor - 1 end_idx = end_floor - 1 # 提取范围内的所有消息(不筛选) return messages[start_idx:end_idx + 1] @staticmethod def _build_summary_prompt( messages: List[ChatMessage], summary_config: SummaryConfig ) -> str: """ 构建总结提示词 Args: messages: 需要总结的消息列表 summary_config: 总结配置 Returns: 完整的提示词 """ # 使用用户自定义的总结提示词,或默认提示词 base_prompt = summary_config.summaryPrompt or ( "请总结以下对话内容,保留关键信息和上下文。" "用简洁的语言概括主要事件、人物状态和重要细节。" ) # 构建对话内容 conversation_text = "\n\n".join([ f"{'用户' if msg.is_user else msg.name}: {msg.mes}" for msg in messages ]) # 组合完整提示词 full_prompt = f"""{base_prompt} 对话内容: {conversation_text} 总结要求: 1. 保持简洁明了 2. 保留关键情节和设定 3. 不超过{summary_config.maxSummaryLength}字 4. 使用客观叙述语气 总结:""" return full_prompt @staticmethod async def _call_llm_for_summary( prompt: str, api_config: Dict[str, str], max_length: int ) -> str: """ 调用LLM生成总结 Args: prompt: 总结提示词 api_config: API配置 max_length: 最大长度限制 Returns: 总结文本 """ try: # 导入LLM客户端 from utils.llm_client import llm_client # 构建消息 messages = [ { "role": "system", "content": "你是一个专业的对话总结助手,擅长提取关键信息并用简洁的语言概括。" }, { "role": "user", "content": prompt } ] # 调用LLM response = await llm_client.chat_completion( messages=messages, api_url=api_config.get("api_url", ""), api_key=api_config.get("api_key", ""), model=api_config.get("model", "gpt-3.5-turbo"), temperature=0.3, # 总结需要较低的随机性 max_tokens=max_length, request_timeout=30 ) # 提取总结文本 if isinstance(response, dict): summary = response.get("choices", [{}])[0].get("message", {}).get("content", "") else: summary = str(response) # 清理和截断 summary = summary.strip() if len(summary) > max_length: summary = summary[:max_length] + "..." return summary except Exception as e: print(f"[ChatSummary] ❌ LLM总结失败: {e}") # 返回降级总结(基于规则的简单摘要) return ChatSummaryService._fallback_summary(prompt) @staticmethod def _fallback_summary(prompt: str) -> str: """ 降级总结(当LLM调用失败时使用) Args: prompt: 原始提示词 Returns: 简单的降级总结 """ # 提取对话中的关键信息 lines = prompt.split("\n") user_lines = [l for l in lines if l.startswith("用户:")] ai_lines = [l for l in lines if l.startswith("AI:")] fallback = f"[自动总结] 对话包含 {len(user_lines)} 条用户消息和 {len(ai_lines)} 条AI回复。" return fallback # 全局实例 chat_summary_service = ChatSummaryService()