# routes/TeachingModelController.py import json import subprocess import tempfile import time import urllib import uuid from io import BytesIO from fastapi import APIRouter, Depends from sse_starlette import EventSourceResponse from starlette.responses import StreamingResponse from auth.dependencies import * from utils.LightRagUtil import * from utils.PageUtil import * from utils.ParseRequest import * from lightrag import * # 创建一个路由实例,需要依赖get_current_user,登录后才能访问 router = APIRouter(dependencies=[Depends(get_current_user)]) rag_type: str = "file" # rag_type: str = "pg" # 【TeachingModel-1】获取主题列表 @router.get("/getTrainedTheme") async def get_trained_theme(request: Request): # 获取参数 bureau_id = await get_request_str_param(request, "bureau_id", True, True) stage_id = await get_request_num_param(request, "stage_id", True, True, None) subject_id = await get_request_num_param(request, "subject_id", True, True, None) page_number = await get_request_num_param(request, "page_number", False, True, 1) page_size = await get_request_num_param(request, "page_size", False, True, 10) # 数据库查询 select_trained_theme_sql: str = f"SELECT * FROM t_ai_teaching_model_theme WHERE is_deleted = 0 and search_flag = 1 AND bureau_id = '{bureau_id}' AND stage_id = {stage_id} AND subject_id = {subject_id}" print(select_trained_theme_sql) page = await get_page_data_by_sql(select_trained_theme_sql, page_number, page_size) page = await translate_person_bureau_name(page) # 结果返回 return {"success": True, "message": "查询成功！", "data": page} # 【TeachingModel-2】获取热门主题列表 @router.get("/getHotTheme") async def get_hot_theme(request: Request): # 获取参数 bureau_id = await get_request_str_param(request, "bureau_id", True, True) page_number = await get_request_num_param(request, "page_number", False, True, 1) page_size = await get_request_num_param(request, "page_size", False, True, 3) # 数据库查询 select_hot_theme_sql: str = f"SELECT * FROM t_ai_teaching_model_theme WHERE is_deleted = 0 and search_flag = 1 and bureau_id = '{bureau_id}' ORDER BY quote_count DESC" print(select_hot_theme_sql) page = await get_page_data_by_sql(select_hot_theme_sql, page_number, page_size) page = await translate_person_bureau_name(page) # 结果返回 return {"success": True, "message": "查询成功！", "data": page} # 【TeachingModel-3】获取最新主题列表 @router.get("/getNewTheme") async def get_new_theme(request: Request): bureau_id = await get_request_str_param(request, "bureau_id", True, True) page_number = await get_request_num_param(request, "page_number", False, True, 1) page_size = await get_request_num_param(request, "page_size", False, True, 3) # 数据库查询 select_new_theme_sql: str = f"SELECT * FROM t_ai_teaching_model_theme WHERE is_deleted = 0 and search_flag = 1 and bureau_id = '{bureau_id}' ORDER BY create_time DESC" print(select_new_theme_sql) page = await get_page_data_by_sql(select_new_theme_sql, page_number, page_size) page = await translate_person_bureau_name(page) # 结果返回 return {"success": True, "message": "查询成功！", "data": page} # 【TeachingModel-4】获取问题列表 @router.get("/getQuestion") async def get_question(request: Request): # 获取参数 bureau_id = await get_request_str_param(request, "bureau_id", True, True) person_id = await get_request_str_param(request, "person_id", True, True) theme_id = await get_request_num_param(request, "theme_id", True, True, None) question_type = await get_request_num_param(request, "question_type", True, True, None) page_number = await get_request_num_param(request, "page_number", False, True, 1) page_size = await get_request_num_param(request, "page_size", False, True, 10) person_sql = "" if question_type == 2: person_sql = f"AND person_id = '{person_id}'" # 数据库查询 select_question_sql: str = f"SELECT * FROM t_ai_teaching_model_question WHERE is_deleted = 0 and bureau_id = '{bureau_id}' AND theme_id = {theme_id} AND question_type = {question_type} {person_sql}" print(select_question_sql) page = await get_page_data_by_sql(select_question_sql, page_number, page_size) return {"success": True, "message": "查询成功！", "data": page} # 【TeachingModel-5】提问 @router.post("/sendQuestion") async def send_question(request: Request): # 获取参数 bureau_id = await get_request_str_param(request, "bureau_id", True, True) person_id = await get_request_str_param(request, "person_id", True, True) theme_id = await get_request_num_param(request, "theme_id", True, True, None) question = await get_request_str_param(request, "question", True, True) question_type = await get_request_num_param(request, "question_type", False, True, 0) theme_object = await find_by_id("t_ai_teaching_model_theme", "id", theme_id) if theme_object is None: return {"success": False, "message": "主题不存在！"} if question_type == 1: # 处理常见问题引用次数 update_common_question_sql: str = f"update t_ai_teaching_model_question set quote_count = quote_count + 1 where question_type = 1 and theme_id = {theme_id} and question = '{question}' and is_deleted = 0" await execute_sql(update_common_question_sql, ()) elif question_type == 2: # 处理个人历史问题引用次数 update_person_question_sql: str = f"update t_ai_teaching_model_question set quote_count = quote_count + 1 where question_type = 2 and theme_id = {theme_id} and question = '{question}' and person_id = '{person_id}' and is_deleted = 0" await execute_sql(update_person_question_sql, ()) else: # 新问题，保存个人历史问题 param = {} param["stage_id"] = int(theme_object["stage_id"]) param["subject_id"] = int(theme_object["subject_id"]) param["theme_id"] = theme_id param["question"] = question param["question_type"] = 2 param["quote_count"] = 0 param["question_person_id"] = person_id param["person_id"] = person_id param["bureau_id"] = bureau_id question_id = await insert("t_ai_teaching_model_question", param) # 处理theme的调用次数 update_sql: str = f"UPDATE t_ai_teaching_model_theme SET quote_count = quote_count + 1, update_time = now() WHERE id = {theme_id}" await execute_sql(update_sql, ()) # 向rag提问 topic = theme_object["short_name"] # mode = "hybrid" prompt = "\n 1、不要输出参考资料 或者 References ！" prompt = prompt + "\n 2、资料中提供化学反应方程式的，一定要严格按提供的Latex公式输出，绝对不允许对Latex公式进行修改 ！" prompt = prompt + "\n 3、如果资料中提供了图片的，一定要严格按照原文提供图片输出，绝对不能省略或不输出！" prompt = prompt + "\n 4、知识库中存在的问题，严格按知识库中的内容回答，不允许扩展！" prompt = prompt + "\n 5、如果问题与提供的知识库内容不符，则明确告诉未在知识库范围内提到！" prompt = prompt + "\n 6、发现输出内容中包含Latex公式的，一定要检查是不是包含了$$或$的包含符号,不能让Latex无包含符号出现！" WORKING_PATH = "./Topic/" + topic if rag_type == "file": async def generate_response_stream(query: str, mode: str, user_prompt: str): try: rag = await initialize_rag(WORKING_PATH) resp = await rag.aquery( query=query, param=QueryParam(mode=mode, stream=True, user_prompt=user_prompt, enable_rerank=True)) async for chunk in resp: if not chunk: continue yield f"data: {json.dumps({'reply': chunk})}\n\n" print(chunk, end='', flush=True) except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" finally: # 清理资源 await rag.finalize_storages() return EventSourceResponse(generate_response_stream(query=question, mode="hybrid", user_prompt=prompt)) elif rag_type == "pg": workspace = theme_object["short_name"] # 使用PG库后，这个是没有用的,但目前的项目代码要求必传，就写一个吧。 WORKING_DIR = 'WorkingPath/' + workspace if not os.path.exists(WORKING_DIR): os.makedirs(WORKING_DIR) async def generate_response_stream(query: str, mode: str, user_prompt: str): try: logger.info("workspace=" + workspace) rag = await initialize_pg_rag(WORKING_DIR=WORKING_DIR, workspace=workspace) resp = await rag.aquery( query=query, param=QueryParam(mode=mode, stream=True, user_prompt=user_prompt)) async for chunk in resp: if not chunk: continue yield f"data: {json.dumps({'reply': chunk})}\n\n" print(chunk, end='', flush=True) except Exception as e: yield f"data: {json.dumps({'error': str(e)})}\n\n" finally: # 发送流结束标记 yield "data: [DONE]\n\n" # 清理资源 await rag.finalize_storages() return EventSourceResponse(generate_response_stream(query=question, mode="hybrid", user_prompt=prompt)) @router.post("/saveWord") async def save_word(request: Request): # 获取参数 theme_id = await get_request_num_param(request, "theme_id", True, True, None) markdown_content = await get_request_str_param(request, "markdown_content", True, True) question = await get_request_str_param(request, "question", True, True) theme_object = await find_by_id("t_ai_teaching_model_theme", "id", theme_id) if theme_object is None: return {"success": False, "message": "主题不存在！"} filename = "【理想大模型】" + str(theme_object["theme_name"]) + "(" + str(question) + ")" + str(time.time()) + ".docx" print(filename) output_file = None try: # 创建临时Markdown文件 temp_md = os.path.join(tempfile.gettempdir(), uuid.uuid4().hex + ".md") with open(temp_md, "w", encoding="utf-8") as f: f.write(markdown_content) # 使用pandoc转换 output_file = os.path.join(tempfile.gettempdir(), filename) subprocess.run(['pandoc', temp_md, '-o', output_file, '--resource-path=static'], check=True) # 读取生成的Word文件 with open(output_file, "rb") as f: stream = BytesIO(f.read()) # 返回响应 encoded_filename = urllib.parse.quote(filename) return StreamingResponse( stream, media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", headers={"Content-Disposition": f"attachment; filename*=UTF-8'{encoded_filename}'"}) except HTTPException: raise except Exception as e: logger.error(f"Unexpected error: {str(e)}") raise HTTPException(status_code=500, detail="Internal server error") finally: ... # 清理临时文件 try: if temp_md and os.path.exists(temp_md): os.remove(temp_md) if output_file and os.path.exists(output_file): os.remove(output_file) except Exception as e: logger.warning(f"Failed to clean up temp files: {str(e)}")