【AI大模型应用开发】【项目实战】24.RAG智慧问答项目-(十二)融合MySQL的RAG系统之基于 FastAPI 的 FAQ 与知识库查询接口

目标：理解如何基于FastAPI构建高效的问答系统 API 接口掌握WebSocket 流式输出和HTTP 接口的实现，优化用户交互体验前后端交互、前端静态文件服务以及日常问候处理的集成一、开发完RAG系统以后，怎么对外提供服务？对外：调用链路： 前端 - 后端 - RAG系统功能：前端和后端是已有的业务系统，比如电商系统增加一个售后机器人：前端和后端改造代码， RAG系统从0到1开发职责划分：前端和后端对应的人员进行维护开发， AI工程师只负责做RAG系统RAG系统专门作为一个单独的系统对外服务，前端、后端从0到1开发，RAG系统也是从0到1职责划分：AI工程师负责做RAG系统， 也有可能兼职做后端（小团队甚至要做前端）,后端仍然需要权限验证、登录等功能二、查询流程图app.py是对new_main.py的进一步优化，基于 FastAPI 框架构建了RESTful API和WebSocket 接口，集成了 RAG 系统 和 RAG 系统，支持流式输出、对话历史管理和日常问候处理,同时，通过静态文件服务支持前端页面交互，适用于项目场景下的实时问答系统下面是基于 FastAPI 的问答系统查询流程图，展示从前端请求到后端响应的处理逻辑三、流程说明1.请求接收：前端通过HTTP POST (/api/query)或WebSocket (/api/stream)发送查询请求，包含查询内容 (query)、学科过滤 (source_filter)和会话 ID (session_id)2.日常问候处理：检查查询是否为日常问候（如“你好”），若匹配则返回模板化回复3.BM25 搜索：使用BM25 算法搜索MySQL 知识库，设置相似度阈值 0.854.答案判断:若找到可靠答案（相似度 0.85）,通过HTTP 一次性返回或WebSocket 流式返回若无可靠答案且需要 RAG，HTTP 接口提示使用 WebSocket，WebSocket 接口流式返回 RAG 答案若无需 RAG，返回“未找到答案”5.历史更新：将查询和答案存入 MySQL 的conversations表，保留最近 5 轮对话6.响应输出：HTTP 返回 JSON 格式响应，WebSocket 流式发送 JSON 数据（包含start、token、end或error类型)四、代码介绍下面是app.py的完整代码，包含逐行注释，详细解析功能与实现逻辑实现步骤，准备工作创建 FastAPI 应用并配置构建一个FastAPI对象配置跨域初始化RAG系统配置问答模板，定义日常问答模式定义请求和响应的数据结构，给核心query接口使用request: query，source_filter, session_idResponse: answer,is_streaming, session_id ,processing_timeis_streaming代表返回的结果是批式还是流式的，如果是批式的，前端直接获取answer。如果是流式的，需要前端去构建websocket通道，接收流式输出的内容挂载静态文件，重定向: / 到静态文件中的前端页面历史对话操作创建历史对话查询历史消息清除历史消息1.导入必备的工具包# 导入 FastAPI 相关模块，用于构建 API 和 WebSocket from fastapi import FastAPI, WebSocket, HTTPException, Query, Depends # 导入 FastAPI 响应类型，用于流式响应和文件服务 from fastapi.responses import StreamingResponse, FileResponse # 导入 CORS 中间件，支持跨域请求 from fastapi.middleware.cors import CORSMiddleware # 导入静态文件服务模块 # 给前端使用 from fastapi.staticfiles import StaticFiles # 导入 WebSocket 断开异常 from starlette.websockets import WebSocketDisconnect # 导入系统操作模块，用于文件目录管理 import os # 导入 Pydantic 模型，用于请求验证 from pydantic import BaseModel # 导入异步事件循环模块 import asyncio # 导入 JSON 处理模块 import json # 导入 UUID 模块，生成唯一会话 ID import uuid # 导入类型注解模块 from typing import Optional, List, Dict, Any # 导入时间模块，记录处理时间 import time # 导入正则表达式模块，用于匹配日常问候 import re # 导入优化后的问答系统 from new_main import IntegratedQASystem from base.logger import single_logger as logger2.创建 FastAPI 应用并配置创建 FastAPI 实例并配置 CORS和静态文件服务功能：初始化 FastAPI 应用，允许跨域请求，挂载静态文件目录关键点：CORS 配置在生产环境中需限制域名，静态文件服务支持前端页面# 创建 FastAPI 应用实例，设置标题和描述 app = FastAPI(title="问答系统API", description="集成MySQL和RAG的智能问答系统") # 配置 CORS 中间件，允许跨域请求 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 允许所有来源（生产环境需限制） allow_credentials=True, # 允许凭证 allow_methods=["*"], # 允许所有 HTTP 方法 allow_headers=["*"], # 允许所有头部 ) # 创建静态文件目录（如果不存在） os.makedirs("static", exist_ok=True) # 创建全局问答系统实例 qa_system = IntegratedQASystem()3. 定义日常问候模式定义问候语正则匹配模式和回复功能：快速响应常见问候，提升用户体验关键点：使用正则表达式匹配，支持大小写无关匹配# 定义日常问候用语模式和回复 GREETING_PATTERNS = [ { "pattern": r"^(你好|您好|hi|hello)", # 匹配问候语 "response": "你好！我是小爱同学，专注于答疑解惑，很高兴为你服务！" }, { "pattern": r"^(你是谁|您是谁|你叫什么|你的名字|who are you)", # 匹配身份询问 "response": "我是小爱同学，你的智能助手，致力于提供 各方面 内容相关的解答！" }, { "pattern": r"^(在吗|在不在|有人吗)", # 匹配在线确认 "response": "我在！我是小爱同学，随时为你解答问题！" }, { "pattern": r"^(干嘛呢|你在干嘛|做什么)", # 匹配状态询问 "response": "我正在待命，随时为你解答 互联网 相关的问题！有什么我可以帮你的？" } ]4.定义请求和响应模型使用Pydantic定义请求和响应数据结构功能：规范 API 输入和输出格式，确保数据验证关键点：支持可选字段，方便扩展# 定义查询请求模型 # TODO 相当于java中的注解 @RequestBody class QueryRequest(BaseModel): # TODO @Param(param = "query", nullable = false) query: str # 查询内容，必填 # TODO @Param(param = "query", nullable = true) subject_filter: Optional[str] = None # 学科过滤，可选 session_id: Optional[str] = None # 会话 ID，可选 # 定义查询响应模型 # TODO 相当于java中的注解 @ResponseBody class QueryResponse(BaseModel): answer: str # 答案内容 # TODO 这个bool值的意思是告诉前端，将要接受流式输出，需要建立一个websocket通道 is_streaming: bool # 是否流式响应 session_id: str # 会话 ID processing_time: float # 处理时间5.挂载静态文件服务挂载静态文件目录并设置根路径功能：服务前端 HTML 文件，支持交互页面关键点：根路径重定向到index.html# 挂载静态文件目录，服务前端文件 # TODO app.mount("/static" - 127.0.0.1/static app.mount("/static", StaticFiles(directory="static"),) # 根路径重定向到 index.html # TODO 127.0.0.1/static/index.html - 127.0.0.1/ @app.get("/") async def read_root(): return FileResponse("static/index.html")6.创建新会话/api/create_session接口功能：生成新的唯一会话 ID关键点：使用 UUID 确保唯一性# 创建新会话接口 # TODO 类似于java中的@PostMapping @app.post("/api/create_session") async def create_session(): session_id = str(uuid.uuid4()) # 生成唯一会话 ID return {"session_id": session_id} # 返回会话 ID7.查询历史消息/api/history/{session_id}接口功能：获取指定会话的历史记录关键点：异常处理确保接口健壮性# 查询历史消息接口 # TODO restful api 规范 。比如某厂规范：internal-api.xxxx.com/{业务线: driver/passanger}/{二级:运营、打车}/xxxx # TODO 对外的api(openapi) : /api/v1/edu-rag/history/{session_id} @app.get("/api/history/{session_id}") async def get_history(session_id: str): try: # 获取指定会话的历史记录 history = qa_system.get_session_history(session_id) # 返回会话 ID 和历史记录 # TODO 规范操作：{"data": {"session_id": session_id, "history": history} ,"errno":0,"errmsg":报错信息, "log_id":日志编号} return {"session_id": session_id, "history": history} except Exception as e: # 抛出 HTTP 异常，包含错误信息 raise HTTPException(status_code=500, detail=f"获取历史记录失败: {str(e)}")8.清除历史消息/api/history/{session_id}接口功能：删除指定会话的所有历史记录关键点：返回操作状态，异常处理确保健壮性# 清除历史消息接口 @app.delete("/api/history/{session_id}") async def clear_history(session_id: str): # 清除指定会话的历史记录 success = qa_system.clear_session_history(session_id) if success: # 返回成功状态 return {"status": "success", "message": "历史记录已清除"} else: # 抛出 HTTP 异常 raise HTTPException(status_code=500, detail="清除历史记录失败")9.检查日常问候check_greeting函数功能：检查查询是否匹配日常问候，返回模板化回复关键点：正则匹配支持大小写无关，简化交互逻辑# 检查是否为日常问候用语并返回模板回复 def check_greeting(query: str) - Optional[str]: query_text = query.strip() # 去除首尾空格 for pattern_info in GREETING_PATTERNS: # 使用正则匹配，忽略大小写 if re.match(pattern_info["pattern"], query_text, re.IGNORECASE): return pattern_info["response"] # 返回匹配的回复 return None # 无匹配返回 None10.非流式查询接口/api/query接口功能：处理非流式查询，支持 MySQL 答案和日常问候，提示使用 WebSocket 获取 RAG 流式答案关键点：返回 JSON 格式响应，记录处理时间后端代码:# 非流式查询接口 @app.post("/api/query") async def query(request: QueryRequest): """ 用户输入的query，前端先去调用当前接口根据返回值的is_streaming字段决定是否输出答案，还是调用websocket接收流式输出 :param request: :return: """ start_time = time.time() # 记录开始时间 # 使用请求中的 session_id 或生成新 ID session_id = request.session_id or str(uuid.uuid4()) # 检查是否为日常问候 greeting_response = check_greeting(request.query) if greeting_response: # 返回问候回复 return { "answer": greeting_response, "is_streaming": False, "session_id": session_id, "processing_time": time.time() - start_time } # 执行 BM25 搜索 answer, need_rag = qa_system.faq.search(request.query, threshold=0.85) if need_rag: # 需要 RAG，提示使用 WebSocket return { "answer": "请使用WebSocket接口获取流式响应", "is_streaming": True, "session_id": session_id, "processing_time": time.time() - start_time } # 返回 MySQL 答案 return { "answer": answer, "is_streaming": False, "session_id": session_id, "processing_time": time.time() - start_time }前端调用代码:// 接收消息时 socket.onmessage = function(event) { const data = JSON.parse(event.data); switch (data.type) { case 'start': console.log('流式响应开始'); break; case 'token': // 追加 token 到内容 accumulatedContent += data.token; // 使用 marked.js 解析 Markdown 格式 contentElement.innerHTML = marked.parse(accumulatedContent); scrollToBottom(); break; case 'end': console.log('流式响应结束'); socket.close(); resolve(); break; case 'error': console.error('流式响应错误:', data.error);