improve: 优化模型调用并对各组件升级

app.py

@@ -0,0 +1,45 @@
+from transformers import AutoProcessor, Gemma3nForConditionalGeneration
+from PIL import Image
+import requests
+import torch
+
+model_id = "google/gemma-3n-e2b-it"
+
+model = Gemma3nForConditionalGeneration.from_pretrained(model_id, device="cuda", torch_dtype=torch.bfloat16,).eval()
+
+processor = AutoProcessor.from_pretrained(model_id)
+
+messages = [
+    {
+        "role": "system",
+        "content": [{"type": "text", "text": "You are a helpful assistant."}]
+    },
+    {
+        "role": "user",
+        "content": [
+            {"type": "image", "image": "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/bee.jpg"},
+            {"type": "text", "text": "Describe this image in detail."}
+        ]
+    }
+]
+
+inputs = processor.apply_chat_template(
+    messages,
+    add_generation_prompt=True,
+    tokenize=True,
+    return_dict=True,
+    return_tensors="pt",
+).to(model.device, dtype=torch.bfloat16)
+
+input_len = inputs["input_ids"].shape[-1]
+
+with torch.inference_mode():
+    generation = model.generate(**inputs, max_new_tokens=100, do_sample=False)
+    generation = generation[0][input_len:]
+
+decoded = processor.decode(generation, skip_special_tokens=True)
+print(decoded)
+
+# **Overall Impression:** The image is a close-up shot of a vibrant garden scene,
+# focusing on a cluster of pink cosmos flowers and a busy bumblebee.
+# It has a slightly soft, natural feel, likely captured in daylight.

modules/__init__.py

@@ -0,0 +1,29 @@
+# 导入所有模块类
+from .config_loader import ConfigLoader
+from .ai_model import AIModel
+from .knowledge_base import KnowledgeBase
+from .info_extractor import InfoExtractor
+from .session_manager import SessionManager
+from .response_generator import ResponseGenerator
+from .travel_assistant import TravelAssistant
+
+# 定义包的公共接口
+__all__ = [
+    'ConfigLoader',
+    'AIModel', 
+    'KnowledgeBase',
+    'InfoExtractor',
+    'SessionManager',
+    'ResponseGenerator',
+    'TravelAssistant'
+]
+
+# 版本信息
+__version__ = '1.0.0'
+
+# 包级别的便捷函数（可选）
+def create_travel_assistant():
+    """
+    便捷函数：创建一个完整配置的旅游助手实例
+    """
+    return TravelAssistant()

modules/ai_model.py

@@ -0,0 +1,189 @@
+# modules/ai_model.py
+import torch
+import base64
+import requests
+from io import BytesIO
+from PIL import Image
+from transformers import AutoProcessor, Gemma3nForConditionalGeneration
+from utils.logger import log
+from typing import Union, Tuple
+
+class AIModel:
+    def __init__(self, model_name: str = "google/gemma-3n-e2b-it"):
+        self.model_name = model_name
+        self.model = None
+        self.processor = None
+        self._initialize_model()
+
+    def _initialize_model(self):
+        """初始化Gemma模型 - 基于官方调用方式"""
+        try:
+            log.info(f"正在加载模型: {self.model_name}")
+            
+            self.model = Gemma3nForConditionalGeneration.from_pretrained(
+                self.model_name,
+                device_map="auto",
+                torch_dtype=torch.bfloat16,
+                trust_remote_code=True
+            ).eval()
+            
+            self.processor = AutoProcessor.from_pretrained(
+                self.model_name,
+                trust_remote_code=True
+            )
+            
+            log.info("✅ Gemma AI 模型初始化成功")
+            
+        except Exception as e:
+            log.error(f"❌ Gemma AI 模型初始化失败: {e}", exc_info=True)
+            self.model = None
+            self.processor = None
+
+    def is_available(self) -> bool:
+        """检查模型是否可用"""
+        return self.model is not None and self.processor is not None
+
+    def detect_input_type(self, input_data: str) -> str:
+        """检测输入类型：图片/音频/文字"""
+        if isinstance(input_data, str):
+            # 检查是否为图片URL或路径
+            if (input_data.startswith(("http://", "https://")) and 
+                any(input_data.lower().endswith(ext) for ext in [".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"])):
+                return "image"
+            elif input_data.endswith((".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp")):
+                return "image"
+            # 检查是否为音频URL或路径
+            elif (input_data.startswith(("http://", "https://")) and 
+                  any(input_data.lower().endswith(ext) for ext in [".wav", ".mp3", ".m4a", ".ogg"])):
+                return "audio"
+            elif input_data.endswith((".wav", ".mp3", ".m4a", ".ogg")):
+                return "audio"
+            # 检查是否为base64编码的图片
+            elif input_data.startswith("data:image/"):
+                return "image"
+        
+        return "text"
+
+    def format_input(self, input_type: str, raw_input: str) -> Tuple[str, Union[str, Image.Image, None]]:
+        """格式化输入数据"""
+        formatted_data = None
+        processed_text = raw_input
+        
+        if input_type == "image":
+            try:
+                if raw_input.startswith("data:image/"):
+                    # 处理base64编码的图片
+                    header, encoded = raw_input.split(",", 1)
+                    image_data = base64.b64decode(encoded)
+                    image = Image.open(BytesIO(image_data)).convert("RGB")
+                elif raw_input.startswith(("http://", "https://")):
+                    # 处理图片URL
+                    response = requests.get(raw_input, timeout=10)
+                    response.raise_for_status()
+                    image = Image.open(BytesIO(response.content)).convert("RGB")
+                else:
+                    # 处理本地图片路径
+                    image = Image.open(raw_input).convert("RGB")
+                
+                formatted_data = image
+                processed_text = "请描述这张图片，并基于图片内容提供旅游建议。"
+                log.info("✅ 图片加载成功")
+                
+            except Exception as e:
+                log.error(f"❌ 图片加载失败: {e}")
+                return "text", f"图片加载失败，请检查图片路径或URL。原始输入: {raw_input}"
+                
+        elif input_type == "audio":
+            # 音频处理逻辑（如果需要的话，目前先返回提示）
+            log.warning("⚠️ 音频处理功能暂未实现")
+            processed_text = "抱歉，音频输入功能正在开发中。请使用文字描述您的需求。"
+            
+        elif input_type == "text":
+            # 文字输入直接使用
+            formatted_data = None
+            processed_text = raw_input
+        
+        return input_type, formatted_data, processed_text
+
+    def run_inference(self, input_type: str, formatted_input: Union[str, Image.Image], prompt: str) -> str:
+        """执行模型推理"""
+        try:
+            if input_type == "image" and isinstance(formatted_input, Image.Image):
+                # 图片输入处理
+                image_token = self.processor.tokenizer.image_token
+                if image_token not in prompt:
+                    prompt = f"{image_token}\n{prompt}"
+                
+                inputs = self.processor(
+                    text=prompt, 
+                    images=formatted_input, 
+                    return_tensors="pt"
+                ).to(self.model.device, dtype=torch.bfloat16)
+            else:
+                # 纯文本输入处理
+                inputs = self.processor(
+                    text=prompt, 
+                    return_tensors="pt"
+                ).to(self.model.device, dtype=torch.bfloat16)
+
+            # 生成响应
+            with torch.inference_mode():
+                outputs = self.model.generate(
+                    **inputs, 
+                    max_new_tokens=512,
+                    do_sample=True,
+                    temperature=0.7,
+                    top_p=0.9,
+                    pad_token_id=self.processor.tokenizer.eos_token_id
+                )
+            
+            # 解码输出
+            decoded = self.processor.tokenizer.decode(outputs[0], skip_special_tokens=True).strip()
+            
+            # 清理输出，移除输入的prompt部分
+            if prompt in decoded:
+                decoded = decoded.replace(prompt, "").strip()
+            
+            return decoded
+            
+        except Exception as e:
+            log.error(f"❌ 模型推理失败: {e}", exc_info=True)
+            return "抱歉，我在处理您的请求时遇到了技术问题，请稍后再试。"
+
+    def generate(self, user_input: str, context: str = "") -> str:
+        """主要的生成方法 - 支持多模态输入"""
+        if not self.is_available():
+            return "抱歉，AI 模型当前不可用，请稍后再试。"
+        
+        try:
+            # 1. 检测输入类型
+            input_type = self.detect_input_type(user_input)
+            log.info(f"检测到输入类型: {input_type}")
+            
+            # 2. 格式化输入
+            input_type, formatted_data, processed_text = self.format_input(input_type, user_input)
+            
+            # 3. 构建prompt
+            if context:
+                prompt = (
+                    f"你是一个专业的旅游助手。请基于以下背景信息，用中文友好地回答用户的问题。\n\n"
+                    f"--- 背景信息 ---\n{context}\n\n"
+                    f"--- 用户问题 ---\n{processed_text}\n\n"
+                    f"请提供专业、实用的旅游建议："
+                )
+            else:
+                prompt = (
+                    f"你是一个专业的旅游助手。请用中文友好地回答用户的问题。\n\n"
+                    f"用户问题：{processed_text}\n\n"
+                    f"请提供专业、实用的旅游建议："
+                )
+            
+            # 4. 执行推理
+            if input_type == "image" and formatted_data is not None:
+                return self.run_inference("image", formatted_data, prompt)
+            else:
+                return self.run_inference("text", processed_text, prompt)
+                
+        except Exception as e:
+            log.error(f"❌ 生成回复时发生错误: {e}", exc_info=True)
+            return "抱歉，我在思考时遇到了点麻烦，请稍后再试。"

modules/config_loader.py

@@ -0,0 +1,33 @@
+# modules/config_loader.py
+import json
+from pathlib import Path
+from utils.logger import log
+
+class ConfigLoader:
+    def __init__(self, config_dir: Path = Path("./config")):
+        self.config_dir = config_dir
+        self.cities = {}
+        self.personas = {}
+        self.interests = {}
+        try:
+            self._load_all()
+            log.info("✅ 所有配置文件加载成功")
+        except Exception as e:
+            log.error(f"❌ 配置文件加载失败: {e}", exc_info=True)
+            raise
+
+    def _load_all(self):
+        # 加载城市
+        with open(self.config_dir / "cities.json", 'r', encoding='utf-8') as f:
+            cities_data = json.load(f)
+            for city in cities_data['cities']:
+                for alias in [city['name']] + city.get('aliases', []):
+                    self.cities[alias.lower()] = city
+        
+        # 加载 personas
+        with open(self.config_dir / "personas.json", 'r', encoding='utf-8') as f:
+            self.personas = json.load(f)['personas']
+        
+        # 加载兴趣
+        with open(self.config_dir / "interests.json", 'r', encoding='utf-8') as f:
+            self.interests = json.load(f)['interests']

modules/info_extractor.py

@@ -0,0 +1,31 @@
+# modules/info_extractor.py
+import re
+from .config_loader import ConfigLoader
+
+class InfoExtractor:
+    def __init__(self, config_loader: ConfigLoader):
+        self.configs = config_loader
+
+    def extract(self, user_input: str) -> dict:
+        """从用户输入中提取目的地、天数和旅行风格"""
+        extracted_info = {}
+        user_lower = user_input.lower()
+
+        # 提取目的地
+        for alias, city_info in self.configs.cities.items():
+            if alias in user_lower:
+                extracted_info["destination"] = city_info
+                break
+
+        # 提取天数
+        match = re.search(r'(\d+)\s*天', user_input)
+        if match:
+            extracted_info["duration"] = {"days": int(match.group(1))}
+        
+        # 提取旅行风格 (persona)
+        for p_name, p_info in self.configs.personas.items():
+            if p_info['name'] in user_input or p_name in user_input:
+                extracted_info["persona"] = p_info
+                break
+        
+        return extracted_info

modules/knowledge_base.py

@@ -0,0 +1,29 @@
+# modules/knowledge_base.py
+import json
+from pathlib import Path
+from utils.logger import log
+
+class KnowledgeBase:
+    def __init__(self, file_path: Path = Path("./config/general_travelplan.json")):
+        self.knowledge = []
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                self.knowledge = json.load(f).get('clean_knowledge', [])
+            log.info(f"✅ 知识库加载完成")
+        except Exception as e:
+            log.error(f"❌ 知识库加载失败: {e}", exc_info=True)
+            raise
+
+    def search(self, query: str) -> list:
+        relevant_knowledge = []
+        query_lower = query.lower()
+        
+        for item in self.knowledge:
+            # 简单实现：如果查询的城市在知识库的目的地中，则返回该知识
+            destinations = item.get('knowledge', {}).get('travel_knowledge', {}).get('destination_info', {}).get('primary_destinations', [])
+            for dest in destinations:
+                if dest.lower() in query_lower:
+                    if item not in relevant_knowledge:
+                        relevant_knowledge.append(item)
+                    break 
+        return relevant_knowledge

modules/response_generator.py

@@ -0,0 +1,45 @@
+# modules/response_generator.py
+from .ai_model import AIModel
+from .knowledge_base import KnowledgeBase
+
+class ResponseGenerator:
+    def __init__(self, ai_model: AIModel, knowledge_base: KnowledgeBase):
+        self.ai_model = ai_model
+        self.kb = knowledge_base
+
+    def generate(self, user_message: str, session_state: dict) -> str:
+        # 1. 优先使用 RAG (检索增强生成)
+        # 我们用目的地名称来强化检索查询
+        search_query = user_message
+        if session_state.get("destination"):
+            search_query += f" {session_state['destination']['name']}"
+            
+        relevant_knowledge = self.kb.search(search_query)
+        if relevant_knowledge:
+            context = self._format_knowledge_context(relevant_knowledge)
+            return self.ai_model.generate(user_message, context)
+
+        # 2. 如果没有知识库匹配，则使用基于规则的引导式对话
+        if not session_state.get("destination"):
+            return "听起来很棒！你想去欧洲的哪个城市呢？比如巴黎, 罗马, 巴塞罗那？"
+        if not session_state.get("duration"):
+            return f"好的，{session_state['destination']['name']}是个很棒的选择！你计划玩几天呢？"
+        if not session_state.get("persona"):
+            return "最后一个问题，这次旅行对你来说什么最重要呢？（例如：美食、艺术、购物、历史）"
+        
+        # 3. 如果信息都收集全了，但没触发RAG，让Gemma生成一个通用计划
+        plan_prompt = (
+            f"请为用户生成一个在 {session_state['destination']['name']} 的 "
+            f"{session_state['duration']['days']} 天旅行计划。"
+            f"旅行风格侧重于: {session_state['persona']['description']}。"
+        )
+        return self.ai_model.generate(plan_prompt, context="用户需要一个详细的旅行计划。")
+
+    def _format_knowledge_context(self, knowledge_items: list) -> str:
+        if not knowledge_items: return "没有特定的背景知识。"
+        # 简化处理，只用最相关的一条知识
+        item = knowledge_items[0]['knowledge']['travel_knowledge']
+        context = f"相关知识：\n- 目的地: {item['destination_info']['primary_destinations']}\n"
+        context += f"- 推荐天数: {item['destination_info']['recommended_duration']}天\n"
+        context += f"- 专业见解: {item['professional_insights']['key_takeaways']}\n"
+        return context

modules/session_manager.py

@@ -0,0 +1,34 @@
+# modules/session_manager.py
+import uuid
+from typing import Dict, Any
+
+class SessionManager:
+    def __init__(self):
+        self.sessions: Dict[str, Dict[str, Any]] = {}
+
+    def get_or_create_session(self, session_id: str = None) -> Dict[str, Any]:
+        if not session_id or session_id not in self.sessions:
+            session_id = str(uuid.uuid4())[:8]
+            self.sessions[session_id] = {
+                "session_id": session_id,
+                "destination": None,
+                "duration": None,
+                "persona": None,
+                "stage": "greeting" # 对话状态机
+            }
+        return self.sessions[session_id]
+
+    def update_session(self, session_id: str, updates: Dict[str, Any]):
+        if session_id in self.sessions:
+            self.sessions[session_id].update(updates)
+
+    def format_session_info(self, session_state: dict) -> str:
+        parts = [f"ID: {session_state.get('session_id', 'N/A')}"]
+        if session_state.get('destination'): parts.append(f"目的地: {session_state['destination']['name']}")
+        if session_state.get('duration'): parts.append(f"天数: {session_state['duration']['days']}")
+        if session_state.get('persona'): parts.append(f"风格: {session_state['persona']['name']}")
+        return " | ".join(parts)
+
+    def reset(self, session_id: str):
+        if session_id in self.sessions:
+            del self.sessions[session_id]

modules/travel_assistant.py

@@ -0,0 +1,45 @@
+# modules/travel_assistant.py
+from .config_loader import ConfigLoader
+from .ai_model import AIModel
+from .knowledge_base import KnowledgeBase
+from .info_extractor import InfoExtractor
+from .session_manager import SessionManager
+from .response_generator import ResponseGenerator
+from utils.logger import log
+
+class TravelAssistant:
+    def __init__(self):
+        # 依赖注入：在这里实例化所有需要的模块
+        log.info("开始初始化 Travel Assistant 核心模块...")
+        self.config = ConfigLoader()
+        self.kb = KnowledgeBase()
+        self.ai_model = AIModel()
+        self.session_manager = SessionManager()
+        self.info_extractor = InfoExtractor(self.config)
+        self.response_generator = ResponseGenerator(self.ai_model, self.kb)
+        log.info("✅ Travel Assistant 核心模块全部初始化完成！")
+
+    def chat(self, message: str, session_id: str, history: list):
+        # 1. 获取或创建会话
+        session_state = self.session_manager.get_or_create_session(session_id)
+        current_session_id = session_state['session_id']
+
+        # 2. 从用户输入中提取信息
+        extracted_info = self.info_extractor.extract(message)
+        
+        # 3. 更新会话状态
+        if extracted_info:
+            self.session_manager.update_session(current_session_id, extracted_info)
+            # 重新获取更新后的状态
+            session_state = self.session_manager.get_or_create_session(current_session_id)
+
+        # 4. 生成回复
+        bot_response = self.response_generator.generate(message, session_state)
+        
+        # 5. 格式化状态信息用于前端显示
+        status_info = self.session_manager.format_session_info(session_state)
+        
+        # 6. 更新对话历史
+        new_history = history + [[message, bot_response]]
+        
+        return bot_response, current_session_id, status_info, new_history