import json from .ai_model import AIModel from .knowledge_base import KnowledgeBase from utils.logger import log class ResponseGenerator: def __init__(self, ai_model: AIModel, knowledge_base: KnowledgeBase): self.ai_model = ai_model self.kb = knowledge_base self.personas = self._load_personas() def _load_personas(self): personas_path = "./config/personas.json" with open(personas_path, 'r', encoding='utf-8') as f: data = json.load(f) log.info(f"✅ 成功加载 {len(data.get('personas', {}))} 个persona配置。") return data.get('personas', {}) def _get_current_persona_config(self, session_state: dict) -> dict: """安全地获取当前会话的persona配置字典""" persona_key = session_state.get("persona", {}).get("key") if persona_key and persona_key in self.personas: return self.personas[persona_key] # 如果没有找到指定的persona，返回一个中立的默认配置 return { "name": "旅行助手", "tone": ["专业", "友好"], "style": "中立" } def generate(self, user_message: str, session_state: dict, extracted_info: dict) -> str: """生成智能响应，整个过程融入Persona风格""" try: response_parts = [] # 1. 生成带有人格风格的确认反馈 acknowledgement = self._generate_acknowledgement(extracted_info, session_state) if acknowledgement: response_parts.append(acknowledgement) # 2. 检查信息完整性，用人格化的方式询问缺失信息 next_question = self._get_next_question(session_state) # 只有在有下一步问题时才加入回复列表 if next_question: # 如果已有确认信息，为了避免语气生硬，加个连接词 if response_parts: response_parts.append("那么，" + next_question[0].lower() + next_question[1:]) else: response_parts.append(next_question) # 3. 如果所有信息都齐全，并且没有追问，则生成最终计划 if not next_question: plan = self._generate_persona_enhanced_plan(user_message, session_state) # 如果已有确认信息，用换行符分隔，使计划更突出 if response_parts: response_parts.append("\n\n" + plan) else: response_parts.append(plan) return " ".join(response_parts) except Exception as e: log.error(f"❌ 响应生成失败: {e}", exc_info=True) return "抱歉，我在处理您的请求时遇到了问题，请稍后再试。" def _get_next_question(self, session_state: dict) -> str: """根据Persona的风格，获取下一个需要询问的问题""" persona_config = self._get_current_persona_config(session_state) persona_style = persona_config.get("style", "") destination_name = session_state.get('destination', {}).get('name', '那里') days = session_state.get('duration', {}).get('days', '几') # --- 按优先级检查缺失信息，并根据Persona风格提问 --- # 1. 目的地 if not session_state.get("destination"): if "社交" in persona_style: return "哈喽！准备好去哪里嗨皮了吗？告诉我想去哪个城市，我们来一场刷爆朋友圈的旅行吧！✨" if "体验" in persona_style: return "你好，旅行者。为了开启一段独特的深度体验，你心中的目的地是哪里？" # 默认使用规划师或中立风格 return "您好！为了高效地开始规划，请首先明确您的目的地城市。" # 2. 天数 if not session_state.get("duration"): if "社交" in persona_style: return f"{destination_name}超棒的！打算和小伙伴们在那玩几天呀？" if "体验" in persona_style: return f"感知到了，{destination_name}。你希望在这片土地上沉浸多少个日夜？" return f"目的地已锁定：{destination_name}。请提供计划的旅行天数。" # 3. 预算 if not session_state.get("budget"): if "社交" in persona_style: return f"太棒啦，{days}天的行程！这次出去玩，预算大概是多少呀？是经济实惠，还是想来个轻奢体验呢？" if "体验" in persona_style: return f"{days}天的探索之旅，听起来很不错。对于这次旅行的开销，你有什么样的构想？" return f"已记录：行程共{days}天。请明确您的预算范围（例如：经济型、舒适型，或具体金额）。" return "" # 所有信息已收集完毕 def _generate_acknowledgement(self, extracted_info: dict, session_state: dict) -> str: """根据Persona的风格，生成对新提取信息的确认反馈""" if not extracted_info: return "" persona_config = self._get_current_persona_config(session_state) persona_style = persona_config.get("style", "") ack_parts = [] if "destination" in extracted_info: name = extracted_info['destination'].get('name', '目的地') if "社交" in persona_style: ack_parts.append(f"目的地锁定{name}！已经开始期待啦！💖") elif "体验" in persona_style: ack_parts.append(f"我感知到了，{name}，一个充满故事的地方") else: ack_parts.append(f"确认：目的地已记录为{name}") if "duration" in extracted_info: days = extracted_info['duration'].get('days', '几') if "社交" in persona_style: ack_parts.append(f"玩{days}天，时间超充裕的") elif "体验" in persona_style: ack_parts.append(f"{days}个日夜，足够深入探索了") else: ack_parts.append(f"行程时长已设定为{days}天") if "budget" in extracted_info: budget_desc = self._format_budget_info(extracted_info['budget']) if "社交" in persona_style: ack_parts.append(f"{budget_desc}的预算，妥妥的") elif "体验" in persona_style: ack_parts.append(f"了解，{budget_desc}的投入，追求的是价值而非价格") else: ack_parts.append(f"预算已明确为{budget_desc}") return "，".join(ack_parts) + "。" if ack_parts else "" # --- 以下方法保留您原有的优秀实现，无需修改 --- def _generate_persona_enhanced_plan(self, user_message: str, session_state: dict) -> str: persona_config = self._get_current_persona_config(session_state) destination = session_state.get("destination", {}) duration = session_state.get("duration", {}) budget = session_state.get("budget", {}) location = destination.get('name', '目的地') days = duration.get('days', '几') budget_info = self._format_budget_info(budget) if self.ai_model.is_available(): prompt = self._build_prompt(session_state, persona_config) log.info(f"🚀 使用Persona '{persona_config.get('name')}' 构建的Prompt进行生成。") return self.ai_model.generate(user_message, prompt) else: log.warning("⚠️ AI模型不可用，生成备用计划。") return self._generate_fallback_plan(session_state) def _build_prompt(self, session_state: dict, persona_config: dict) -> str: destination = session_state.get("destination", {}) duration = session_state.get("duration", {}) budget = session_state.get("budget", {}) location = destination.get('name', '目的地') days = duration.get('days', '几') budget_info = self._format_budget_info(budget) template = persona_config.get('prompt_template') if template: try: # 填充模板所需的所有潜在变量 return template.format( location=location, days=days, budget=budget_info, date=session_state.get('date', '近期'), user_tags=", ".join(session_state.get('user_tags', [])), commercial_preference=session_state.get('commercial_preference', '适中'), group_description=session_state.get('group_description', '个人'), tags=", ".join(session_state.get('tags', [])) ) except KeyError as e: log.warning(f"Persona模板格式化失败，缺少键: {e}。将使用通用模板。") # Fallback to generic prompt if template fails return self._build_generic_prompt(session_state) return self._build_generic_prompt(session_state) def _format_budget_info(self, budget: dict) -> str: if not budget: return "未指定" if budget.get('amount') and budget.get('currency'): return f"{budget['amount']}{budget['currency']}" if budget.get('description'): return budget['description'] if budget.get('type'): type_map = {'economy': '经济型', 'comfortable': '舒适型', 'luxury': '豪华型'} return type_map.get(budget['type'], budget['type']) return "未指定" def _build_generic_prompt(self, session_state: dict, knowledge_context: str = "") -> str: # (此方法及以下方法保持您原有的实现) destination = session_state.get("destination", {}) duration = session_state.get("duration", {}) budget = session_state.get("budget", {}) location = destination.get('name', '目的地') days = duration.get('days', '几') budget_info = self._format_budget_info(budget) prompt = f"""你是一个专业的旅游助手。请为用户生成一个详细的旅行计划。 【基本信息】 - 目的地：{location} - 旅行天数：{days}天 - 预算：{budget_info} 【要求】 - 提供具体的景点推荐和路线安排 - 包含交通、住宿、餐饮建议 - 确保所有推荐都在预算范围内 - 提供实用的旅行贴士""" if knowledge_context: prompt += f"\n\n【背景信息】\n{knowledge_context}" prompt += "\n\n请生成一份实用、详细的旅行计划。" return prompt def _generate_fallback_plan(self, session_state: dict, knowledge_context: str = "") -> str: # (此方法保持您原有的实现) destination = session_state.get("destination", {}) duration = session_state.get("duration", {}) budget = session_state.get("budget", {}) persona_config = self._get_current_persona_config(session_state) location = destination.get('name', '目的地') days = duration.get('days', '几') budget_info = self._format_budget_info(budget) persona_name = persona_config.get('name', '旅行者') plan = f"为您推荐 {location} {days}天旅行计划：\n\n" plan += f"👤 旅行者类型：{persona_name}\n" plan += f"💰 预算范围：{budget_info}\n\n" if knowledge_context: plan += f"📚 {knowledge_context}\n\n" highlights = destination.get('highlights', '精彩景点等待您的探索') plan += f"🎯 主要景点：{highlights}\n\n" persona_key = session_state.get("persona", {}).get("key") if persona_key == 'planner': plan += "📋 建议制定详细的每日行程表。\n" elif persona_key == 'social': plan += "📸 推荐寻找热门打卡点！\n" elif persona_key == 'experiential': plan += "🎨 建议深入当地社区，寻找地道体验。\n" plan += "\n如需更详细的个性化规划，请告诉我您的具体需求！" return plan