【AI大模型应用开发】【项目实战】23.RAG智慧问答项目-(十一)融合MySQL的RAG系统之融合FQA和知识库查询

一.融合FQA和知识库查询目标：理解智能问答系统的端到端工作流程掌握MySQL FAQ(高频问答) 和 RAG 系统的集成与协调学习如何通过日志记录和错误处理提升系统健壮性1.查询流程图old_main.py是一个融合 MySQL FAQ 和 RAG 系统的交互式问答程序，专注于快速、精准的答案生成,它通过MySQL 提供结构化数据的高效查询，通过RAG 系统处理复杂问题，同时配备日志记录和用户友好的命令行界面，适合项目场景的实时问答需求以下是智能问答系统的查询流程图，展示从用户输入到答案输出的完整逻辑2.流程说明1.输入处理：用户提供查询(query)和可选的学科过滤 (source_filter)2.BM25 搜索：使用BM25 算法在MySQL 知识库中搜索，设置相似度阈值 0.853.答案判断：若找到可靠答案（相似度 0.85），直接返回若无可靠答案且需要 RAG，调用 RAG 系统生成答案若无需 RAG，返回“未找到答案”4.日志记录：记录查询内容、答案和处理时间，便于调试和性能分析5.输出：将答案返回给用户3.代码介绍以下是old_main.py的完整代码，包含详细注释，逐行解析功能与实现逻辑思路:""" 需求：实现edu-rag项目 思路步骤： 1. 初始化：bm25search、大模型、rag_qa_system 2. 将用户query发送给FAQ模块进行处理，得到答案和是否进入RAG模块的标识 3. 如果答案可靠，直接返回 4. 如果答案不可靠，且需要进入RAG模块，调用rag模块，得到答案 5. 如果答案不可靠，且不需要进入RAG模块，直接走兜底逻辑 """(1).导入必备的工具包import time from langchain_core.messages import SystemMessage, HumanMessage # 导入 OpenAI 客户端，用于调用 DashScope API from langchain_openai import ChatOpenAI # 导入配置和日志工具，用于系统配置和日志记录 from base.logger import single_logger as logger from base.config import single_config as config # 导入 MySQL 系统组件，用于数据库操作和搜索 from mysql_qa.retreival.bm25_search import Bm25Search from mysql_qa.cache.redis_client import RedisClient from mysql_qa.db.mysql_client import MysqlClient # 导入 RAG 系统组件，用于知识库检索和答案生成 from rag_qa.core.rag_system import RagSystem(2).系统初始化__init__)初始化方法功能：初始化日志、配置、数据库客户端、搜索模块、向量存储和RAG系统关键点: 通过Config 管理 API 密钥和模型参数,异常处理确保OpenAI 客户端初始化成功代码如下:class IntegratedQASystem: # 1. 初始化：bm25search、大模型、rag_qa_system def __init__(self): # bm25search # 初始化配置对象，加载系统参数 self.config = config # 初始化日志工具，用于记录系统运行信息 self.logger = logger # 初始化 MySQL 客户端，用于数据库操作 self.db = MysqlClient() # 初始化 Redis 客户端，用于缓存管理 self.cache = RedisClient() # 初始化 BM25 搜索模块，结合 MySQL 和 Redis self.bm25_search = Bm25Search( mysql_client=self.db, redis_client=self.cache ) # 初始化 OpenAI 客户端，连接 DashScope API # 大模型 self.llm = ChatOpenAI( model=config.LLM_MODEL, api_key=config.DASHSCOPE_API_KEY, base_url=config.DASHSCOPE_BASE_URL, temperature=0.1, ) # # 初始化 RAG 系统: rag_qa_system self.rag_system = RagSystem()(2).调用 DashScope API_call_model方法功能：通过 OpenAI 客户端调用 DashScope API，基于用户提示生成答案关键点：设置系统提示为“你是一个有用的助手”，异常处理捕获 API 调用失败，返回错误信息代码如下:def _call_model(self, prompt, count_down=3): """ 定义调用 DashScope API 的方法，生成自然语言答案 :param prompt: :param count_down: :return: """ start_time = time.time() logger.debug(f'开始调用大模型，最大重试次数: {count_down}') for i in range(count_down): try: # 创建聊天完成请求 messages = [ SystemMessage('你是一个有用的助手，能够准确无误的完成用户的命令'), HumanMessage(prompt), ] # 调用 DashScope API result = self.llm.invoke(messages).content elapsed = time.time() - start_time logger.debug(f'大模型调用成功，第 {i + 1} 次尝试, 耗时: {elapsed:.2f}s') return result except Exception as e: logger.error(f'大模型调用失败，第 {i + 1}/{count_down} 次尝试，原因: {e}') if i count_down - 1: logger.debug(f'等待 {1} 秒后重试...') time.sleep(1) continue elapsed = time.time() - start_time logger.warning(f'大模型调用失败，已重试 {count_down} 次，放弃调用, 总耗时: {elapsed:.2f}s') return None(4).查询处理search方法(主逻辑)功能：处理用户查询，优先通过 BM25 搜索 MySQL，若无可靠答案则回退到 RAG 系统关键点：设置 BM25 相似度阈值 0.85，记录查询和处理时间，支持学科过滤代码如下:def _fallback_answer(self): """兜底逻辑：当所有模块都无法返回有效答案时使用""" phone = self.config.CUSTOMER_SERVICE_PHONE return f'抱歉，系统繁忙，暂时无法为您解答。请拨打客服电话 {phone} 获取帮助。' def search(self, query, subject_filter=None): """ 处理用户查询，优先通过 BM25 搜索 MySQL，若无可靠答案则回退到 RAG 系统 :param query: :param subject_filter: :return: """ try: # 2. 将用户query发送给faq模块进行处理，得到答案和是否进入RAG模块的标识 answer, need_rag = self.bm25_search.search(query) except Exception as e: logger.error(f'FAQ模块调用异常: {e}') return self._fallback_answer() # 3. 如果答案可靠，直接返回 if answer: # 如果找到可靠答案，记录答案到日志 logger.info(f"MySQL 答案: {answer}") # 返回 MySQL 答案 return answer # 4. 如果答案不可靠，且需要进入RAG模块，调用rag模块，得到答案 if need_rag: try: # 如果需要 RAG，记录回退信息到日志 logger.info("无可靠 MySQL 答案，回退到 RAG") # 调用 RAG 系统生成答案，支持学科过滤 answer = self.rag_system.generate_answer(query, subject_filter) # 记录 RAG 答案到日志 logger.info(f"RAG 答案: {answer}") except Exception as e: logger.error(f'RAG模块调用异常: {e}') return self._fallback_answer() if answer: return answer # 5. 兜底逻辑 # 如果无答案，记录信息到日志 logger.info("未找到答案") return self._fallback_answer()(5).命令行交互main函数功能：提供交互式命令行界面，接受用户查询和学科过滤，显示答案关键点：验证学科过滤的有效性，异常处理和资源清理确保系统健壮代码如下:def main(): # 定义主函数，提供命令行交互界面 qa_system = IntegratedQASystem() # 初始化问答系统 try: # 打印欢迎信息 print("\n欢迎使用集成问答系统！") # 打印支持的学科类别 print(f"支持的来源: {qa_system.config.VALID_SOURCES}") # 提示用户输入查询或退出 print("输入查询进行问答，输入 'exit' 退出。") while True: # 获取用户输入的查询 query = input("\n输入查询: ").strip() if query.lower() == "exit": # 如果用户输入 exit，记录退出日志 logger.info("退出系统") # 打印退出信息 print("再见！") # 退出循环 break # 获取用户输入的学科过滤 source_filter = input(f"输入来源过滤 ({'/'.join(qa_system.config.VALID_SOURCES)}) (按 Enter 跳过): ").strip() if source_filter and source_filter not in qa_system.config.VALID_SOURCES: # 如果学科过滤无效，记录警告日志 logger.warning(f"无效来源 '{source_filter}'，忽略过滤") # 打印无效信息，忽略过滤 print(f"无效来源 '{source_filter}'，继续无过滤。") source_filter = None # 执行查询，获取答案 answer = qa_system.search(query, source_filter) # 打印答案 print(f"\n答案: {answer}") except Exception as e: # 记录系统错误日志 logger.error(f"系统错误: {e}") # 打印错误信息 print(f"发生错误: {e}") finally: # 无论是否发生错误，关闭 MySQL 连接 qa_system.db.close() if __name__ == "__main__": # 如果脚本作为主程序运行，调用 main 函数 main() # if __name__ == '__main__': # qa_system = IntegratedQASystem() # qa_system_search = qa_system.search('如何创建线程安全的单例对象') # time.sleep(5) # print(qa_system_search)整体代码如下:""" 需求：实现edu-rag项目 思路步骤： 1. 初始化：bm25search、大模型、rag_qa_system 2. 将用户query发送给FAQ模块进行处理，得到答案和是否进入RAG模块的标识 3. 如果答案可靠，直接返回 4. 如果答案不可靠，且需要进入RAG模块，调用rag模块，得到答案 5. 如果答案不可靠，且不需要进入RAG模块，直接走兜底逻辑 """ import time from langchain_core.messages import SystemMessage, HumanMessage # 导入 OpenAI 客户端，用于调用 DashScope API from langchain_openai import ChatOpenAI # 导入配置和日志工具，用于系统配置和日志记录 from base.logger import single_logger as logger from base.config import single_config as config # 导入 MySQL 系统组件，用于数据库操作和搜索 from mysql_qa.retreival.bm25_search import Bm25Search from mysql_qa.cache.redis_client import RedisClient from mysql_qa.db.mysql_client import MysqlClient # 导入 RAG 系统组件，用于知识库检索和答案生成 from rag_qa.core.rag_system import RagSystem class IntegratedQASystem: # 1. 初始化：bm25search、大模型、rag_qa_system def __init__(self): # bm25search # 初始化配置对象，加载系统参数 self.config = config # 初始化日志工具，用于记录系统运行信息 self.logger = logger # 初始化 MySQL 客户端，用于数据库操作 self.db = MysqlClient() # 初始化 Redis 客户端，用于缓存管理 self.cache = RedisClient() # 初始化 BM25 搜索模块，结合 MySQL 和 Redis self.bm25_search = Bm25Search( mysql_client=self.db, redis_client=self.cache ) # 初始化 OpenAI 客户端，连接 DashScope API # 大模型 self.llm = ChatOpenAI( model=config.LLM_MODEL, api_key=config.DASHSCOPE_API_KEY, base_url=config.DASHSCOPE_BASE_URL, temperature=0