基于MySQL数据库实现问答系统与BM25算法详解学习目标通过本章学习，你将掌握：FQA（常见问题解答）系统的整体架构与工作流程基于MySQL数据库存储问答知识库的实现方案Redis缓存加速问答系统响应的设计思路BM25文本检索算法的原理与代码实现Softmax归一化在相似度判断中的应用完整的Python项目模块化设计与实现一、FQA系统概述1 系统流程FQA（Frequently Asked Questions）系统是一种基于知识库的智能问答系统，核心是通过文本检索技术从预存的问答库中找到与用户问题最匹配的答案。2 项目结构本项目采用模块化设计，将不同功能封装为独立模块，便于维护和扩展。1 系统流程FQA系统的完整工作流程如下：用户提问：用户输入自然语言问题文本预处理：对用户问题进行分词、清洗等预处理操作BM25检索：使用BM25算法计算用户问题与知识库中所有问题的相似度得分Softmax归一化：将BM25原始得分转换为0-1之间的概率值，便于阈值判断缓存判断：若最高相似度得分超过阈值（如0.85），且Redis中存在缓存答案，直接返回缓存结果数据库查询：若未命中缓存但得分达标，从MySQL数据库中查询对应答案结果回写：将高置信度的问答对写入Redis缓存，提升后续查询速度兜底处理：若得分未达标，可转向RAG大模型生成答案或返回"未找到相关答案"核心设计思想：通过Redis缓存高频问答对，减少数据库查询和BM25计算开销，实现冷热数据分离，大幅提升系统响应速度和并发能力。2 项目结构项目采用标准的Python模块化设计，目录结构如下：fqa_system/ ├── config/ │ └── config.ini # 配置文件 ├── src/ │ ├── __init__.py │ ├── config_manager.py # 配置管理模块 │ ├── logger.py # 日志记录模块 │ ├── mysql_db.py # MySQL数据库操作模块 │ ├── redis_cache.py # Redis缓存操作模块 │ ├── text_preprocess.py # 文本预处理模块 │ └── bm25_search.py # BM25检索模块 ├── data/ │ └── qa_dataset.csv # 问答数据集 ├── logs/ │ └── app.log # 日志文件 ├── main.py # 主程序入口 └── requirements.txt # 依赖文件各模块职责清晰，遵循单一职责原则，便于单元测试和维护。三、代码实现1 配置文件 (config.ini)使用INI格式的配置文件统一管理系统配置，将数据库连接、缓存参数、日志配置等与代码分离，便于不同环境切换。[mysql] host = 127.0.0.1 port = 3306 user = root password = 123456 database = fqa_db charset = utf8mb4 [redis] host = 127.0.0.1 port = 6379 password = db = 0 decode_responses = True [log] log_file = logs/app.log level = INFO format = %(asctime)s | %(levelname)s | %(message)s [bm25] k1 = 1.5 b = 0.75 threshold = 0.85 [system] top_k = 5配置说明：配置文件按功能划分为多个section（区块），分别管理MySQL、Redis、日志、BM25算法等不同模块的参数，实现配置解耦。2 配置管理2.1 功能配置管理模块负责读取和解析INI配置文件，提供统一的配置访问接口，避免各模块直接读取文件，实现配置集中管理。支持按section分组读取配置自动类型转换（字符串、整数、布尔值）单例模式，全局唯一配置实例配置项缺失时给出明确错误提示2.2 代码实现importconfigparserimportosclassConfigManager:"""配置管理类，单例模式"""_instance=Nonedef__new__(cls,config_path=None):ifcls._instanceisNone:cls._instance=super().__new__(cls)cls._instance._init_config(config_path)returncls._instancedef_init_config(self,config_path=None):"""初始化配置"""ifconfig_pathisNone:# 默认配置文件路径config_path=os.path.join(os.path.dirname(os.path.dirname(__file__)),"config","config.ini")ifnotos.path.exists(config_path):raiseFileNotFoundError(f"配置文件不存在:{config_path}")self.config=configparser.ConfigParser()self.config.read(config_path,encoding="utf-8")defget(self,section,key,fallback=None):"""获取字符串类型配置"""returnself.config.get(section,key,fallback=fallback)defgetint(self,section,key,fallback=None):"""获取整数类型配置"""returnself.config.getint(section,key,fallback=fallback)defgetfloat(self,section,key,fallback=None):"""获取浮点数类型配置"""returnself.config.getfloat(section,key,fallback=fallback)defgetboolean(self,section,key,fallback=None):"""获取布尔类型配置"""returnself.config.getboolean(section,key,fallback=fallback)defget_section(self,section):"""获取整个section的配置字典"""ifsectionnotinself.config:return{}returndict(self.config[section])# 全局配置实例defget_config():returnConfigManager()2.3 说明单例模式：确保整个应用只有一个配置实例，避免重复读取文件，节省资源类型安全：提供getint、getfloat、getboolean等方法，自动进行类型转换容错处理：支持fallback参数，配置项缺失时返回默认值使用方式：通过get_config()获取全局配置实例，如get_config().get("mysql", "host")3 日志记录3.1 功能日志记录模块封装Python标准logging库，提供统一的日志记录接口，支持控制台和文件双输出，便于开发调试和线上问题排查。支持控制台和文件双渠道输出日志分级（DEBUG/INFO/WARNING/ERROR/CRITICAL）标准化日志格式，包含时间、级别、内容自动创建日志目录全局单例日志器3.2 代码实现importloggingimportosfromlogging.handlersimportRotatingFileHandlerfromconfig_managerimportget_configclassLogger:"""日志管理类，单例模式"""_instance=Nonedef__new__(cls):ifcls._instanceisNone:cls._instance=super().__new__(cls)cls._instance._init_logger()returncls._instancedef_init_logger(self):"""初始化日志器"""config=get_config()# 创建日志器self.logger=logging.getLogger("fqa_system")self.logger.setLevel(getattr(logging,config.get("log","level","INFO")))self.logger.propagate=False# 日志格式log_format=config.get("log","format","%(asctime)s | %(levelname)s | %(message)s")formatter=logging.Formatter(log_format)# 控制台处理器console_handler=logging.StreamHandler()console_handler.setLevel(logging.INFO)console_handler.setFormatter(formatter)self.logger.addHandler(console_handler)# 文件处理器log_file=config.get("log","log_file","logs/app.log")log_dir=os.path.dirname(log_file)iflog_dirandnotos.path.exists(log_dir):os.makedirs(log_dir,exist_ok=True)# 使用滚动文件处理器，避免单文件过大file_handler=RotatingFileHandler(log_file,maxBytes=10*1024*1024,# 10MBbackupCount=5,encoding="utf-8")file_handler.setLevel(logging.DEBUG)file_handler.setFormatter(formatter)self.logger.addHandler(file_handler)defdebug(self,message):self.logger.debug(message)definfo(self,message):self.logger.info(message)defwarning(self,message):self.logger.warning(message)deferror(self,message,exc_info=False):self.logger.error(message,exc_info=exc_info)defcritical(self,message,exc_info=False):self.logger.critical(message,exc_info=exc_info)# 全局日志实例defget_logger():returnLogger()3.3 说明双输出渠道：控制台输出便于开发调试，文件输出用于线上问题排查日志滚动：使用RotatingFileHandler实现日志文件滚动，单个文件最大10MB，保留5个备份分级过滤：控制台输出INFO及以上级别，文件记录DEBUG及以上级别，兼顾性能与排错需求使用方式：通过get_logger()获取全局日志实例，如get_logger().info("系统启动成功")线上规范：生产环境建议将日志级别设置为INFO，屏蔽DEBUG调试信息，减少磁盘占用；数据库操作、缓存操作、检索计算等关键环节必须打印INFO日志记录入参和耗时。4 MySQL操作模块4.1 功能MySQL操作模块封装数据库的增删改查操作，负责问答知识库的持久化存储，是系统的底层数据源。问答知识库的CRUD操作批量导入问答数据连接池管理，自动重连机制参数化查询，防止SQL注入统一的异常处理和日志记录4.2 代码实现importpymysqlfrompymysql.cursorsimportDictCursorfromdbutils.pooled_dbimportPooledDBfromconfig_managerimportget_configfromloggerimportget_loggerclassMySQLDB:"""MySQL数据库操作类"""_instance=Nonedef__new__(cls):ifcls._instanceisNone:cls._instance=super().__new__(cls)cls._instance._init_pool()returncls._instancedef_init_pool(self):"""初始化数据库连接池"""config=get_config()self.logger=get_logger()try:self.pool=PooledDB(creator=pymysql,maxconnections=10,mincached=2,maxcached=5,host=config.get("mysql","host"),port=config.getint("mysql","port"),user=config.get("mysql","user"),password=config.get("mysql","password"),database=config.get("mysql","database"),charset=config.get("mysql","charset","utf8mb4"),cursorclass=DictCursor,ping=4,# 连接前自动ping，断连自动重连)self.logger.info("MySQL连接池初始化成功")exceptExceptionase:self.logger.error(f"MySQL连接池初始化失败:{e}",exc_info=True)raisedefget_connection(self):"""从连接池获取连接"""returnself.pool.connection()defget_all_questions(self):"""获取所有问题，用于BM25索引构建"""conn=Nonetry:conn=self.get_connection()withconn.cursor()ascursor:sql="SELECT id, question, answer FROM qa_table"cursor.execute(sql)results=cursor.fetchall()self.logger.info(f"从MySQL获取了{len(results)}条问答数据")returnresultsexceptExceptionase:self.logger.error(f"获取所有问题失败:{e}",exc_info=True)return[]finally:ifconn:conn.close()defget_answer_by_question(self,question):"""根据问题查询答案"""conn=Nonetry:conn=self.get_connection()withconn.cursor()ascursor:sql="SELECT answer FROM qa_table WHERE question = %s"cursor.execute(sql,(question,))result=cursor.fetchone()returnresult["answer"]ifresultelseNoneexceptExceptionase:self.logger.error(f"根据问题查询答案失败:{e}",exc_info=True)returnNonefinally: