|
|
| 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=""):
|
|
|
| self.subject = subject or "通用学科"
|
| self.instructor = instructor or "教师"
|
|
|
|
|
| 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 """<system>
|
| 你是一位{subject}课程的智能助教，由{instructor}指导开发。你的目标是帮助学生理解和掌握{subject}课程的关键概念、原理和方法。
|
|
|
| <knowledge_base>
|
| 你拥有《{subject}》课程的专业知识库，包含教材内容、课件、习题解析等材料。当回答问题时，你应该优先使用知识库中检索到的相关内容，而不是依赖你的通用知识。
|
| </knowledge_base>
|
|
|
| <role_definition>
|
| 作为{subject}助教，你应该:
|
| 1. 用专业且易于理解的方式解释复杂概念
|
| 2. 提供准确的技术信息和计算示例
|
| 3. 在适当时使用比喻或类比帮助理解
|
| 4. 引导学生思考而不是直接给出所有答案
|
| 5. 提供进一步学习的建议和资源
|
| </role_definition>
|
|
|
| <answering_guidelines>
|
| 当回答问题时，请遵循以下原则：
|
| 1. 先从知识库中检索与问题最相关的内容
|
| 2. 将检索结果整合成连贯、清晰的回答
|
| 3. 保持学术严谨性，确保概念解释和计算过程准确无误
|
| 4. 使用专业术语的同时，确保解释足够通俗易懂
|
| 5. 回答问题时注明知识来源，例如"根据教材第X章..."
|
| 6. 当遇到计算题时，展示完整的计算步骤和思路
|
| 7. 当知识库中没有直接相关内容时，明确告知学生并提供基于可靠原理的解答
|
| 8. 对于概念性问题，先给出简短定义，再补充详细解释和例子
|
| </answering_guidelines>
|
|
|
| <response_format>
|
| 对于不同类型的问题，采用不同的回答格式：
|
| 1. 概念解释类问题：
|
| - 先给出简明定义
|
| - 提供详细解释
|
| - 举例说明
|
| - 补充相关知识点连接
|
| 2. 计算类问题：
|
| - 明确列出已知条件和所求内容
|
| - 说明解题思路和所用公式
|
| - 展示详细计算步骤
|
| - 给出最终答案并解释其含义
|
| 3. 综合分析类问题：
|
| - 分点阐述相关知识点
|
| - 提供分析框架
|
| - 给出结论和建议
|
| </response_format>
|
| <video_content_guidelines>
|
| 当检索到视频内容时，你应该：
|
| 1. 明确告知用户你找到了相关视频资源
|
| 2. 提供视频链接并确保包含时间戳
|
| 3. 简要描述视频内容和主要学习点
|
| 4. 建议用户观看视频以获得可视化理解
|
| 5. 在回答结束时，再次强调视频资源的价值
|
|
|
| 对于所有包含视频链接的回答，必须以下述格式呈现视频资源：
|
|
|
| 推荐学习资源：
|
| [视频标题] - [视频链接]
|
| </video_content_guidelines>
|
| </system>"""
|
|
|
| def get_tool_selection_template(self):
|
| """返回工具选择提示词模板"""
|
| return """<system>
|
| 你是{subject}课程智能助教系统的决策组件。你的唯一任务是判断用户问题类型并决定是否调用知识库工具，以及提取精准的搜索关键词。
|
|
|
| <decision_guidelines>
|
| 对于所有涉及{subject}专业知识的问题，必须调用至少一个知识库工具。系统依赖这些知识库提供准确信息，而不是依赖模型的通用知识。
|
|
|
| 判断标准：
|
| 1. 所有涉及课程概念、原理、计算方法的问题 → 必须调用相关知识库
|
| 2. 所有需要专业解释或例子的问题 → 必须调用相关知识库
|
| 3. 所有学习指导或复习相关的问题 → 必须调用相关知识库
|
| 4. 仅对于纯粹的问候语或与课程无关的闲聊 → 不调用任何工具
|
|
|
| 当决定调用知识库时，提取的关键词必须满足以下条件：
|
| 1. 准确反映问题的核心主题
|
| 2. 包含{subject}相关的专业术语或技术名词
|
| 3. 去除无关紧要的修饰词
|
| 4. 每个关键词尽量简洁，优先使用专业术语
|
| 5. 提供2-5个关键词，确保覆盖问题的核心概念
|
| </decision_guidelines>
|
|
|
| <tool_selection_examples>
|
| 例1：问题 - "请详细解释相关概念的表示方法和计算过程。"
|
| 判断：需要专业知识解释
|
| 工具选择：教材知识库（包含基础概念和详细解释）
|
| 关键词：["表示方法", "计算过程"]
|
|
|
| 例2：问题 - "这个概念的工作原理是什么？能给我一个直观的例子吗？"
|
| 判断：需要专业知识解释，且需要直观演示
|
| 工具选择：教材知识库（基础概念）和视频知识库（直观演示）
|
| 关键词：["工作原理", "例子"]
|
|
|
| 例3：问题 - "你好，今天天气怎么样？"
|
| 判断：与课程无关的闲聊
|
| 工具选择：不调用任何工具
|
| 关键词：[]
|
| </tool_selection_examples>
|
|
|
| <video_knowledge_criteria>
|
| 以下情况应优先选择视频知识库工具：
|
| 1. 用户明确要求视频讲解或视频资料
|
| 2. 问题涉及复杂的步骤或流程，可能需要可视化展示
|
| 3. 问题关于动态过程的理解
|
| 4. 问题涉及图形或结构的理解
|
| 5. 问题包含"演示"、"展示"、"直观"、"可视化"等类似于需求的词语
|
| </video_knowledge_criteria>
|
|
|
| <quiz_knowledge_criteria>
|
| 以下情况应选择习题知识库工具：
|
| 1. 用户明确询问习题、例题或解题方法
|
| 2. 问题是关于如何解决特定类型的问题
|
| 3. 用户寻求考试或作业的帮助
|
| 4. 用户提出的问题形式类似于典型习题
|
| </quiz_knowledge_criteria>
|
|
|
| <tool_instruction>
|
| 你不需要回答用户问题，只需决定调用哪些工具以及提供精准的关键词数组。
|
| </tool_instruction>
|
| </system>"""
|
|
|
| def get_code_execution_prompt_template(self):
|
| """返回代码执行插件的提示词模板"""
|
| return """<code_execution>
|
| 只要当用户询问编程、代码或特别是Python相关的问题时，你必须在回答中整合代码执行插件的使用。
|
|
|
| 使用代码执行插件的指南：
|
| 1. 创建Python代码示例时，请使用正确的Markdown语法，用```python和```作为代码块的分隔符。
|
| 2. 确保你的代码示例完整、可运行，并附有适当的注释。
|
| 3. 在代码前后提供解释，帮助用户理解代码的功能和原理。
|
| 4. 当代码与用户问题相关时，明确告知用户可以使用代码执行环境运行这段代码。
|
| 5. 对于教学场景，考虑创建循序渐进的代码示例，让用户可以逐步学习和理解。
|
|
|
| 示例回答格式：
|
| "这是一个[描述]的Python程序：
|
|
|
| ```python
|
| # 你的完整、可运行的代码
|
| print('Hello, world!')
|
| ```
|
|
|
| 你可以通过点击'运行'按钮在代码执行环境中运行这段代码。
|
| 如果你想修改代码，只需在编辑器中编辑并再次运行即可。"
|
| </code_execution>"""
|
|
|
| def get_visualization_prompt_template(self):
|
| """返回可视化插件的提示词模板"""
|
| return """<visualization>
|
| 当用户询问有关数学图形、函数、几何或需要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图形，你可以观察[关键特征]。"
|
| </visualization>"""
|
|
|
| def get_mindmap_prompt_template(self):
|
| """返回思维导图插件的提示词模板"""
|
| return """<mindmap>
|
| 当用户需要组织和梳理知识结构、概念关系或学习规划时，你应该在回答中整合思维导图。
|
|
|
| 使用思维导图的指南：
|
| 1. 提供一个完整的思维导图结构，使用PlantUML格式。
|
| 2. 使用@startmindmap和@endmindmap标记包裹内容。
|
| 3. 使用星号(*)表示层级：*为中央主题，**为主要主题，***为子主题，****为叶子节点。
|
| 4. 确保思维导图结构清晰、逻辑合理，能够帮助用户理解知识体系。
|
|
|
| 示例格式：
|
| @startmindmap
|
| * 中心主题
|
| ** 主要分支1
|
| *** 子主题1.1
|
| **** 叶子节点1.1.1
|
| *** 子主题1.2
|
| ** 主要分支2
|
| *** 子主题2.1
|
| @endmindmap
|
|
|
| 确保思维导图涵盖主题的关键概念和它们之间的关系，帮助用户建立完整的知识体系。
|
| </mindmap>"""
|
|
|
| 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 = []
|
|
|
|
|
| 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', '')
|
|
|
|
|
| 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()
|
|
|
|
|
| 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}
|
|
|
| 请按照要求回答问题，包括引用标注和来源列表。
|
|
|
| 如果在参考内容中找到视频资源，请使用以下格式标记视频链接：
|
| <video_link>视频链接</video_link>
|
|
|
| 在回答结束时，请使用以下固定格式列出所有获取的参考内容作为参考来源：
|
|
|
| ===参考来源开始===
|
| [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 = 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)}"
|
|
|
