import json import random from .ai_model import AIModel from .knowledge_base import KnowledgeBase from .session_manager import SessionManager 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() self._init_response_templates() 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 _init_response_templates(self): """初始化各种动态回复模板""" # 欧洲城市特色描述 (保留原有) self.city_descriptions = { "巴黎": ["浪漫之都", "艺术之城", "时尚之都", "光影流转的塞纳河畔", "充满香槟气息的花都"], "罗马": ["永恒之城", "历史的活化石", "每块石头都有故事", "古典与现代交融的奇迹", "凯撒大帝走过的土地"], "伦敦": ["绅士的故乡", "雾都传奇", "文艺复兴的摇篮", "泰晤士河的守护者", "莎士比亚笔下的世界"], "维也纳": ["音乐之都", "华尔兹的发源地", "莫扎特的灵感之地", "咖啡文化的天堂", "皇室优雅的化身"], "布拉格": ["千塔之城", "中世纪的童话", "波西米亚的浪漫", "查理桥上的传奇", "啤酒花香弥漫的古城"], "布达佩斯": ["多瑙河明珠", "东欧巴黎", "温泉之都", "建筑艺术的博物馆", "匈牙利王冠上的明珠"], "萨尔茨堡": ["音乐神童的故乡", "《音乐之声》的拍摄地", "阿尔卑斯山下的明珠", "莫扎特的诞生地", "巴洛克建筑的典范"], "哈尔施塔特": ["世界最美小镇", "湖光山色的仙境", "阿尔卑斯山的秘境", "明信片上的童话", "奥地利的瑰宝"], } # 保留原有的问候语和确认模板 (简化以节省空间) self.greetings = { "social": [ "哈喽！准备开启一场说走就走的欧洲之旅吗？✨", "嗨呀！听说有人要去欧洲拍美照啦？📸", ], "experiential": [ "你好，旅行者。欧洲的古老土地正在召唤着你...", "感知到了一颗渴望探索的心。欧洲有太多故事等你去发现。", ], "planner": [ "您好！让我来帮您规划一次完美的欧洲之旅。", "欧洲旅行规划专家上线！准备为您定制专属行程。", ] } def _get_current_persona_config(self, session_state: dict) -> dict: """获取当前persona配置""" persona_info = session_state.get("persona") if not isinstance(persona_info, dict) or "key" not in persona_info: # 如果 persona 尚未设置或格式不正确，记录日志并抛出异常 log.error(f"❌ 在会话 {session_state.get('session_id')} 中缺少有效的 persona 配置。当前 persona 状态: {persona_info}") raise ValueError("无法在会话状态中找到有效的 'persona' 配置。对话流程可能存在问题。") persona_key = persona_info["key"] # 既然检查过，就可以安全地直接访问 persona_config = self.personas.get(persona_key) # 3. 检查获取到的 key 是否真的存在于配置中 if not persona_config: log.error(f"❌ persona key '{persona_key}' 在系统中未定义。") raise ValueError(f"提供的 persona key '{persona_key}' 是一个无效的配置项。") log.info(f"✅ 成功加载 Persona: {persona_config.get('name', persona_key)}") return persona_config def generate(self, user_message: str, session_state: dict, extracted_info: dict) -> str: """生成融合知识库的智能回复""" try: response_parts = [] # 1. 生成确认信息（更生动） acknowledgement = self._generate_vivid_acknowledgement(extracted_info, session_state) if acknowledgement: response_parts.append(acknowledgement) # 如果确认信息本身已经是一个问题（比如追问货币），就直接返回，避免再问下一个问题 if acknowledgement.strip().endswith(('?', '？')): return " ".join(response_parts) # 2. 检查是否需要询问下一个信息 next_question = self._get_dynamic_next_question(session_state) if next_question: if response_parts: connectors = ["那么，", "接下来，", "好的，", ""] connector = random.choice(connectors) response_parts.append(connector + next_question) else: response_parts.append(next_question) # 3. 如果所有信息收集完毕，生成知识库增强的旅行计划 if not next_question: plan = self._generate_knowledge_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 _generate_vivid_acknowledgement(self, extracted_info: dict, session_state: SessionManager) -> str: if "destination" in extracted_info and extracted_info["destination"]: dest_name = extracted_info["destination"]['name'] if dest_name in self.city_descriptions: feature = random.choice(self.city_descriptions[dest_name]) return f"{dest_name}！一个绝佳的选择，那可是著名的'{feature}'。目的地已为您记录。" else: dest_country = extracted_info["destination"][0]['country'] return f"好的，目的地已确认为 {dest_country} 的 {dest_name}！一个充满魅力的地方。" def _get_dynamic_next_question(self, session_state: SessionManager) -> str: if not session_state.get('destination'): return "请问您想去哪个或哪些城市呢？" if not session_state.get('duration'): return "计划玩几天呢？" if not session_state.get('budget'): return "您的旅行预算大概是多少？您可以金额+币种的格式输入,例如：5000元人民币 或 800 eur" return "" # 所有信息都已收集 def _get_destination_name(self, session_state: dict) -> str: destination_info = session_state.get('destination') if destination_info and isinstance(destination_info, dict): return destination_info.get('name', '未知目的地') return '未知目的地' def _get_duration_days(self, session_state: dict) -> int: duration_info = session_state.get('duration') if duration_info and isinstance(duration_info, dict): # 确保即使 'days' 键不存在，也返回一个数字 return duration_info.get('days', 0) # 关键：返回 0 而不是 None return 0 def _format_budget_info(self, budget_data: dict | None) -> str: if budget_data and isinstance(budget_data, dict): # 优先使用现成的描述，因为它最准确 if budget_data.get('description'): return budget_data['description'] # 如果没有描述，则动态创建一个 amount = budget_data.get('amount', '') currency = budget_data.get('currency', '') if amount: # 仅在有金额时才显示 return f"{amount} {currency}".strip() # 如果没有预算信息或信息不完整，返回默认字符串 return '预算未设定' def _generate_knowledge_enhanced_plan(self, user_message: str, session_state: dict) -> str: """生成融合知识库信息的旅行计划""" # 1. 获取目的地信息 destination_name = self._get_destination_name(session_state) days = int(self._get_duration_days(session_state)) budget_info = self._format_budget_info(session_state.get("budget")) log.info(f"🔍 开始搜索知识库中关于 '{destination_name}' 的信息...") # 2. 搜索知识库中的相关信息 relevant_knowledge = self._search_destination_knowledge(destination_name) # 3. 如果有AI模型，生成增强版计划 if self.ai_model and self.ai_model.is_available(): return self._generate_ai_enhanced_plan(session_state, relevant_knowledge) else: # 4. 否则生成基于知识库的详细备用计划 return self._generate_knowledge_based_fallback_plan(session_state, relevant_knowledge) def _search_destination_knowledge(self, destination_name: str) -> dict: """搜索知识库中与目的地相关的信息""" if not self.kb or not hasattr(self.kb, 'knowledge') or not self.kb.knowledge: log.warning("⚠️ 知识库为空或不可用") return {} relevant_info = { "budget_analysis": {}, "itinerary_suggestions": [], "professional_insights": {}, "destination_specific": {} } log.info(f"📚 在 {len(self.kb.knowledge)} 条知识中搜索关于 '{destination_name}' 的信息...") # 遍历知识库 for item in self.kb.knowledge: knowledge = item.get('knowledge', {}).get('travel_knowledge', {}) if not knowledge: continue # 检查是否与目标目的地相关 dest_info = knowledge.get('destination_info', {}) primary_destinations = dest_info.get('primary_destinations', []) countries = dest_info.get('countries', []) # 判断相关性 is_relevant = False match_reason = "" # 直接匹配城市名 if destination_name in primary_destinations: is_relevant = True match_reason = f"直接匹配城市: {destination_name}" # 通过国家匹配 if not is_relevant: dest_country = self._get_destination_country(destination_name) if dest_country and dest_country in countries: is_relevant = True match_reason = f"通过国家匹配: {dest_country}" # 地区匹配 (如果目的地在同一地区) if not is_relevant: region_destinations = self._get_same_region_cities(destination_name) if any(city in primary_destinations for city in region_destinations): is_relevant = True match_reason = f"同地区匹配: {region_destinations}" if is_relevant: log.info(f"✅ 找到相关知识: {match_reason}") # 提取预算分析 if 'budget_analysis' in knowledge: relevant_info['budget_analysis'] = knowledge['budget_analysis'] # 提取行程建议 if 'detailed_itinerary' in knowledge: relevant_info['itinerary_suggestions'].extend(knowledge['detailed_itinerary']) # 提取专业洞察 if 'professional_insights' in knowledge: relevant_info['professional_insights'].update(knowledge['professional_insights']) # 提取目的地特定信息 relevant_info['destination_specific'] = dest_info if relevant_info['budget_analysis'] or relevant_info['itinerary_suggestions']: log.info(f"📊 成功提取知识库信息: 预算分析={bool(relevant_info['budget_analysis'])}, 行程建议={len(relevant_info['itinerary_suggestions'])}条") else: log.warning(f"⚠️ 未找到关于 '{destination_name}' 的相关知识") return relevant_info def _get_destination_country(self, city_name: str) -> str: """获取城市所属国家""" city_country_mapping = { "布拉格": "捷克", "维也纳": "奥地利", "萨尔茨堡": "奥地利", "布达佩斯": "匈牙利", "布拉迪斯拉发": "斯洛伐克", "哈尔施塔特": "奥地利", "巴德伊舍": "奥地利", "库特纳霍拉": "捷克", "布尔诺": "捷克", "巴黎": "法国", "里昂": "法国", "尼斯": "法国", "柏林": "德国", "慕尼黑": "德国", "汉堡": "德国", "阿姆斯特丹": "荷兰", "鹿特丹": "荷兰", "布鲁塞尔": "比利时", "安特卫普": "比利时", "苏黎世": "瑞士", "日内瓦": "瑞士", } return city_country_mapping.get(city_name, "") def _get_same_region_cities(self, city_name: str) -> list: """获取同地区的其他城市""" region_mapping = { # 中欧城市 "布拉格": ["维也纳", "萨尔茨堡", "布达佩斯", "布拉迪斯拉发", "哈尔施塔特"], "维也纳": ["布拉格", "萨尔茨堡", "布达佩斯", "布拉迪斯拉发", "哈尔施塔特"], "萨尔茨堡": ["维也纳", "布拉格", "哈尔施塔特", "慕尼黑"], "布达佩斯": ["布拉格", "维也纳", "布拉迪斯拉发"], "哈尔施塔特": ["萨尔茨堡", "维也纳", "巴德伊舍"], # 西欧城市 "巴黎": ["布鲁塞尔", "阿姆斯特丹", "科隆", "斯特拉斯堡"], "阿姆斯特丹": ["布鲁塞尔", "科隆", "巴黎"], "布鲁塞尔": ["阿姆斯特丹", "巴黎", "科隆"], # 德语区 "柏林": ["慕尼黑", "科隆", "汉堡", "维也纳", "苏黎世"], "慕尼黑": ["柏林", "萨尔茨堡", "苏黎世", "维也纳"], "苏黎世": ["慕尼黑", "维也纳", "萨尔茨堡"], } return region_mapping.get(city_name, []) def _generate_ai_enhanced_plan(self, session_state: dict, knowledge: dict) -> str: """使用AI模型生成融合知识库的计划""" # 构建包含知识库信息的enhanced prompt enhanced_prompt = self._build_knowledge_enhanced_prompt(session_state, knowledge) try: log.info("🤖 使用AI模型生成知识库增强计划...") response = self.ai_model.run_inference( input_type="text", formatted_input=None, prompt=enhanced_prompt, temperature=0.7 ) return response except Exception as e: log.error(f"❌ AI增强计划生成失败: {e}") return self._generate_knowledge_based_fallback_plan(session_state, knowledge) def _build_knowledge_enhanced_prompt(self, session_state: dict, knowledge: dict) -> str: """构建融合知识库信息的增强prompt""" destination_name = self._get_destination_name(session_state) days = self._get_duration_days(session_state) budget_desc = self._format_budget_info(session_state.get("budget")) persona_config = self._get_current_persona_config(session_state) # 基础prompt prompt = f"""你是一位专业的欧洲旅行顾问，请基于以下知识库信息为用户设计{destination_name}的详细旅行计划。 🎯 【用户需求】 📍 目的地: {destination_name} ⏰ 旅行天数: {days}天 💰 预算: {budget_desc} 🎭 旅行风格: {persona_config.get('name', '标准旅行者')} 📚 【知识库参考信息】""" # 添加预算分析信息 if knowledge.get('budget_analysis'): budget_analysis = knowledge['budget_analysis'] prompt += f""" 💰 【预算参考】 • 总预算范围: {budget_analysis.get('total_budget_range', 'N/A')} • 日均开支: {budget_analysis.get('daily_average', 'N/A')}""" breakdown = budget_analysis.get('budget_breakdown', {}) if breakdown: prompt += "\n• 预算分配:" for category, info in breakdown.items(): if isinstance(info, dict): percentage = info.get('percentage', '') daily_range = info.get('daily_range', '') if percentage and daily_range: category_name = {'accommodation': '住宿', 'transportation': '交通', 'food': '餐饮', 'attractions': '景点'}.get(category, category) prompt += f"\n - {category_name}: {percentage}, {daily_range}" # 添加行程参考信息 if knowledge.get('itinerary_suggestions'): prompt += f""" 🗓️ 【行程参考】""" for day_plan in knowledge['itinerary_suggestions'][:3]: # 只取前3天作为参考 day_num = day_plan.get('day_number', 'N/A') location = day_plan.get('location', 'N/A') theme = day_plan.get('theme', 'N/A') prompt += f"\n• Day {day_num} ({location}): {theme}" # 添加具体活动 morning_activities = day_plan.get('morning_activities', []) for activity in morning_activities[:2]: # 只取前2个活动 name = activity.get('activity_name', '') duration = activity.get('duration', '') tips = activity.get('professional_tips', '') if name: prompt += f"\n - {name} ({duration}) - {tips}" # 添加专业洞察 if knowledge.get('professional_insights'): insights = knowledge['professional_insights'] prompt += f""" 💡 【专业建议】""" if insights.get('seasonal_considerations'): seasonal = insights['seasonal_considerations'] best_months = seasonal.get('best_months', []) if best_months: prompt += f"\n• 最佳旅行时间: {', '.join(best_months)}" if insights.get('common_mistakes'): mistakes = insights['common_mistakes'][:3] # 只取前3个 prompt += f"\n• 常见误区: {', '.join(mistakes)}" if insights.get('insider_secrets'): secrets = insights['insider_secrets'][:3] # 只取前3个 prompt += f"\n• 内行贴士: {', '.join(secrets)}" # 结尾指令 prompt += f""" 🌟 【生成要求】 请基于以上知识库信息，结合用户的具体需求，生成一份详细、实用、个性化的{destination_name}{days}天旅行计划。计划应该包括: • 每日详细行程安排 • 具体的景点推荐和游览建议 • 住宿和餐饮推荐 • 交通方式和路线优化 • 预算分解和省钱贴士 • 实用的旅行提醒 请用生动有趣的语言撰写，让用户感受到{destination_name}的魅力！""" return prompt def _generate_knowledge_based_fallback_plan(self, session_state: dict, knowledge: dict) -> str: """基于知识库生成详细的备用计划""" destination_name = self._get_destination_name(session_state) days = int(self._get_duration_days(session_state)) budget_desc = self._format_budget_info(session_state.get("budget")) persona_config = self._get_current_persona_config(session_state) persona_key = persona_config.get('key', 'planner') # 获取城市特色描述 city_desc = random.choice(self.city_descriptions.get(destination_name, ["迷人的城市"])) # 开场 if persona_key == 'social': plan = f"🎉 {destination_name}{days}天深度攻略（知识库加持版）！\n\n" elif persona_key == 'experiential': plan = f"🎭 {destination_name}{days}日文化探索之旅\n\n" else: plan = f"📋 {destination_name}{days}天专业规划方案\n\n" plan += f"🌟 城市印象：{city_desc}\n" plan += f"💰 预算范围：{budget_desc}\n\n" # 如果有知识库中的预算分析 if knowledge.get('budget_analysis'): budget_analysis = knowledge['budget_analysis'] plan += "💰 【预算详解】（基于真实旅行经验）\n" total_budget = budget_analysis.get('total_budget_range', '') daily_avg = budget_analysis.get('daily_average', '') if total_budget: plan += f"• 参考总预算：{total_budget}\n" if daily_avg: plan += f"• 日均开支：{daily_avg}\n" breakdown = budget_analysis.get('budget_breakdown', {}) if breakdown: plan += "• 开支分配：\n" category_names = { 'accommodation': '🏨 住宿', 'transportation': '🚇 交通', 'food': '🍽️ 餐饮', 'attractions': '🎯 景点' } for category, info in breakdown.items(): if isinstance(info, dict): name = category_names.get(category, category) percentage = info.get('percentage', '') daily_range = info.get('daily_range', '') if percentage and daily_range: plan += f" - {name}：{percentage}，{daily_range}\n" # 添加具体建议 if category == 'accommodation' and info.get('recommendations'): recs = ', '.join(info['recommendations']) plan += f" 推荐：{recs}\n" elif category == 'transportation' and info.get('money_saving_tips'): tips = ', '.join(info['money_saving_tips']) plan += f" 省钱技巧：{tips}\n" plan += "\n" # 详细行程规划（基于知识库） plan += "🗓️ 【详细行程】（来自实地经验）\n" if knowledge.get('itinerary_suggestions'): # 使用知识库中的行程建议 itinerary = knowledge['itinerary_suggestions'] for i, day_plan in enumerate(itinerary[:days]): # 限制在用户要求的天数内 day_num = day_plan.get('day_number', i+1) location = day_plan.get('location', destination_name) theme = day_plan.get('theme', '城市探索') plan += f"\n📅 Day {day_num} - {location}（{theme}）\n" # 上午活动 morning_activities = day_plan.get('morning_activities', []) if morning_activities: plan += "🌅 上午：\n" for activity in morning_activities: name = activity.get('activity_name', '') duration = activity.get('duration', '') cost = activity.get('cost', '') tips = activity.get('professional_tips', '') plan += f" • {name}" if duration: plan += f" ({duration})" if cost and cost != "免费": plan += f" - {cost}" plan += "\n" if tips: plan += f" 💡 专业提醒：{tips}\n" # 下午活动 afternoon_activities = day_plan.get('afternoon_activities', []) if afternoon_activities: plan += "🌞 下午：\n" for activity in afternoon_activities: name = activity.get('activity_name', '') duration = activity.get('duration', '') cost = activity.get('cost', '') plan += f" • {name}" if duration: plan += f" ({duration})" if cost: plan += f" - {cost}" plan += "\n" # 餐饮建议 dining = day_plan.get('dining', {}) if dining: plan += "🍽️ 餐饮推荐：\n" for meal_type, meal_info in dining.items(): if isinstance(meal_info, dict): meal_names = {'breakfast': '早餐', 'lunch': '午餐', 'dinner': '晚餐'} meal_name = meal_names.get(meal_type, meal_type) recommendation = meal_info.get('recommendation', '') cost_range = meal_info.get('cost_range', '') if recommendation: plan += f" • {meal_name}：{recommendation}" if cost_range: plan += f" ({cost_range})" plan += "\n" # 住宿建议 accommodation = day_plan.get('accommodation', {}) if accommodation and day_num == 1: # 只在第一天显示住宿建议 plan += "🏨 住宿推荐：\n" area = accommodation.get('recommended_area', '') safety = accommodation.get('safety_level', '') if area: plan += f" • 推荐区域：{area}" if safety: plan += f"（安全等级：{safety}）" plan += "\n" budget_options = accommodation.get('budget_options', []) for option in budget_options: if isinstance(option, dict): category = option.get('category', '') price_range = option.get('price_range', '') if category and price_range: plan += f" • {category}：{price_range}\n" else: # 如果没有具体行程，生成通用建议 plan += f"根据{destination_name}的特色，为您推荐以下{days}天行程框架：\n\n" # 根据不同城市提供基础框架 if destination_name in ["布拉格", "Prague"]: plan += "📅 Day 1: 老城区探索（老城广场→天文钟→查理大桥）\n" plan += "📅 Day 2: 城堡区深度游（布拉格城堡→圣维特大教堂→黄金小巷）\n" if days >= 3: plan += "📅 Day 3: 新城区体验（瓦茨拉夫广场→国家博物馆→当地美食）\n" elif destination_name in ["维也纳", "Vienna"]: plan += "📅 Day 1: 皇室风采（美泉宫→霍夫堡宫→圣斯蒂芬大教堂）\n" plan += "📅 Day 2: 音乐文化（维也纳国家歌剧院→金色大厅→艺术史博物馆）\n" if days >= 3: plan += "📅 Day 3: 咖啡文化体验（中央咖啡馆→萨赫咖啡馆→多瑙河漫步）\n" elif destination_name in ["布达佩斯", "Budapest"]: plan += "📅 Day 1: 布达一侧（布达城堡→渔夫堡→马加什教堂）\n" plan += "📅 Day 2: 佩斯一侧（匈牙利国会大厦→链子桥→中央市场）\n" if days >= 3: plan += "📅 Day 3: 温泉文化（塞切尼温泉→多瑙河游船→夜景欣赏）\n" # 添加专业洞察 if knowledge.get('professional_insights'): insights = knowledge['professional_insights'] plan += "\n💡 【专业贴士】（来自旅行达人）\n" # 季节建议 seasonal = insights.get('seasonal_considerations', {}) if seasonal: best_months = seasonal.get('best_months', []) weather = seasonal.get('weather_patterns', '') if best_months: plan += f"• 🌤️ 最佳旅行时间：{', '.join(best_months)}\n" if weather: plan += f"• 🌡️ 天气特点：{weather}\n" #