RAG06-融合FQA的RAG系统
融合FQA的RAG系统
1 融合FQA的RAG系统-基础版
1.1 学习目标： ¶
理解智能问答系统的端到端工作流程。
掌握 FQA 和 RAG 系统的集成与协调。
学习如何通过日志记录和错误处理提升系统健壮性。
1.2 查询流程图 ¶
old_main.py 是一个融合 FQA 和 RAG 系统的交互式问答程序，专注于快速、精准的答案生成。通过 FQA系统 提供结构化数据的高效查询，通过 RAG 系统处理复杂问题，配备日志记录和用户友好的命令行界面，适合教育场景的实时问答需求。
以下是智能问答系统的查询流程图
整体流程
flowchart LR;
%% =========================
%% 主数据流
%% =========================
Query([用户问题query]) --> FQA检索
FQA检索 -->|不超过阈值| RAG检索
RAG检索 --> FinalAnswer([最终答案])
FQA检索 -->|超过阈值| FinalAnswer([最终答案])
FQA系统
flowchart LR;
%% =========================
%% 离线阶段
%% =========================
subgraph 数据准备-离线阶段
A[1.结构化FQA数据]
B[(2.MySQL数据库)]
C[3.jieba分词]
E[4.构建BM25检索器]
A --> B
B --> C
C --> E
end
style 数据准备-离线阶段 fill:none,stroke:#000000
%% =========================
%% 在线阶段
%% =========================
subgraph 问题检索-在线阶段
U[1.用户问题]
V[2.jieba分词]
W[3.BM25相似度检索]
X{4.双重阈值判断}
Y[5.1 Redis查询FQA答案]
Z[6.返回FQA答案]
N[5.2 调用RAG系统]
U --> V
V --> W
W --> X
X -- 是 --> Y
Y --> Z
X -- 否 --> N
end
style 问题检索-在线阶段 fill:none,stroke:#000000
%% =========================
%% 共享组件
%% =========================
R[(Redis缓存FQA)]
%% 在线读缓存
Y --> R
%% Redis未命中查MySQL
R -.未命中.-> B
%% 离线BM25提供在线检索
E -.BM25索引.-> W
RAG系统
flowchart LR;
%%{init:{"themeVariables":{"fontSize":"18px","nodeSpacing":15,"rankSpacing":15}}}%%
%% =========================
%% 在线阶段
%% =========================
subgraph 用户提问阶段-在线
direction TB
Start([用户查询]) --> Classify{查询分类<br/>bert意图识别模型}
Classify -->|通用知识| DirectLLM["直接调用LLM<br/>生成回答"]
Classify -->|专业咨询| Strategy{选择检索策略<br/>LLM}
Strategy -->|明确查询| DirectRetrieval["直接检索"]
Strategy -->|抽象问题| HyDERetrieval["假设问题检索<br/>生成假设答案"]
Strategy -->|复杂查询| SubQuery["子查询检索"]
Strategy -->|啰嗦问题| BacktrackRetrieval["回溯问题检索"]
DirectRetrieval --> OptimizeQuery["LLM优化query"]
HyDERetrieval --> OptimizeQuery
SubQuery --> OptimizeQuery
BacktrackRetrieval --> OptimizeQuery
OptimizeQuery --> Search["检索向量库<br/>得到上下文context"]
Search --> BuildPrompt["拼接Prompt<br/>优化query+context+history"]
BuildPrompt --> Generate["LLM生成回答"]
DirectLLM --> Output([返回回答])
Generate --> HasAnswer["是否有答案<br/>无答案则引导人工支持"]
HasAnswer --> Output
end
style 用户提问阶段-在线 fill:none,stroke:#000000
%% =========================
%% 离线阶段
%% =========================
subgraph 文档向量存储阶段-离线
direction TB
StartDoc([本地知识文档])
StartDoc --> PDFLoader["PDF加载器"]
StartDoc --> DOCLoader["DOC加载器"]
StartDoc --> TXTLoader["TXT加载器"]
StartDoc --> PPTLoader["PPT加载器"]
PDFLoader --> Document["文档对象"]
DOCLoader --> Document
TXTLoader --> Document
PPTLoader --> Document
Document --> ParentChunk["父块分块"]
ParentChunk --> SubChunk["子块分块"]
SubChunk --> SubChunkList["子块集合"]
SubChunkList --> Embed["文档向量化<br/>(bge-m3本地模型)"]
Embed --> Milvus[(Milvus)]
classDef milvusLarge font-size:32px,stroke-width:5px;
class Milvus milvusLarge;
end
style 文档向量存储阶段-离线 fill:none,stroke:#000000
%% =========================
%% 在线检索连接离线向量库
%% =========================
Milvus --> Search
1.3 流程说明 ¶
输入处理 ：用户提供查询 (query) 和可选的学科过滤 (source_filter)。
FQA系统 ：使用 BM25 算法在 MySQL 知识库中搜索FQA数据，超过阈值则命中返回答案，否则调用RAG检索。
RAG系统 ：将用户查询向量化，检索向量库（milvus+混合检索+重排），然后拼接检索结果contexts 和 query为 提示词，调用LLM生成答案。
输出 ：将答案返回给用户。
1.4 代码介绍 ¶
以下是 old_main.py 的完整代码，包含详细注释，逐行解析功能与实现逻辑。
"""
简化版: 实现FQA和RAG结合的问答系统，用户输入一个query，返回最终的答案
工作流：
1.初始化方法
1.初始化 MySQL 客户端
2.初始化 Redis 客户端
3.初始化 FQA系统的BM25检索器
4.初始化 RAG系统
5.初始化 OpenAI客户端
2.调用 DashScope API
3.主流程：问题 -> 答案
0. 添加query校验逻辑，防止为空
1. 记录开始时间
2. 调用FQA系统的BM25Search.query
3. 如果得到答案，直接返回
4. 如果需要查询RAG系统，调用RAGSystem.generate_answer
"""
# 导入 MySQL 系统组件，用于数据库操作和搜索
from mysql_qa.cache.redis_client import RedisClient
from mysql_qa.db.mysql_client import MysqlClient
from mysql_qa.retrieval.bm25_search import BM25Search
# 导入 RAG 系统组件，用于知识库检索和答案生成
from rag_qa.core.vector_store import VectorStore
from rag_qa.core.rag_system import RAGSystem
# 导入配置和日志工具，用于系统配置和日志记录
from base.config import config
from base.logger import logger
from rag_qa.core.prompts import RAGPrompts
# 导入 OpenAI 客户端，用于调用 DashScope API
from openai import OpenAI
# 导入时间库，用于记录处理时间
import time
# 定义类，实现FQA和RAG结合的问答系统
class IntegratedQASystem():
"""
这是一个FQA+RAG的问答系统，实现从用户问题query到最终答案answer的全流程
"""
# 1.初始化方法
def __init__(self):
# 1.初始化 MySQL 客户端
self.mysql_client = MysqlClient()
# 2.初始化 Redis 客户端
self.redis_client = RedisClient()
# 3.初始化 FQA系统的BM25检索器
self.fqa_bm25search = BM25Search(
redis_client=self.redis_client,
mysql_client=self.mysql_client
)
# 4.初始化 RAG系统
self.rag = RAGSystem(
vector_store=VectorStore(),
llm=self.call_dashscope_api,
)
# 5.初始化 OpenAI客户端
self.client = OpenAI(
api_key=config.DASHSCOPE_API_KEY,
base_url=config.DASHSCOPE_BASE_URL
)
# 2.调用 DashScope API
def call_dashscope_api(self, prompt, system_prompt=None):
"""DashScope API 调用，支持传入 system_prompt"""
try:
# 1.使用传入的 system_prompt，未传入时默认使用 RAG 系统提示词
if system_prompt is None:
system_prompt = RAGPrompts.rag_system_prompt()
# 2.创建聊天对话
completion = self.client.chat.completions.create(
model=config.LLM_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=0.1
)
# 3.返回LLM生成的答案
return completion.choices[
0].message.content if completion.choices else f"未找到对应答案，请联系客服: {config.CUSTOMER_SERVICE_PHONE}"
except Exception as e:
logger.error(f"调用DashScope API异常: {e}")
return f"出问题了,请联系客服: {config.CUSTOMER_SERVICE_PHONE}!"
# 3.主流程：问题 -> 答案
def query(self, query, source_filter=None):
# 0.添加query校验逻辑，防止为空
if not query:
return "请输入问题!"
# 1. 记录开始时间
start_time = time.time()
# 2. 调用FQA系统的BM25Search.search
answer, is_need_rag = self.fqa_bm25search.search(query, threshold=(0.85,18.0))
# 3. 如果得到答案，直接返回
duration = time.time() - start_time
if answer:
logger.info(f"FQA系统获得了答案，耗时: {duration:.2f}s")
return answer
logger.info(f"FQA系统未获得答案，耗时: {duration:.2f}s")
# 4. 如果需要查询RAG系统，调用RAGSystem.generate_answer
if is_need_rag:
logger.info(f"开始查询RAG系统, 问题: {query}")
# generate_answer 是生成器函数(yield from)，需要消耗生成器拿到完整答案
answer = ''.join(self.rag.generate_answer(query, source_filter=source_filter))
duration = time.time() - start_time
logger.info(f"RAG系统获得了答案，耗时: {duration:.2f}s")
return answer
else:
duration = time.time() - start_time
logger.info(f"RAG系统未获得答案，耗时: {duration:.2f}s")
return f"未找到对应答案，请联系客服: {config.CUSTOMER_SERVICE_PHONE}"
def main():
# 定义主函数，提供命令行交互界面
qa_system = IntegratedQASystem() # 初始化问答系统
try:
# 打印欢迎信息
print("\n欢迎使用集成问答系统！")
# 打印支持的学科类别
print(f"支持的来源: {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(config.VALID_SOURCES)}) (按 Enter 跳过): ").strip()
if source_filter and source_filter not in config.VALID_SOURCES:
# 如果学科过滤无效，记录警告日志
logger.warning(f"无效来源 '{source_filter}'，忽略过滤")
# 打印无效信息，忽略过滤
print(f"无效来源 '{source_filter}'，继续无过滤。")
source_filter = None
# 执行查询，获取答案
answer = qa_system.query(query, source_filter)
# 打印答案
print(f"\n答案: {answer}")
except Exception as e:
# 记录系统错误日志
logger.error(f"系统错误: {e}")
# 打印错误信息
print(f"发生错误: {e}")
finally:
# 无论是否发生错误，关闭 MySQL 连接
qa_system.mysql_client.close()
if __name__ == "__main__":
# 如果脚本作为主程序运行，调用 main 函数
main()
1.5 章节小结 ¶
本章展示了如何通过 old_main.py 构建一个融合 FQA 和 RAG 的智能问答系统。系统利用 BM25 算法快速匹配结构化数据，RAG 系统处理复杂查询，配备日志记录和错误处理，适合教育场景下的实时问答需求。
2 融合FQA的RAG系统-优化版
2.1 学习目标： ¶
深入理解优化后的智能问答系统，掌握端到端工作流程。
学习如何通过对话历史和流式输出提升用户交互体验。
掌握 FQA 和 RAG 系统的集成，以及健壮的错误处理和日志记录。
2.2 查询流程图 ¶
new_main.py 是对 old_main.py 的优化版本，新增了对话历史管理和流式输出功能。系统整合了 FQA 问答对匹配、RAG 知识库检索、MySQL 对话历史存储，以及流式答案生成。
RAG系统流程（优化版）
flowchart LR;
Query([用户问题 query]) --> SessionCheck[获取对话历史]
SessionCheck --> FQA[FQA 检索]
FQA -->|不超过阈值| RAG[RAG 检索]
FQA -->|超过阈值| UpdateHistory
RAG --> StreamGen[流式生成答案]
StreamGen --> UpdateHistory[更新对话历史]
UpdateHistory --> ReturnResp([返回答案和对话历史])
2.3 流程说明 ¶
输入处理 ：用户提供查询 ( query )、会话 ID ( session_id ) 和可选的学科过滤 ( source_filter )。
会话管理 ：若提供 session_id ，从 MySQL 获取最近 5 轮对话历史；否则生成新的 UUID。
BM25 搜索 ：使用 BM25 算法搜索 MySQL 知识库，设置双重相似度阈值 (0.85,18.0)。
答案判断 ：
若找到可靠答案（相似度 > (0.85,18.0)），一次性返回。
若无可靠答案且需要 RAG，调用 RAG 系统以流式方式生成答案。
若无需 RAG，返回“未找到答案”。
历史更新 ：将查询和答案存入 MySQL 的 conversations 表，保留最近 5 轮对话。
输出 ：通过流式输出（RAG）或一次性输出（MySQL）返回答案，并展示对话历史。
2.4 代码介绍 ¶
以下是 new_main.py 的完整代码，包含逐行注释，详细解析功能与实现逻辑。
"""
优化版: 实现FQA和RAG结合的问答系统，用户输入一个query，返回最终的答案。并支持对话历史管理与流式输出。
整体流程：
1. 初始化系统组件
1.1 初始化 MySQL 客户端
1.2 初始化 Redis 客户端
1.3 初始化 FQA 检索器
1.4 初始化 RAG 系统
1.5 初始化 OpenAI 客户端
1.6 保存日志与配置对象
1.7 初始化对话表
2. 调用 DashScope API
2.1 构造系统提示词
2.2 发起流式请求
2.3 逐块收集模型输出
2.4 异常时返回错误提示
3. 初始化对话表
3.1 创建对话历史表
3.2 提交事务
3.3 记录初始化成功日志
3.4 记录初始化失败日志
4. 获取最近对话历史
4.1 查询最近 5 轮对话
4.2 组装历史记录
4.3 调整为正序，便于上下文拼接
4.4 记录查询失败日志
5. 对外获取对话历史
5.1 直接返回最近历史
6. 更新对话历史
6.1 写入新对话
6.2 重新读取最新历史
6.3 提交事务
6.4 记录更新成功日志
6.5 记录数据库异常并回滚
6.6 记录未知异常并回滚
7. 清除对话历史
7.1 逻辑删除当前对话记录
7.2 提交事务
7.3 记录清除成功日志
7.4 记录清除失败日志
8. 主流程：问题 -> 答案
8.1 记录开始时间
8.2 读取历史对话
8.3 先走 FQA 检索
8.4 FQA 命中直接返回
8.5 需要时再走 RAG
8.6 调用 RAG 生成答案
8.7 更新对话历史
8.8 记录耗时
8.9 未命中时返回兜底提示
"""
import uuid # 生成对话ID
import pymysql
# 数据库与缓存组件
from mysql_qa.cache.redis_client import RedisClient
from mysql_qa.db.mysql_client import MysqlClient
from mysql_qa.retrieval.bm25_search import BM25Search
# RAG 组件
from rag_qa.core.vector_store import VectorStore
from rag_qa.core.rag_system import RAGSystem
from rag_qa.core.prompts import RAGPrompts
# 配置与日志
from base.config import config
from base.logger import logger
# LLM 客户端
from openai import OpenAI
# 时间统计
import time
# FQA + RAG 集成问答系统
class IntegratedQASystem():
"""整合 FQA 与 RAG 的问答系统。"""
# 1.初始化系统组件
def __init__(self):
# 1.初始化 MySQL 客户端
self.mysql_client = MysqlClient()
# 2.初始化 Redis 客户端
self.redis_client = RedisClient()
# 3.初始化 FQA 检索器
self.fqa_bm25search = BM25Search(
redis_client=self.redis_client,
mysql_client=self.mysql_client
)
# 4.初始化 RAG 系统
self.rag = RAGSystem(
vector_store=VectorStore(),
llm=self.call_dashscope_api,
)
# 5.初始化 OpenAI 客户端
self.client = OpenAI(
api_key=config.DASHSCOPE_API_KEY,
base_url=config.DASHSCOPE_BASE_URL
)
# 6.保存日志与配置对象
self.logger = logger
self.config = config
# 7.初始化对话表
self.init_conversations_table()
# 2.调用 DashScope API
def call_dashscope_api(self, prompt, system_prompt=None):
try:
# 1.使用传入的 system_prompt，未传入时默认使用 RAG system prompt
if system_prompt is None:
system_prompt = RAGPrompts.rag_system_prompt()
# 2.发起流式请求
completion = self.client.chat.completions.create(
model=self.config.LLM_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
],
timeout=30,
stream=True
)
# 3.逐块收集模型输出
# 初始化收集流式输出的字符串
collected_content = ""
for chunk in completion:
if chunk.choices and chunk.choices[0].delta.content:
# 累积内容
collected_content += chunk.choices[0].delta.content
yield chunk.choices[0].delta.content
return collected_content
except Exception as e:
# 4.异常时返回错误提示
logger.error(f"调用 DashScope API 异常: {e}")
yield f"出问题了,请联系客服: {config.CUSTOMER_SERVICE_PHONE}!"
# 3.初始化对话表
def init_conversations_table(self):
try:
# 1.创建对话历史表
self.mysql_client.cursor.execute("""
CREATE TABLE IF NOT EXISTS conversations
(
id BIGINT AUTO_INCREMENT PRIMARY KEY,
session_id VARCHAR(36) NOT NULL,
question TEXT NOT NULL,
answer TEXT NOT NULL,
timestamp DATETIME NOT NULL,
_is_deleted BOOLEAN DEFAULT FALSE NOT NULL,
INDEX idx_session_id (session_id)
)
""")
# 2.提交事务
self.mysql_client.connect.commit()
# 3.记录初始化成功日志
self.logger.info("对话历史表初始化成功")
except pymysql.MySQLError as e:
# 4.记录初始化失败日志
self.logger.error(f"初始化对话历史表失败: {e}")
raise
# 4.获取最近对话历史
def _fetch_recent_history(self, session_id: str) -> list:
try:
# 1.查询最近 5 轮对话，按照时间倒序（从大到小），时间越新/大越靠前
self.mysql_client.cursor.execute("""
SELECT question, answer
FROM conversations
WHERE session_id = %s
and _is_deleted = FALSE
ORDER BY timestamp DESC
LIMIT %s
""", (session_id, 5))
# 2.组装历史记录
history = [{"question": row[0], "answer": row[1]} for row in self.mysql_client.cursor.fetchall()]
# 3.调整为正序，便于上下文拼接，时间正序（从小到大）
return history[::-1]
except pymysql.MySQLError as e:
# 4.记录查询失败日志
self.logger.error(f"获取对话历史失败: {e}")
return []
# 5.对外获取对话历史
def get_session_history(self, session_id: str) -> list:
# 1.直接返回最近历史
return self._fetch_recent_history(session_id)
# 6.更新对话历史
def update_session_history(self, session_id: str, question: str, answer: str) -> list:
try:
# 1.写入新对话
self.mysql_client.cursor.execute("""
INSERT INTO conversations (session_id, question, answer, timestamp)
VALUES (%s, %s, %s, NOW())
""", (session_id, question, answer))
# 2.重新读取最新历史
history = self._fetch_recent_history(session_id)
# 删除超出 5 轮的旧记录
# TODO：复杂嵌套SQL，从最内层的括号开始看，然后逐渐向外
# 1. 【取最近5条记录】获取最近5轮对话的ID
# 2. 【取全集和最近5条记录差集】查询当前会话(session_id)下，id不在 获取最近5轮对话的ID以内的
# 3. 【删除差集】删除第二步的结果
# self.mysql_client.cursor.execute("""
# DELETE
# FROM conversations
# WHERE session_id = %s
# AND id NOT IN (SELECT id
# FROM (SELECT id
# FROM conversations
# WHERE session_id = %s
# ORDER BY timestamp DESC
# LIMIT %s) AS sub)
# """, (session_id, session_id, 5))
# 3.提交事务
self.mysql_client.connect.commit()
# 4.记录更新成功日志
self.logger.info(f"对话 {session_id} 历史更新成功")
return history
except pymysql.MySQLError as e:
# 5.记录数据库异常并回滚
self.logger.error(f"更新对话历史失败: {e}")
self.mysql_client.connect.rollback()
raise
except Exception as e:
# 6.记录未知异常并回滚
self.logger.error(f"更新对话历史意外错误: {e}")
self.mysql_client.connect.rollback()
raise
# 7.清除对话历史
def clear_session_history(self, session_id: str) -> bool:
try:
# 1.逻辑删除当前对话记录
new_sql = """
update conversations
set _is_deleted = True
WHERE session_id = %s \
"""
self.mysql_client.cursor.execute(new_sql, (session_id,))
# 2.提交事务
self.mysql_client.connect.commit()
# 3.记录清除成功日志
self.logger.info(f"对话 {session_id} 历史已清除")
return True
except pymysql.MySQLError as e:
# 4.记录清除失败日志
self.logger.error(f"清除对话历史失败: {e}")
self.mysql_client.connect.rollback()
return False
# 8.主流程：问题 -> 答案
def query(self, query, session_id=None, source_filter=None):
# 0.添加query校验逻辑，防止为空
if not query:
yield "请输入问题!", True
return
# 1.记录开始时间
start_time = time.time()
# 2.读取历史对话
history = self._fetch_recent_history(session_id=session_id) if session_id else []
# 3.先走 FQA 检索
answer, is_need_rag = self.fqa_bm25search.search(query, threshold=(0.85,18.0))
# 4.FQA 命中直接返回
if answer:
duration = time.time() - start_time
logger.info("在 FQA 模块获取到了答案，执行时间: {}".format(duration))
if session_id:
self.update_session_history(session_id=session_id, question=query, answer=answer)
yield answer, True
return
logger.info(f"FQA 未命中，问题：{query}")
# 5.需要时再走 RAG
if is_need_rag:
# 5.1 调用 RAG 生成答案
logger.info(f"尝试查询 RAG 模块，问题：{query}")
collected_answer = ''
for token in self.rag.generate_answer(query, source_filter=source_filter, history=history):
collected_answer += token
yield token, False
# 5.2 更新对话历史
if session_id:
self.update_session_history(session_id=session_id, question=query, answer=collected_answer)
# 5.3 记录耗时
duration = time.time() - start_time
logger.info("在 RAG 系统中获取到了答案，执行时间: {}".format(duration))
yield '', True
else:
# 6.未命中时返回兜底提示
duration = time.time() - start_time
logger.info("未能查询到对应的答案: {}".format(duration))
yield f"未找到对应的答案。请联系客服：{config.CUSTOMER_SERVICE_PHONE}", True
def main():
# 定义主函数，提供命令行交互界面
qa_system = IntegratedQASystem() # 初始化问答系统
# 生成唯一对话 ID
session_id = str(uuid.uuid4())
# 打印欢迎信息
print("\n欢迎使用集成问答系统！")
# 打印对话 ID
print(f"对话ID: {session_id}")
# 打印支持的学科类别
print(f"支持的学科类别：{qa_system.config.VALID_SOURCES}")
# 提示用户输入查询或退出
print("输入查询进行问答，输入 'exit' 退出。")
try:
while True:
time.sleep(1)
# 获取用户输入的查询
query = input("\n输入查询: ").strip()
if query.lower() == "exit":
# 如果用户输入 exit，记录退出日志
logger.info("退出系统")
# 打印退出信息
print("再见！")
# 退出循环
break
# 获取用户输入的学科过滤
source_filter = input(f"请输入学科类别 ({'/'.join(qa_system.config.VALID_SOURCES)}) (直接回车默认不过滤): ").strip()
if source_filter and source_filter not in qa_system.config.VALID_SOURCES:
# 如果学科过滤无效，记录警告日志
logger.warning(f"无效的学科类别 '{source_filter}'，将不过滤")
# 设置为空，忽略过滤
source_filter = None
# 打印答案提示
print("\n答案: ", end="", flush=True)
# 初始化累积答案的字符串
answer = ""
# 迭代 query 方法的生成器
for token, is_complete in qa_system.query(query, source_filter=source_filter, session_id=session_id):
if token:
# 仅当 token 非空时打印
print(token, end="", flush=True)
# 累积答案
answer += token
if is_complete:
# 如果是完整答案或流结束，换行并退出循环
print()
break
# 打印对话历史
history = qa_system.get_session_history(session_id)
print("\n最近对话历史:")
for idx, entry in enumerate(history, 1):
# 按顺序打印历史记录
print(f"{idx}. 问: {entry['question']}\n 答: {entry['answer']}")
except Exception as e:
# 记录系统错误日志
logger.error(f"系统错误: {e}")
# 打印错误信息
print(f"发生错误: {e}")
finally:
# 关闭 MySQL 连接
qa_system.mysql_client.close()
if __name__ == "__main__":
# 如果脚本作为主程序运行，调用 main 函数
main()
2.5 章节小结 ¶
本章展示了优化后的 new_main.py ，通过新增对话历史管理和流式输出功能，显著提升了交互性和实时性。系统结合 FQA 的快速匹配、RAG 系统的复杂查询处理、MySQL 的历史存储，以及流式答案生成，配备健壮的日志记录和错误处理，适合教育场景下的实时问答需求
3 基于 FastAPI 的融合FQA 的RAG系统
3.1 学习目标： ¶
理解如何基于 FastAPI 构建高效的问答系统 API 接口。
掌握 WebSocket 流式输出和 HTTP 接口的实现，优化用户交互体验。
学习前后端交互、前端静态文件服务以及日常问候处理的集成。
3.2 查询流程图 ¶
app.py 是对 new_main.py 的进一步优化，基于 FastAPI 框架构建了 RESTful API 和 WebSocket 接口，集成了 FQA 和 RAG 系统，支持流式输出、对话历史管理和日常问候处理。同时，通过静态文件服务支持前端页面交互，适用于教育场景下的实时问答系统。
以下是基于 FastAPI 的问答系统查询流程图，展示从前端请求到后端响应的处理逻辑。
基于FastAPI的RAG系统整体流程
flowchart LR;
Query([用户问题 query]) --> request[前端发送请求</br>HTTP/WebSocket]
request --> check[规则检索<br/>检查是否为问候语]
check -->|是| GreetingResp[返回问候语回复]
GreetingResp --> ReturnResp
check -->|否| FQA[FQA 检索]
FQA -->|BM25检索，不超过阈值| RAG[RAG 检索]
FQA -->|BM25检索，超过阈值| UpdateHistory
RAG --> StreamGen[流式生成答案]
StreamGen --> UpdateHistory[更新对话历史]
UpdateHistory --> ReturnResp([返回答案和对话历史])
3.3 流程说明 ¶
请求接收 ：前端通过 HTTP POST ( /api/query ) 或 WebSocket ( /api/stream ) 发送查询请求，包含查询内容 ( query )、学科过滤 ( source_filter ) 和会话 ID ( session_id )。
规则检索：检查查询是否为日常问候（如”你好”），若匹配则返回模板化回复。
BM25 搜索 ：使用 BM25 算法搜索 MySQL 知识库，设置 双重相似度阈值 (0.85,18.0)。
答案判断 :
若找到可靠答案（相似度 > (0.85,18.0)），通过 HTTP 一次性返回或 WebSocket 流式返回。
若无可靠答案且需要 RAG，HTTP 接口提示使用 WebSocket，WebSocket 接口流式返回 RAG 答案。
若无需 RAG，返回”未找到答案”。
历史更新 ：将查询和答案存入 MySQL 的 conversations 表，保留最近 5 轮对话。
响应输出 ：HTTP 返回 JSON 格式响应，WebSocket 流式发送 JSON 数据（包含 start 、 token 、 end 或 error 类型）。
3.4 代码介绍 ¶
以下是 app.py 的完整代码，包含逐行注释，详细解析功能与实现逻辑。
"""
基于FastAPI 的问答系统主服务入口。
业务工作流(面试回答)：
1.整个问答系统采用三层架构：规则检索 + FQA + RAG
2.用户提交问题后，首先创建或获取当前会话session
3.第一层是规则检索，匹配问候语，如果匹配则直接返回答案
4.第二层是FQA检索，基于BM25算法检索高频问答数据，如果匹配则直接返回答案
5.第三层是RAG检索，首先从向量数据库中检索相关文档（混合检索 + 重排序），构造提示词，然后调用大模型（千问）获得回答，采用流式输出（WebSocket）返回答案
6.系统记录对话历史，实现多轮上下文问答
代码工作流：
1.创建 FastAPI 应用并配置
2.定义日常问候模板
3.定义请求/响应
4.页面与会话管理接口
5.检查日常问候
6.非流式查询接口
7.WebSocket流式查询接口
功能概览：
1. 提供 HTTP 接口：创建会话、非流式输出、查询/清空历史、健康检查、学科来源查询。
2. 提供 WebSocket 流式输出接口：按 `start`/`token`/`end`/`error`事件协议返回增量结果。
3. 内置问候语规则短路：命中常见寒暄时直接返回固定答复，减少模型调用开销。
4. 集成 `IntegratedQASystem`：先走 FQA(BM25) 检索，必要时切换到 RAG 流式生成。
5. 挂载静态资源目录 `static`，用于前端页面与资源访问。
运行方式：
-直接执行 `python app.py` 启动 Uvicorn。
-通过环境变量 `HOST`、`PORT` 配置监听地址与端口。
注意:
1.session会话 与 conversation对话的区别: session就是 临时对话，存储在内存中，conversation对话存储在 数据库中。
session应该翻译为 临时对话，但是 翻译成了 会话，导致难以理解。
面试时，如何避免术语混乱：第一性原理，用英文！
"""
# 1 导入依赖：按“框架 -> 标准库 -> 项目模块”分组
from fastapi import FastAPI, WebSocket, HTTPException, Query, Depends
from fastapi.responses import StreamingResponse, FileResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from starlette.websockets import WebSocketDisconnect
import os
# 异步调用模块
import asyncio
import json
import uuid
import time
import re
from typing import Optional, List, Dict, Any
from pydantic import BaseModel
from new_main import IntegratedQASystem
from base.logger import logger
# 2 创建 FastAPI 应用并配置
# 2.1 创建 FastAPI 应用实例
app = FastAPI(title="问答系统API", description="集成MySQL和RAG的智能问答系统")
# 2.2 配置跨域：开发阶段允许全部来源，生产建议收敛白名单
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 2.3 准备静态目录并挂载
os.makedirs("static", exist_ok=True)
app.mount("/static", StaticFiles(directory="static"), name="static")
# 2.4 初始化全局问答系统（应用启动后复用）
qa_system = IntegratedQASystem()
# 3 定义日常问候模板
GREETING_PATTERNS = [
{
"pattern": r"^(你好|您好|hi|hello)",
"response": "你好！我是黑马程序员，专注于为学生答疑解惑，很高兴为你服务！"
},
{
"pattern": r"^(你是谁|您是谁|你叫什么|你的名字|who are you)",
"response": "我是黑马程序员，你的智能学习助手，致力于提供 IT 教育相关的解答！"
},
{
"pattern": r"^(在吗|在不在|有人吗)",
"response": "我在！我是黑马程序员，随时为你解答问题！"
},
{
"pattern": r"^(干嘛呢|你在干嘛|做什么)",
"response": "我正在待命，随时为你解答 IT 学习相关的问题！有什么我可以帮你的？"
},
{
"pattern": r'^(吃了吗|你吃饭了吗|你吃了吗|你吃了什么|你吃啥了|你吃了啥)',
"response": "我不需要吃饭，随时为你解答 IT 学习相关的问题！有什么我可以帮你的？"
}
]
# 4 定义请求/响应
# 4.1 非流式查询请求体
class QueryRequest(BaseModel):
query: str # 用户查询，必填
source_filter: Optional[str] = None # 学科过滤，可选
session_id: Optional[str] = None # 会话 ID，可选
# 4.2 非流式查询响应体（当前接口直接返回 dict，也保留模型便于后续规范化）
class QueryResponse(BaseModel):
answer: str # 答案
is_streaming: bool # 是否为流式返回
session_id: str # 会话 ID
processing_time: float # 处理时间
# 5 页面与会话管理接口
# 5.1 首页：返回前端页面
@app.get("/")
async def read_root():
return FileResponse("static/index.html")
# 5.2 创建新会话：返回 session_id
@app.post("/api/create_session")
async def create_session():
session_id = str(uuid.uuid4())
return {"session_id": session_id}
# 5.3 获取会话历史
@app.get("/api/history/{session_id}")
async def get_history(session_id: str):
try:
history = qa_system.get_session_history(session_id)
return {"session_id": session_id, "history": history}
except Exception as e:
raise HTTPException(status_code=500, detail=f"获取历史记录失败: {str(e)}")
# 5.4 清空会话历史
@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": "历史记录已清除"}
raise HTTPException(status_code=500, detail="清除历史记录失败")
# 6 检查日常问候
# 6.1 命中则直接返回固定回复，未命中返回 None
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
# 7 非流式查询接口
# 7.1 初始化计时与会话
# 7.2 先走问候语短路
# 7.3 再走 FQA（BM25）
# 7.4 若需 RAG，提示前端改用 WebSocket 流式接口
@app.post("/api/query")
async def query(request: QueryRequest):
# 7.1 初始化计时与会话
start_time = time.time()
session_id = request.session_id or str(uuid.uuid4())
# 7.2 问候语短路
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
}
# 7.3 FQA 检索
answer, need_rag = qa_system.fqa_bm25search.search(request.query, threshold=(0.85, 10.0))
# 7.4 RAG 提示（该接口本身不走流式生成）
if need_rag:
return {
"answer": "请使用WebSocket接口获取流式响应",
"is_streaming": True,
"session_id": session_id,
"processing_time": time.time() - start_time
}
return {
"answer": answer,
"is_streaming": False,
"session_id": session_id,
"processing_time": time.time() - start_time
}
# 8 WebSocket 流式输出接口
# 消息协议（与前端约定）：
# - start: 开始
# - token: 增量内容
# - end: 完成
# - error: 错误
@app.websocket("/api/stream")
async def websocket_endpoint(websocket: WebSocket):
# 8.1 建立连接
await websocket.accept()
try:
while True:
# 8.2 接收并解析请求
data = await websocket.receive_text()
request_data = json.loads(data)
query = request_data.get("query")
source_filter = request_data.get("source_filter")
session_id = request_data.get("session_id", str(uuid.uuid4()))
start_time = time.time()
# 8.3 发送 start 事件
if websocket.client_state == websocket.client_state.CONNECTED:
await websocket.send_json({
"type": "start",
"session_id": session_id
})
# 8.4 问候语短路：直接 token + end
greeting_response = check_greeting(query)
if greeting_response:
if websocket.client_state == websocket.client_state.CONNECTED:
await websocket.send_json({
"type": "token",
"token": greeting_response,
"session_id": session_id
})
await websocket.send_json({
"type": "end",
"session_id": session_id,
"is_complete": True,
"processing_time": time.time() - start_time
})
continue # 继续等待下一条消息，不断开连接
# 8.5 调用系统 流式输出答案
# 用队列桥接同步生成器和异步事件循环，实现流式输出：
# - 子线程：执行同步生成器，每产生一个 token 就推入队列
# - 主线程：从队列逐个取出 token 发送给前端，保持流式效果
# 如果直接用 for 循环迭代替换生成器，会阻塞 asyncio 事件循环
gen_queue = asyncio.Queue()
def _run_query():
try:
for item in qa_system.query(query, source_filter=source_filter, session_id=session_id):
gen_queue.put_nowait(item)
finally:
gen_queue.put_nowait(None) # 哨兵值，通知主线程生成结束
task = asyncio.get_event_loop().run_in_executor(None, _run_query)
collected_answer = ""
while True:
item = await gen_queue.get()
if item is None: # 收到哨兵值，生成结束
break
token, is_complete = item
collected_answer += token
# 8.5.1 结束事件
if is_complete:
if websocket.client_state == websocket.client_state.CONNECTED:
await websocket.send_json({
"type": "end",
"session_id": session_id,
"is_complete": True,
"processing_time": time.time() - start_time
})
break
# 8.5.2 发送 token 事件
if token and websocket.client_state == websocket.client_state.CONNECTED:
await websocket.send_json({
"type": "token",
"token": token,
"session_id": session_id
})
# 8.5.3 控制输出速度（10ms）
await asyncio.sleep(0.01)
# 等待线程完成，确保子线程异常能正确传播
await task
except WebSocketDisconnect as e:
# 8.6 客户端断开
logger.error(f"WebSocket disconnected: code={e.code}, reason={e.reason}")
except Exception as e:
# 8.7 运行异常，尽量回传 error 事件
logger.error(f"WebSocket error: {str(e)}")
if websocket.client_state == websocket.client_state.CONNECTED:
await websocket.send_json({
"type": "error",
"error": str(e)
})
finally:
# 8.8 安全关闭连接
try:
if websocket.client_state == websocket.client_state.CONNECTED:
await websocket.close()
except Exception as e:
logger.error(f"Error closing WebSocket: {str(e)}")
# 9 运维辅助接口与本地启动入口
# 9.1 健康检查
@app.get("/health")
async def health_check():
return {"status": "healthy"}
# 9.2 获取支持的学科列表
@app.get("/api/sources")
async def get_sources():
return {"sources": qa_system.config.VALID_SOURCES}
# 9.3 本地运行入口
if __name__ == "__main__":
import uvicorn
host = os.getenv("HOST", "0.0.0.0") # 监听所有地址
port = int(os.getenv("PORT", 8080))
# 运行 FastAPI 应用，监听指定的主机和端口
uvicorn.run("app:app", host=host, port=port, reload=False)
3.5 章节小结 ¶
本章展示了基于 FastAPI 的 app.py ，通过 RESTful API 和 WebSocket 接口实现了高效的问答系统。系统集成了 FQA 和 RAG 的查询功能，支持流式输出、对话历史管理和日常问候处理，结合静态文件服务实现了前后端交互，配备健壮的错误处理和日志记录，适合教育场景下的实时问答需求。
4 FastAPI和Flask的区别和联系(扩展资料)
联系：
它们都是 Python Web 框架，用来创建 API 或 Web 应用。
它们都能接收 HTTP 请求，根据 URL 路由到相应的处理函数，然后返回 HTTP 响应。
Flask 也可以通过一些方式（比如使用 gevent 或 asyncio 扩展）实现类似异步的功能，但不是其天生设计。FastAPI 也可以运行同步代码（它会自动在线程池里运行，避免阻塞主循环）。
4.1 FastAPI和Flask示例 ¶
Flask: 同步顺序
想象一下你在经营一家餐厅：
Flask 的同步路由 (Synchronous): 就像只有一个厨师，而且他一次只能做一个菜。
工作方式：
第一个客人点单（一个请求来了）。
厨师开始做这道菜（处理这个请求）。
关键点： 在这道菜完全做好、端给客人之前，即使第二个客人已经点好单，厨师也不会开始做第二道菜。他必须等第一道菜完成。如果第一道菜是“佛跳墙”（一个很慢的操作，比如等待数据库、等待外部 API、或者等待大模型生成结果），那么后面所有的客人都要排队等着。
代码示例 (简化示意):
# Flask (同步)
from flask import Flask
import time
app = Flask(__name__)
@app.route('/cook_sync')
def cook_sync():
print("开始做菜 A...")
time.sleep(5) # 模拟做一个很慢的菜 (比如等大模型响应)
print("菜 A 做好了!")
return "菜 A 完成"
# 当一个请求访问 /cook_sync 时，服务器会卡住 5 秒，
# 这期间无法处理其他请求（除非你开了多个服务员/进程）。
- 特点： 简单直接，符合我们通常的编程思路（一步接一步）。但是，如果某个步骤很慢（等待 I/O 操作），整个流程就会被阻塞，效率不高，无法同时处理很多“等待中”的任务。
FastAPI 的异步路由 (Asynchronous): 就像一个很厉害的厨师，他可以同时处理多个任务。
异步同时处理多个任务
- 工作方式：
第一个客人点单（请求 1）。厨师开始做菜 A，比如先把米饭放进电饭煲（开始一个需要等待的操作，比如请求大模型）。
关键点： 在等待米饭煮熟（等待大模型响应）的时候，厨师不会闲着。他会立刻去看有没有其他客人的订单（处理请求 2）。
第二个客人点单（请求 2）。厨师开始处理菜 B，比如开始洗菜（执行一些快代码）。
如果菜 B 也需要等待（比如等水烧开），厨师又会去看有没有其他事做（处理请求 3 或查看米饭是否熟了）。
一旦米饭煮熟了（大模型的响应回来了），厨师就会回来继续完成菜 A 的后续步骤。
- 代码示例 (简化示意):
# FastAPI (异步)
from fastapi import FastAPI
import asyncio # Python 的异步库
app = FastAPI()
@app.get('/cook_async')
async def cook_async(): # 注意这里的 async
print("开始做菜 A (异步)...")
await asyncio.sleep(5) # 模拟做一个很慢的菜，但这里用 await
# 表示“在这里等待，但允许服务器去做别的事”
print("菜 A (异步) 做好了!")
return "菜 A (异步) 完成"
# 当一个请求访问 /cook_async 时，服务器在 `await asyncio.sleep(5)` 这里
# 会“暂停”这个任务，转而去处理其他进来的请求。5秒后，当等待结束，
# 服务器会回来继续执行这个任务。这使得服务器能同时处理很多“等待中”的请求。
# # 运行 FastAPI 应用，监听指定的主机和端口
# uvicorn.run("app:app", host=host, port=port, reload=False)
- 特点： 需要使用 async 和 await 关键字。特别擅长处理那些包含等待时间（I/O 密集型，如网络请求、读写文件、数据库查询、等待大模型响应）的任务。同一个厨师（服务器进程）可以同时”照看”很多道菜，大大提高了效率（并发能力）。
4.2 在利用大模型（LLM）部署时怎么选择？ ¶
答案：强烈推荐使用 FastAPI (异步)。
原因：
LLM 调用是典型的 I/O 密集型操作： 当你的 API 调用一个大模型（无论是本地部署的还是云上的 API），你的程序需要等待模型进行计算并返回结果。这个等待时间可能从几百毫秒到几秒甚至更长。
异步能极大提高并发处理能力：
使用 Flask (同步) ，当一个请求在等待 LLM 响应时，处理这个请求的那个服务器进程/线程就完全卡住了，不能为其他用户服务。如果你想同时服务 100 个用户，你可能需要启动 100 个（甚至更多）Flask 工作进程，这会消耗大量内存和 CPU 资源。
使用 FastAPI (异步) ，当一个请求 await LLM 的响应时，服务器会把这个任务暂时挂起，然后立即去处理其他用户的请求。当 LLM 的响应回来后，服务器再继续处理之前挂起的任务。这样，一个 FastAPI 进程就能同时处理成百上千个“等待 LLM 响应”的并发请求，资源利用率极高。
FastAPI 的内置特性很方便： FastAPI 基于 Pydantic 进行自动的数据校验和文档生成（Swagger UI / ReDoc）。这对于定义清晰的 LLM 输入（如 prompt、参数）和输出结构非常有帮助，能减少很多模板代码，提高开发效率和接口质量。
简单来说：
如果你希望你的 LLM 应用能够同时服务很多用户，并且高效地利用服务器资源， FastAPI 的异步特性是巨大的优势 。
如果你只是做一个内部小工具，用户量极少，或者你的团队完全不熟悉异步编程且不愿意学习，用 Flask 也能实现功能，但性能和并发能力会差很多，需要更多服务器资源来弥补。
结论： 对于需要调用大模型的场景，FastAPI 的异步模型几乎是必然的选择，因为它能更好地应对 LLM 调用带来的等待时间，实现高并发、高性能的服务。
版权声明
本文为 程序员青阳 原创文章，遵循 CC BY-NC-SA 4.0 版权协议，转载请附上原文链接及本声明。