【AI大模型应用开发】【项目实战】16.RAG智慧问答项目-(四)基于MySql库的问答系统(FQA模块)之基于Mysql数据库实现问答系统

目标

• 理解FQA问答系统的整体流程

• 掌握如何整合MySQL、Redis和BM25算法构建QA问答系统

一、FQA系统概述

系统从MySQL数据库检索问答对，使用BM25算法计算相似度，并通过Softmax归一化将得分转换为概率值，阈值0.85判断答案可靠性。Redis仅缓存高可靠性结果（相似度>0.85且有答案）。若MySQL无可靠答案，则调用RAG系统检索

1.系统流程

• 1.数据存储：MySQL存储FQA高频问答对数据
• 2.问题检索：BM25计算相似度，Softmax归一化后判断阈值0.85
• 3.缓存管理：Redis仅存储相似度>0.85且有答案的数据
• 4.答案返回：
  • 若MySQL返回可靠答案，直接返回
  • 否则，调用RAG系统检索

对于以上流程，具体的实现如下图：

二、项目结构

三、代码实现

1.配置文件 (config.ini)

# MySQL 配置 [mysql] host = localhost user = root password = 123456 database = subjects # Redis 配置 [redis] host = localhost port = 6379 password = 123456 db = 0 # 日志配置: 写自己的日志路径，最好给绝对路径 [logger] log_file = /path/to/your/logs/app.log

2.配置管理

2.1 功能

config.py文件定义了Config类，用于集中管理系统中的所有配置参数,这些参数包括数据库连接信息、模型选择、分块策略、API设置等,通过集中管理配置，系统可以方便地调整参数、适配不同环境，并支持通过环境变量进行灵活配置

2.2 代码实现

""" 需求：从配置文件中加载各类配置信息 思路步骤： 1. 创建配置解析器对象，并读取配置文件 2. 加载mysql配置 2.1 host 2.2 user 2.3 password 2.4 database 3. 加载redis配置 3.1 host 3.2 user 3.3 password 3.4 db 4. 加载日志路径 """

base/config.py

# base/config.py # 导入配置解析库 import configparser # 解析.ini文件的 # 导入路径操作库 import os class Config: # 初始化配置，加载 config.ini 文件 def __init__(self, config_file=None): # 创建配置解析器，启用插值功能 self.config = configparser.ConfigParser(interpolation=configparser.ExtendedInterpolation()) # 如果没有提供配置文件路径，则使用默认路径 self.PROJECT_ROOT = os.path.dirname(os.path.dirname(__file__)) self.LOG_DIR = os.path.join(self.PROJECT_ROOT, 'logs') if config_file is None: config_file = os.path.join(self.PROJECT_ROOT, 'config.ini') # 读取配置文件 self.config.read(config_file, encoding='utf-8') # MySQL 配置 # MySQL 主机地址 # fallback 表示默认值, 如果.ini配置文件中没有的话, 就只用fallback默认值数据 self.MYSQL_HOST = os.getenv('MYSQL_HOST', self.config.get('mysql', 'host', fallback='localhost')) # MySQL 用户名 self.MYSQL_USER = os.getenv('MYSQL_USER', self.config.get('mysql', 'user', fallback='edu_rag')) # MySQL 密码 self.MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD', self.config.get('mysql', 'password', fallback='123456')) # MySQL 数据库名 self.MYSQL_DATABASE = os.getenv('MYSQL_DATABASE', self.config.get('mysql', 'database', fallback='subjects')) # Redis 配置 # Redis 主机地址 self.REDIS_HOST = os.getenv('REDIS_HOST', self.config.get('redis', 'host', fallback='localhost')) # Redis 端口 self.REDIS_PORT = int(os.getenv('REDIS_PORT', self.config.get('redis', 'port', fallback=6379))) # Redis 密码 self.REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', self.config.get('redis', 'password', fallback='')) # Redis 数据库编号 self.REDIS_DB = int(os.getenv('REDIS_DB', self.config.get('redis', 'db', fallback=0))) # 日志文件路径 self.LOG_FILE = os.path.join(self.LOG_DIR, 'app.log') single_config = Config() if __name__ == '__main__': conf = single_config print(conf.MYSQL_USER) print(conf.LLM_MODEL)

config.py读取config.ini为什么不能用相对路径:

2.3 说明

• 默认值：每个参数设有默认值，确保未配置环境变量时系统仍可运行
• 参数分类：按功能分类（如数据库、模型、分块等），便于管理和维护

3.日志记录

3.1 功能

logger.py文件定义了get_logger函数，用于配置系统的日志记录器,日志记录器将运行信息、警告和错误输出到文件和控制台，便于开发、调试和运维人员监控系统状态

3.2 代码实现

""" 需求： 封装一个日志获取工具，用于项目中所有模块的日志处理。需要在控制台和文件中分类 思路： 1. 确保日志目录存在，不存在创建 2. 创建日志记录器: Logger 2.1 获取Logger对象 2.2 设置日志级别为所有控制器最低的(设置全局的日志级别) 3. 创建控制台控制器：StreamHandler 3.1 创建控制台处理器对象 3.2 设置日志级别为INFO 4. 创建文件处理器:FileHandler，并指定目录 4.1 创建文件处理对象 4.2 设置日志级别为DEBUG 5. 定义并设置日志格式： 5.1 定义日志格式：logging.Formatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s') 5.2 设置处理器日志格式 6. 把处理器添加到logger中 """

需要注意，日志路径尽量写绝对路径

import logging import os from base.config import Config """ 需求： 封装一个日志获取工具，用于项目中所有模块的日志处理。需要在控制台和文件中分类 思路： 1. 确保日志目录存在，不存在创建 2. 创建日志记录器: Logger 2.1 获取Logger对象 2.2 设置日志级别为所有控制器最低的(设置全局的日志级别) 3. 创建控制台控制器：StreamHandler 3.1 创建控制台处理器对象 3.2 设置日志级别为INFO 4. 创建文件处理器:FileHandler，并指定目录 4.1 创建文件处理对象 4.2 设置日志级别为DEBUG 5. 定义并设置日志格式： 5.1 定义日志格式：logging.Formatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s') 5.2 设置处理器日志格式 6. 把处理器添加到logger中 """ def get_logger(project_name , level=logging.DEBUG , file_level=logging.DEBUG , console_level=logging.INFO , formatter='%(asctime)s - %(levelname)s - %(name)s - %(module)s - %(lineno)d - %(message)s' , encode='utf-8' , mode='a' , log_url='logs/app.log'): """ 团队通用日志打印工具 :param project_name: 项目名，必填 :param level: 全局日志级别... :param file_level: :param console_level: :param formatter: 格式 :param encode: 编码 :param mode: mode 类型 :param log_url: 日志打印到哪个文件 :return: """ # 1. 确保日志目录存在，不存在创建 # os.path.dirname(log_url) 获取文件所在的目录名 # exist_ok=True: 如果文件夹存在, 不报错 os.makedirs(os.path.dirname(log_url), exist_ok=True) # 2. 创建日志记录器: Logger # 2.1 获取Logger对象 logger = logging.getLogger(project_name) # 2.2 设置日志级别为所有控制器最低的(设置全局的日志级别) logger.setLevel(level) # 设置记录器级别 # 3. 创建控制台控制器：StreamHandler # 3.1 创建控制台处理器对象 console_handler = logging.StreamHandler() # 3.2 设置日志级别为INFO console_handler.setLevel(console_level) # 控制台显示INFO及以上级别 # 4. 创建文件处理器:FileHandler，并指定目录 # 4.1 创建文件处理对象 # TODO 需要注意，非特殊情况，mode一定不要写w file_handler = logging.FileHandler(log_url, mode=mode, encoding=encode) # 4.2 设置日志级别为DEBUG # TODO 实际工作中INFO就可以， DEBUG更多是用于本地调试代码的时候使用 file_handler.setLevel(file_level) # 文件记录DEBUG及以上级别 # 5. 定义并设置日志格式： # 5.1 定义日志格式：eg: logging.Formatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s') formatter = logging.Formatter(fmt=formatter) # 5.2 设置处理器日志格式 console_handler.setFormatter(formatter) file_handler.setFormatter(formatter) # 6.避免重复添加 if not logger.handlers: # 6.1 把处理器添加到logger中 logger.addHandler(console_handler) logger.addHandler(file_handler) re