Travel_Assistant / modules /response_generator.py
Eliot0110's picture
fix: redundant symbol
2c6e1d9
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"]['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} 天旅行计划,建议包括以下内容:
1. 每日详细行程安排(按 Day 1、Day 2... 排列)
- 用“•”分点列出上午、下午活动,可加括号注明时间或费用(如:90分钟 / 免费)
2. 餐饮推荐:用 🍽️ 符号标注,并说明推荐理由
3. 住宿建议:用 🏨 符号标注,写明区域、安全性、价格段
4. 交通方式:如 🚇 地铁 / 🚶‍ 徒步 等
5. 每部分之间请换行分隔,避免将内容挤成一段。
✍️【风格要求】
请用亲切自然、具有温度的语气来写这份计划,像一位体贴的私人旅行顾问在说话。同时保持逻辑清晰,避免使用 markdown 格式,只需自然排版。
"""
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"
#