系列文章：本篇是第 1 篇 / 共 4 篇

项目地址：GitHub - RAG_Agent_project

技术栈：Python 3.11 + FastAPI + LangChain + Milvus + MySQL + Redis + BGE-M3 + Qwen3.7-Max

---

[引言]

做过 RAG 的同学应该都有体会：用户问一句"什么是决策"，你非要走一遍 embedding → 向量检索 → rerank → LLM 生成，等 3-5 秒才给出一个教科书上原原本本的答案。

这不是技术先进，这是浪费算力。

我做这个问答系统的时候，第一件事就是想：怎么让简单问题不走 RAG？ 答案是双层 Pipeline —— 第一层 BM25 精确匹配，50 毫秒搞定；命中不了再走第二层语义检索。实测下来，60% 的问题在第一层就解决了。

---

整个系统长什么样

先上一张全景图。整个系统分 5 层，从上到下职责清晰：

层级	模块	核心职责
前端层	static/ (login.html + index.html)	登录注册、聊天界面、WebSocket 流式接收
API 层	rag_api.py (FastAPI)	REST 接口 + WebSocket 流式推送 + JWT 鉴权
编排层	main_plus.py (IntegratedQASystem)	BM25 FAQ 短路 → RAG 管道 → 历史管理
RAG 核心层	rag_qa/core/	意图识别 → 策略选择 → 向量检索 → LLM 生成
基础设施层	MySQL + Redis + Milvus	数据存储、缓存加速、向量数据库

为什么要分这么多层？因为每层的迭代节奏不一样。API 层可能一周改一次，RAG 核心层可能一天改三次。解耦之后各改各的，不会牵一发而动全身。

---

核心设计：能精确匹配的就别走语义检索

这是整个系统最核心的思路。画成流程图就是上面这张图，代码实现其实很简洁：

# mysql_qa/core_search.py
class FAQSearchEngine:
    def search(self, query: str) -> dict:
        # 1. jieba 分词
        tokens = list(jieba.cut(query))
        
        # 2. BM25 打分
        scores = self.bm25.get_scores(tokens)
        
        # 3. Softmax 归一化到 [0,1]
        normalized = softmax(scores)
        
        # 4. 阈值判断
        if normalized.max() >= self.threshold:  # 阈值 0.75
            # 命中！直接返回 Redis 缓存的答案
            return {"answer": self.redis_cache.get(top_idx), "source": "faq"}
        else:
            # 未命中，走 RAG
            return {"need_rag": True}

这里有 3 个让我踩了坑的设计点，展开说说。

第一个：Redis 双重缓存。 FAQ 的问题和答案都缓存在 Redis 里（TTL=180s），不用每次查询都重建 BM25 索引。你可能觉得 BM25 索引构建很快，但当 FAQ 库有几百条、每条都要分词计算的时候，这个缓存能省掉不少时间。

第二个：Softmax 归一化。 原始 BM25 分数的范围是不固定的——跟文档长度、词频分布都有关系，直接设阈值没意义。用 Softmax 映射到 [0,1] 之后，阈值 0.75 才有稳定的语义："模型有 75% 以上的把握认为匹配到了"。

第三个：阈值 0.75 怎么来的。 这个不是拍脑袋定的，是实验调出来的。一开始用 0.85，结果太严了——大量"什么是XX"这种简单问题漏过 FAQ 层进入 RAG，响应时间从 50ms 直接飙到 3-5 秒。降到 0.65 又太松，误匹配率上去了。0.75 是命中率和准确率的平衡点，实测 FAQ 命中率 62%。

---

BM25 没命中，然后呢？

进入 RAG 管道之前，还有一个小优化——热门问题缓存。

MySQL 里有张 hot_questions 表，记录每个问题被问了多少次。被问过 3 次以上的问题，直接把答案缓存起来，下次再问就跳过整个 RAG 流程。

这个优化的逻辑很简单：考试场景下高频问题占比特别大。"什么是决策"、"泰勒的科学管理理论"这种问题，一天能被问几十次。每次都调 LLM 生成一遍，纯属浪费钱。

完整管道是这样的：

热门问题缓存 → 意图识别 → 策略选择 → 向量检索+精排 → LLM 生成 → 记录问答

其中意图识别、策略选择、向量检索这些，后面几篇会一个一个拆。

---

不是所有输入都应该走 RAG

"你好"、"你是谁"、"帮我写个总结" —— 这些走 RAG 纯属浪费。

# rag_qa/core/intent_recognizer.py
class IntentRecognizer:
    def recognize(self, query: str) -> str:
        # 第1步：规则匹配（O(1)，毫秒级）
        if any(kw in query for kw in ["你好", "谢谢", "你是谁"]):
            return "闲聊"
        if any(kw in query for kw in ["帮我写", "总结一下", "翻译"]):
            return "任务型"
        
        # 第2步：LLM 分类（temperature=0.1，保证确定性）
        result = self.llm.invoke(intent_prompt(query))
        return result  # "知识问答" / "闲聊" / "任务型"

规则优先 + LLM 兜底。90% 的闲聊靠规则拦截（毫秒级），剩下边界 case 交给 LLM 分类。只有"知识问答"意图才进入 RAG 管道。

这里有个细节：LLM 分类时 temperature=0.1，不是 0。完全设为 0 有时候会让模型陷入确定性死循环，0.1 留一点点随机性，效果更好。

---

API 层怎么设计的

POST /api/auth/register    # 注册（无需认证）
POST /api/auth/login       # 登录，返回 JWT
POST /api/auth/redeem      # 兑换邀请码（需 JWT）
GET  /api/auth/usage       # 查询剩余配额（需 JWT）
​
POST /api/query            # BM25 FAQ 查询（需 JWT）
WS   /api/stream           # RAG 流式问答（JWT 通过 query param 传递）
​
POST /api/create_session   # 创建会话（生成 UUID）
GET  /api/history/{id}     # 获取对话历史
DELETE /api/history/{id}   # 清除历史

有个值得提的点：WebSocket 不支持自定义 Header，所以 JWT 是通过 ?token=xxx query param 传递的。这不是最佳实践（token 会出现在 URL 里），但 WebSocket 场景下这是最常用的妥协方案。

另一个是 WebSocket 流式响应——用户像 ChatGPT 一样逐字看到 LLM 输出，而不是干等 3-5 秒看一个完整答案。体验差距非常大。

---

项目目录结构

RAG_Agent_project/
├── main_plus.py                  # 核心编排：IntegratedQASystem
├── rag_api.py                    # FastAPI 接口层
├── config/config.yaml            # 全局配置（阈值、模型路径、数据库连接）
├── auth/auth_service.py          # JWT + 邀请码 + 配额
├── static/                       # 前端 (login.html + index.html)
├── rag_qa/
│   ├── core/
│   │   ├── rag_system.py         # RAG 主系统
│   │   ├── vector_store.py       # Milvus + 混合检索
│   │   ├── strategy_selector.py  # 4 策略选择器（下篇详解）
│   │   ├── intent_recognizer.py  # 意图识别
│   │   ├── prompts.py            # Prompt 模板
│   │   ├── process_documents.py  # 父子块切分（第3篇详解）
│   │   └── llm_provider.py       # LLM 单例
│   ├── rag_assessment/           # RAGAS 评估
│   ├── edu_document_loaders/     # 多格式 OCR 加载器
│   ├── edu_text_spliter/         # 中文切分器
│   └── models/bge-m3/            # BGE-M3 模型
├── mysql_qa/core_search.py       # BM25 FAQ 引擎
└── scripts/                      # init_db / import_faq / smoke_test

---

技术选型：为什么选这些

技术	选择	一句话理由
Embedding	BGE-M3	稠密+稀疏双向量，中文 SOTA
向量数据库	Milvus	原生混合检索 + WeightedRanker
LLM	Qwen3.7-Max	中文强，API 稳定
Web	FastAPI	原生 async + WebSocket
缓存	Redis	BM25 索引 + 热门问题双重缓存

---

小结

本篇覆盖了系统的骨架：5 层架构 → 双层问答 Pipeline → BM25 FAQ 短路 → 意图识别 → API 设计。

核心设计理念就一句话：能快则快，该深则深。60% 的问题在 BM25 层 50 毫秒就解决了，只有 40% 的复杂问题才走完整 RAG 管道花 3-5 秒。

下一篇聊 RAG 管道内部的事 —— 4 种自适应检索策略。系统怎么根据问题类型自动选择直接检索、HyDE、子查询拆分、回溯简化，让每种问题都找到最优检索路径。

项目地址：GitHub - jlu55404-art/RAG_Agent_project: RAG问答系统 · GitHub

有帮助欢迎 Star 支持！问题评论区交流～