# modules/knowledge_base/generator.py from typing import List, Dict, Generator, Union, Optional, Any import requests import os import json import time import re from dotenv import load_dotenv load_dotenv() class Generator: def __init__(self, subject="", instructor=""): # Set defaults if not provided self.subject = subject or "通用学科" self.instructor = instructor or "教师" # 基础API配置 self.api_key = os.getenv("API_KEY") self.api_base = os.getenv("BASE_URL") # 流式文本问答配置 self.stream_api_key = os.getenv("STREAM_API_KEY") self.stream_api_base = os.getenv("STREAM_BASE_URL") self.stream_model = os.getenv("STREAM_MODEL") # 通用提示词模板 - 填充主题和讲师信息 self.system_prompt = self.get_system_prompt_template().format( subject=self.subject, instructor=self.instructor ) def get_system_prompt_template(self): """返回可定制的系统提示词模板""" return """ 你是一位{subject}课程的智能助教，由{instructor}指导开发。你的目标是帮助学生理解和掌握{subject}课程的关键概念、原理和方法。 你拥有《{subject}》课程的专业知识库，包含教材内容、课件、习题解析等材料。当回答问题时，你应该优先使用知识库中检索到的相关内容，而不是依赖你的通用知识。 作为{subject}助教，你应该: 1. 用专业且易于理解的方式解释复杂概念 2. 提供准确的技术信息和计算示例 3. 在适当时使用比喻或类比帮助理解 4. 引导学生思考而不是直接给出所有答案 5. ⁠提供进一步学习的建议和资源 当回答问题时，请遵循以下原则： 1. 先从知识库中检索与问题最相关的内容 2. 将检索结果整合成连贯、清晰的回答 3. 保持学术严谨性，确保概念解释和计算过程准确无误 4. 使用专业术语的同时，确保解释足够通俗易懂 5. 回答问题时注明知识来源，例如"根据教材第X章..." 6. 当遇到计算题时，展示完整的计算步骤和思路 7. 当知识库中没有直接相关内容时，明确告知学生并提供基于可靠原理的解答 8. 对于概念性问题，先给出简短定义，再补充详细解释和例子 对于不同类型的问题，采用不同的回答格式： 1. 概念解释类问题： - 先给出简明定义 - 提供详细解释 - 举例说明 - 补充相关知识点连接 2. 计算类问题： - 明确列出已知条件和所求内容 - 说明解题思路和所用公式 - 展示详细计算步骤 - 给出最终答案并解释其含义 3. 综合分析类问题： - 分点阐述相关知识点 - 提供分析框架 - 给出结论和建议 当检索到视频内容时，你应该： 1. 明确告知用户你找到了相关视频资源 2. 提供视频链接并确保包含时间戳 3. 简要描述视频内容和主要学习点 4. 建议用户观看视频以获得可视化理解 5. 在回答结束时，再次强调视频资源的价值 对于所有包含视频链接的回答，必须以下述格式呈现视频资源： 推荐学习资源： [视频标题] - [视频链接] """ def get_tool_selection_template(self): """返回工具选择提示词模板""" return """ 你是{subject}课程智能助教系统的决策组件。你的唯一任务是判断用户问题类型并决定是否调用知识库工具，以及提取精准的搜索关键词。 对于所有涉及{subject}专业知识的问题，必须调用至少一个知识库工具。系统依赖这些知识库提供准确信息，而不是依赖模型的通用知识。 判断标准： 1. 所有涉及课程概念、原理、计算方法的问题 → 必须调用相关知识库 2. 所有需要专业解释或例子的问题 → 必须调用相关知识库 3. 所有学习指导或复习相关的问题 → 必须调用相关知识库 4. 仅对于纯粹的问候语或与课程无关的闲聊 → 不调用任何工具 当决定调用知识库时，提取的关键词必须满足以下条件： 1. 准确反映问题的核心主题 2. 包含{subject}相关的专业术语或技术名词 3. 去除无关紧要的修饰词 4. 每个关键词尽量简洁，优先使用专业术语 5. 提供2-5个关键词，确保覆盖问题的核心概念 例1：问题 - "请详细解释相关概念的表示方法和计算过程。" 判断：需要专业知识解释 工具选择：教材知识库（包含基础概念和详细解释） 关键词：["表示方法", "计算过程"] 例2：问题 - "这个概念的工作原理是什么？能给我一个直观的例子吗？" 判断：需要专业知识解释，且需要直观演示 工具选择：教材知识库（基础概念）和视频知识库（直观演示） 关键词：["工作原理", "例子"] 例3：问题 - "你好，今天天气怎么样？" 判断：与课程无关的闲聊 工具选择：不调用任何工具 关键词：[] 以下情况应优先选择视频知识库工具： 1. 用户明确要求视频讲解或视频资料 2. 问题涉及复杂的步骤或流程，可能需要可视化展示 3. 问题关于动态过程的理解 4. 问题涉及图形或结构的理解 5. 问题包含"演示"、"展示"、"直观"、"可视化"等类似于需求的词语 以下情况应选择习题知识库工具： 1. 用户明确询问习题、例题或解题方法 2. 问题是关于如何解决特定类型的问题 3. 用户寻求考试或作业的帮助 4. 用户提出的问题形式类似于典型习题 你不需要回答用户问题，只需决定调用哪些工具以及提供精准的关键词数组。 """ def get_code_execution_prompt_template(self): """返回代码执行插件的提示词模板""" return """ 只要当用户询问编程、代码或特别是Python相关的问题时，你必须在回答中整合代码执行插件的使用。 使用代码执行插件的指南： 1. 创建Python代码示例时，请使用正确的Markdown语法，用```python和```作为代码块的分隔符。 2. 确保你的代码示例完整、可运行，并附有适当的注释。 3. 在代码前后提供解释，帮助用户理解代码的功能和原理。 4. 当代码与用户问题相关时，明确告知用户可以使用代码执行环境运行这段代码。 5. 对于教学场景，考虑创建循序渐进的代码示例，让用户可以逐步学习和理解。 示例回答格式： "这是一个[描述]的Python程序： ```python # 你的完整、可运行的代码 print('Hello, world!') ``` 你可以通过点击'运行'按钮在代码执行环境中运行这段代码。 如果你想修改代码，只需在编辑器中编辑并再次运行即可。" """ def get_visualization_prompt_template(self): """返回可视化插件的提示词模板""" return """ 当用户询问有关数学图形、函数、几何或需要3D可视化的内容时，你应该在回答中提供一个完整的Python函数来生成3D图形。 使用3D可视化插件的指南： 1. 你必须提供一个名为create_3d_plot的Python函数，该函数不接受任何参数。 2. 这个函数应该导入必要的库（主要是numpy as np）。 3. 函数需要返回一个包含以下结构的字典： { 'x': x_data, 'y': y_data, 'z': z_data, 'type': 'surface' 或 'scatter3d' (取决于数据类型) } 4. 确保你的代码可以直接运行，无需额外修改。 示例回答格式： "下面是[数学概念]的3D可视化函数： ```python import numpy as np def create_3d_plot(): # 生成数据 x = np.linspace(-5, 5, 100) y = np.linspace(-5, 5, 100) X, Y = np.meshgrid(x, y) Z = np.sin(np.sqrt(X**2 + Y**2)) return { 'x': X.tolist(), 'y': Y.tolist(), 'z': Z.tolist(), 'type': 'surface' } ``` 这个函数创建了[概念描述]的3D图形，你可以观察[关键特征]。" """ def get_mindmap_prompt_template(self): """返回思维导图插件的提示词模板""" return """ 当用户需要组织和梳理知识结构、概念关系或学习规划时，你应该在回答中整合思维导图。 使用思维导图的指南： 1. 提供一个完整的思维导图结构，使用PlantUML格式。 2. 使用@startmindmap和@endmindmap标记包裹内容。 3. 使用星号(*)表示层级：*为中央主题，**为主要主题，***为子主题，****为叶子节点。 4. 确保思维导图结构清晰、逻辑合理，能够帮助用户理解知识体系。 示例格式： @startmindmap * 中心主题 ** 主要分支1 *** 子主题1.1 **** 叶子节点1.1.1 *** 子主题1.2 ** 主要分支2 *** 子主题2.1 @endmindmap 确保思维导图涵盖主题的关键概念和它们之间的关系，帮助用户建立完整的知识体系。 """ def extract_keywords_with_tools(self, question: str, tools: List[Dict]) -> List[Dict]: """使用工具化架构分析问题，决定使用哪些知识库以及提取关键词""" system_prompt = self.get_tool_selection_template().format(subject=self.subject) headers = { "Authorization": f"Bearer {self.stream_api_key}", "Content-Type": "application/json" } response = requests.post( f"{self.stream_api_base}/chat/completions", headers=headers, json={ "model": self.stream_model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": question} ], "tools": tools, "tool_choice": "auto" } ) if response.status_code != 200: raise Exception(f"工具调用出错: {response.text}") response_data = response.json() message = response_data["choices"][0]["message"] # 如果模型决定调用工具 if "tool_calls" in message and message["tool_calls"]: return message["tool_calls"] else: # 模型没有调用工具，返回空列表 return [] def generate_stream(self, query: str, context_docs: List[Dict], process_data: Optional[Dict] = None) -> Generator[Union[str, Dict], None, None]: """流式生成回答 - 用于所有类型的问答""" start_time = time.time() # 构建带有引用标记的上下文 context_with_refs = [] # 检查是否有图片URL has_images = any(doc['metadata'].get('img_url', '') for doc in context_docs) # 检查查询是否与特定插件相关 is_code_related = any(kw in query.lower() for kw in ['code', 'python', 'program', '代码', '编程', 'coding', 'script']) is_visualization_related = any(kw in query.lower() for kw in ['3d', 'graph', 'plot', 'function', 'visualization', '可视化', '图形', '函数']) is_mindmap_related = any(kw in query.lower() for kw in ['mindmap', 'mind map', 'concept map', '思维导图', '概念图', '知识图']) # 处理每个文档 for i, doc in enumerate(context_docs, 1): # 直接使用文件名作为来源 file_name = doc['metadata'].get('file_name', '未知文件') img_url = doc['metadata'].get('img_url', '') # 构建文档内容，如果有图片URL则包含在内 content = doc['content'] if img_url: content += f"\n[图片地址: {img_url}]" context_with_refs.append(f"[{i}] {content}\n来源：{file_name}") context = "\n\n".join(context_with_refs) # 增强系统提示词，根据查询内容添加插件特定提示 enhanced_system_prompt = self.system_prompt # 如果包含图片，添加图片处理相关指令 if has_images: enhanced_system_prompt += """ 此外，如果参考内容中包含图片地址，请在回答中适当引用这些图片信息，并在回答的最后列出所有参考的图片来源。 """ # 添加插件特定提示 if is_code_related: enhanced_system_prompt += "\n\n" + self.get_code_execution_prompt_template() if is_visualization_related: enhanced_system_prompt += "\n\n" + self.get_visualization_prompt_template() if is_mindmap_related: enhanced_system_prompt += "\n\n" + self.get_mindmap_prompt_template() # 流式API headers = { "Authorization": f"Bearer {self.stream_api_key}", "Content-Type": "application/json" } try: response = requests.post( f"{self.stream_api_base}/chat/completions", headers=headers, json={ "model": self.stream_model, "messages": [ {"role": "system", "content": enhanced_system_prompt}, {"role": "user", "content": f""" 参考内容： {context} 问题：{query} 请按照要求回答问题，包括引用标注和来源列表。 如果在参考内容中找到视频资源，请使用以下格式标记视频链接： 视频链接 在回答结束时，请使用以下固定格式列出所有获取的参考内容作为参考来源： ===参考来源开始=== [1] 摘要内容，"文件名" [2] 摘要内容，"文件名" ===参考来源结束=== 如果没有参考内容，则无需包含上述部分。 """} ], "stream": True } ) if response.status_code != 200: yield f"生成回答时出错: {response.text}" return # 返回流式响应 for line in response.iter_lines(): if not line: continue line_text = line.decode('utf-8') if line_text.startswith('data: ') and line_text != 'data: [DONE]': try: json_str = line_text[6:] # 移除 "data: " 前缀 data = json.loads(json_str) content = data.get('choices', [{}])[0].get('delta', {}).get('content', '') if content: yield content except Exception as e: yield f"解析响应出错: {str(e)}" # 如果需要返回处理数据 if process_data: process_data["generation"]["time"] = round(time.time() - start_time, 3) yield {"process_data": process_data} except Exception as e: yield f"连接错误: {str(e)}"