""" 生成集成模块 """ import os import logging from typing import List from langchain_core.prompts import ChatPromptTemplate, PromptTemplate from langchain_community.chat_models.moonshot import MoonshotChat from langchain_core.documents import Document from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser logger = logging.getLogger(__name__) class GenerationIntegrationModule: """生成集成模块 - 负责LLM集成和回答生成""" def __init__(self, model_name: str = "kimi-k2-0711-preview", temperature: float = 0.1, max_tokens: int = 2048): """ 初始化生成集成模块 Args: model_name: 模型名称 temperature: 生成温度 max_tokens: 最大token数 """ self.model_name = model_name self.temperature = temperature self.max_tokens = max_tokens self.llm = None self.setup_llm() def setup_llm(self): """初始化大语言模型""" logger.info(f"正在初始化LLM: {self.model_name}") api_key = os.getenv("MOONSHOT_API_KEY") if not api_key: raise ValueError("请设置 MOONSHOT_API_KEY 环境变量") self.llm = MoonshotChat( model=self.model_name, temperature=self.temperature, max_tokens=self.max_tokens, moonshot_api_key=api_key ) logger.info("LLM初始化完成") def generate_basic_answer(self, query: str, context_docs: List[Document]) -> str: """ 生成基础回答 Args: query: 用户查询 context_docs: 上下文文档列表 Returns: 生成的回答 """ context = self._build_context(context_docs) prompt = ChatPromptTemplate.from_template(""" 你是一位专业的烹饪助手。请根据以下食谱信息回答用户的问题。 用户问题: {question} 相关食谱信息: {context} 请提供详细、实用的回答。如果信息不足，请诚实说明。 回答:""") # 使用LCEL构建链 chain = ( {"question": RunnablePassthrough(), "context": lambda _: context} | prompt | self.llm | StrOutputParser() ) response = chain.invoke(query) return response def generate_step_by_step_answer(self, query: str, context_docs: List[Document]) -> str: """ 生成分步骤回答 Args: query: 用户查询 context_docs: 上下文文档列表 Returns: 分步骤的详细回答 """ context = self._build_context(context_docs) prompt = ChatPromptTemplate.from_template(""" 你是一位专业的烹饪导师。请根据食谱信息，为用户提供详细的分步骤指导。 用户问题: {question} 相关食谱信息: {context} 请灵活组织回答，建议包含以下部分（可根据实际内容调整）： ## 🥘 菜品介绍 [简要介绍菜品特点和难度] ## 🛒 所需食材 [列出主要食材和用量] ## 👨‍🍳 制作步骤 [详细的分步骤说明，每步包含具体操作和大概所需时间] ## 💡 制作技巧 [仅在有实用技巧时包含。优先使用原文中的实用技巧，如果原文的"附加内容"与烹饪无关或为空，可以基于制作步骤总结关键要点，或者完全省略此部分] 注意： - 根据实际内容灵活调整结构 - 不要强行填充无关内容或重复制作步骤中的信息 - 重点突出实用性和可操作性 - 如果没有额外的技巧要分享，可以省略制作技巧部分 回答:""") chain = ( {"question": RunnablePassthrough(), "context": lambda _: context} | prompt | self.llm | StrOutputParser() ) response = chain.invoke(query) return response def query_rewrite(self, query: str) -> str: """ 智能查询重写 - 让大模型判断是否需要重写查询 Args: query: 原始查询 Returns: 重写后的查询或原查询 """ prompt = PromptTemplate( template=""" 你是一个智能查询分析助手。请分析用户的查询，判断是否需要重写以提高食谱搜索效果。 原始查询: {query} 分析规则： 1. **具体明确的查询**（直接返回原查询）： - 包含具体菜品名称：如"宫保鸡丁怎么做"、"红烧肉的制作方法" - 明确的制作询问：如"蛋炒饭需要什么食材"、"糖醋排骨的步骤" - 具体的烹饪技巧：如"如何炒菜不粘锅"、"怎样调制糖醋汁" 2. **模糊不清的查询**（需要重写）： - 过于宽泛：如"做菜"、"有什么好吃的"、"推荐个菜" - 缺乏具体信息：如"川菜"、"素菜"、"简单的" - 口语化表达：如"想吃点什么"、"有饮品推荐吗" 重写原则： - 保持原意不变 - 增加相关烹饪术语 - 优先推荐简单易做的 - 保持简洁性 示例： - "做菜" → "简单易做的家常菜谱" - "有饮品推荐吗" → "简单饮品制作方法" - "推荐个菜" → "简单家常菜推荐" - "川菜" → "经典川菜菜谱" - "宫保鸡丁怎么做" → "宫保鸡丁怎么做"（保持原查询） - "红烧肉需要什么食材" → "红烧肉需要什么食材"（保持原查询） 请输出最终查询（如果不需要重写就返回原查询）:""", input_variables=["query"] ) chain = ( {"query": RunnablePassthrough()} | prompt | self.llm | StrOutputParser() ) response = chain.invoke(query).strip() # 记录重写结果 if response != query: logger.info(f"查询已重写: '{query}' → '{response}'") else: logger.info(f"查询无需重写: '{query}'") return response def query_router(self, query: str) -> str: """ 查询路由 - 根据查询类型选择不同的处理方式 Args: query: 用户查询 Returns: 路由类型 ('list', 'detail', 'general') """ prompt = ChatPromptTemplate.from_template(""" 根据用户的问题，将其分类为以下三种类型之一： 1. 'list' - 用户想要获取菜品列表或推荐，只需要菜名 例如：推荐几个素菜、有什么川菜、给我3个简单的菜 2. 'detail' - 用户想要具体的制作方法或详细信息 例如：宫保鸡丁怎么做、制作步骤、需要什么食材 3. 'general' - 其他一般性问题 例如：什么是川菜、制作技巧、营养价值 请只返回分类结果：list、detail 或 general 用户问题: {query} 分类结果:""") chain = ( {"query": RunnablePassthrough()} | prompt | self.llm | StrOutputParser() ) result = chain.invoke(query).strip().lower() # 确保返回有效的路由类型 if result in ['list', 'detail', 'general']: return result else: return 'general' # 默认类型 def generate_list_answer(self, query: str, context_docs: List[Document]) -> str: """ 生成列表式回答 - 适用于推荐类查询 Args: query: 用户查询 context_docs: 上下文文档列表 Returns: 列表式回答 """ if not context_docs: return "抱歉，没有找到相关的菜品信息。" # 提取菜品名称 dish_names = [] for doc in context_docs: dish_name = doc.metadata.get('dish_name', '未知菜品') if dish_name not in dish_names: dish_names.append(dish_name) # 构建简洁的列表回答 if len(dish_names) == 1: return f"为您推荐：{dish_names[0]}" elif len(dish_names) <= 3: return f"为您推荐以下菜品：\n" + "\n".join([f"{i+1}. {name}" for i, name in enumerate(dish_names)]) else: return f"为您推荐以下菜品：\n" + "\n".join([f"{i+1}. {name}" for i, name in enumerate(dish_names[:3])]) + f"\n\n还有其他 {len(dish_names)-3} 道菜品可供选择。" def generate_basic_answer_stream(self, query: str, context_docs: List[Document]): """ 生成基础回答 - 流式输出 Args: query: 用户查询 context_docs: 上下文文档列表 Yields: 生成的回答片段 """ context = self._build_context(context_docs) prompt = ChatPromptTemplate.from_template(""" 你是一位专业的烹饪助手。请根据以下食谱信息回答用户的问题。 用户问题: {question} 相关食谱信息: {context} 请提供详细、实用的回答。如果信息不足，请诚实说明。 回答:""") chain = ( {"question": RunnablePassthrough(), "context": lambda _: context} | prompt | self.llm | StrOutputParser() ) for chunk in chain.stream(query): yield chunk def generate_step_by_step_answer_stream(self, query: str, context_docs: List[Document]): """ 生成详细步骤回答 - 流式输出 Args: query: 用户查询 context_docs: 上下文文档列表 Yields: 详细步骤回答片段 """ context = self._build_context(context_docs) prompt = ChatPromptTemplate.from_template(""" 你是一位专业的烹饪导师。请根据食谱信息，为用户提供详细的分步骤指导。 用户问题: {question} 相关食谱信息: {context} 请灵活组织回答，建议包含以下部分（可根据实际内容调整）： ## 🥘 菜品介绍 [简要介绍菜品特点和难度] ## 🛒 所需食材 [列出主要食材和用量] ## 👨‍🍳 制作步骤 [详细的分步骤说明，每步包含具体操作和大概所需时间] ## 💡 制作技巧 [仅在有实用技巧时包含。如果原文的"附加内容"与烹饪无关或为空，可以基于制作步骤总结关键要点，或者完全省略此部分] 注意： - 根据实际内容灵活调整结构 - 不要强行填充无关内容 - 重点突出实用性和可操作性 回答:""") chain = ( {"question": RunnablePassthrough(), "context": lambda _: context} | prompt | self.llm | StrOutputParser() ) for chunk in chain.stream(query): yield chunk def _build_context(self, docs: List[Document], max_length: int = 2000) -> str: """ 构建上下文字符串 Args: docs: 文档列表 max_length: 最大长度 Returns: 格式化的上下文字符串 """ if not docs: return "暂无相关食谱信息。" context_parts = [] current_length = 0 for i, doc in enumerate(docs, 1): # 添加元数据信息 metadata_info = f"【食谱 {i}】" if 'dish_name' in doc.metadata: metadata_info += f" {doc.metadata['dish_name']}" if 'category' in doc.metadata: metadata_info += f" | 分类: {doc.metadata['category']}" if 'difficulty' in doc.metadata: metadata_info += f" | 难度: {doc.metadata['difficulty']}" # 构建文档文本 doc_text = f"{metadata_info}\n{doc.page_content}\n" # 检查长度限制 if current_length + len(doc_text) > max_length: break context_parts.append(doc_text) current_length += len(doc_text) return "\n" + "="*50 + "\n".join(context_parts)