一、为什么 RAG 项目的核心优化点在“检索”

在 RAG 系统中，完整链路通常可以拆成三个阶段：

用户问题
  ↓
检索 Retrieval
  ↓
增强 Augmentation
  ↓
生成 Generation
  ↓
最终回答

很多人在做 RAG 项目时，容易把注意力放在“大模型回答得好不好”上，但实际上，RAG 系统回答质量的上限，很大程度取决于 检索阶段是否把正确资料找出来。

如果检索阶段没有找到正确资料，那么后面的大模型再强，也只能基于错误、不完整或者无关的上下文进行回答。

也就是说：

▎ RAG 不是单纯让大模型回答问题，而是让大模型基于检索到的资料回答问题。
▎ 如果检索错了，生成阶段就会被错误上下文带偏。

对于当前这个 RAG 智能客服项目来说，核心代码集中在：

online_chroma/
├── config_data.py        # 检索与模型配置
├── vector_stores.py      # 文档加载、切片、向量库、Retriever
├── rag.py                # RAG Chain、Prompt、问答逻辑
├── app_qa.py             # Streamlit 页面、文件上传、问答入口
├── api.py                # FastAPI 接口
└── file_history.py       # 会话历史

其中与检索质量关系最密切的文件是：

vector_stores.py
rag.py
config_data.py
app_qa.py

当前项目已经实现了基础 RAG 闭环：

文件上传
  ↓
文档解析
  ↓
文本切片
  ↓
Embedding 向量化
  ↓
写入 Chroma
  ↓
用户提问
  ↓
向量检索 Top K
  ↓
拼接上下文
  ↓
大模型生成回答

但如果要让这个项目从“能跑通”升级到“检索更准、回答更稳、可维护性更强”，就需要围绕检索链路做系统优化。

---
二、当前项目的 RAG 检索现状分析

2.1 当前检索配置

当前项目的配置文件是：

online_chroma/config_data.py

核心配置如下：

collections_name = "rag"
persist_directory = "./chroma_db"

chunk_size = 1000
chunk_overlap = 100
separators = ["\n\n", "\n", "!", "。", "，", ",", "."]

max_split_char_number = 1000
similarity_threshold = 2

embedding_model_name = "text-embedding-v4"
chat_model_name = "qwen3-max"

这里有几个关键点：

其中比较需要注意的是：

similarity_threshold = 2

这个变量名叫 similarity_threshold，看起来像“相似度阈值”，但实际上在代码中是作为 k 使用的，也就是返回前几个相似文档。

在 vector_stores.py 中：

def get_retriever(self):
    return self.vector_store.as_retriever(search_kwargs={"k": config.similarity_threshold})

所以当前实际含义是：

每次从向量库中返回最相似的 2 个文档片段

这属于一个基础可用配置，但在复杂知识库场景下可能不够。

---
2.2 当前知识库入库逻辑

当前项目的文档入库逻辑在：

online_chroma/vector_stores.py

核心代码如下：

def add_documents(self, documents):
    if not documents:
        return 0

    split_docs = self.text_splitter.split_documents(documents)
    self.vector_store.add_documents(split_docs)
    return len(split_docs)

处理流程是：

Document 列表
  ↓
RecursiveCharacterTextSplitter 切片
  ↓
Chroma.add_documents()
  ↓
Embedding 向量化
  ↓
写入 Chroma

优点是实现简单，能够快速完成知识库构建。

但不足也比较明显：

1. 没有对切片质量做评估；
2. 没有对文档片段增加更丰富的元数据；
3. 没有对重复文件或重复内容做系统去重；
4. 没有对不同类型文档采用不同切片策略；
5. 没有为后续过滤检索、来源引用、文件删除留下足够结构化信息。

---
2.3 当前检索逻辑

当前检索器生成代码如下：

def get_retriever(self):
    return self.vector_store.as_retriever(search_kwargs={"k": config.similarity_threshold})

这表示当前项目使用的是最基础的向量相似度检索。

其逻辑可以理解为：

用户问题
  ↓
Embedding 模型转换为问题向量
  ↓
在 Chroma 中计算问题向量与文档向量的相似度
  ↓
返回相似度最高的 Top K 文档片段

当前方案属于最标准的 RAG 初始版本。

优点是：

- 实现简单；
- 速度较快；
- 适合小型知识库；
- 适合 Demo 和基础项目。

缺点是：

- 只依赖语义向量，容易漏掉关键词强匹配内容；
- Top K 固定为 2，无法适应不同问题复杂度；
- 没有相似度阈值过滤；
- 没有重排序；
- 没有混合检索；
- 没有基于文件来源、文档类型、时间等元数据过滤；
- 没有查询改写；
- 没有多轮对话下的问题独立化处理。

---
2.4 当前 RAG Chain 逻辑

当前 RAG 核心链路在：

online_chroma/rag.py

核心代码如下：

chain = (
    {
        "input": RunnablePassthrough(),
        "context": RunnableLambda(temp1) | retriever | format_document
    } | RunnableLambda(temp2) | self.prompt_template | print_prompt | self.chat_model | StrOutputParser()
)

可以拆成：

用户输入
  ├── 原样保留为 input
  └── 提取 input 后进入 retriever
          ↓
       检索文档
          ↓
       format_document 格式化
          ↓
组合 input + context + history
          ↓
Prompt
          ↓
ChatTongyi
          ↓
StrOutputParser

当前逻辑已经具备 RAG 基础能力，但在检索增强方面仍有很多可优化空间。

---
三、RAG 检索优化的总体方向

RAG 检索优化不能只看某一个点，而应该从完整链路看：

文档进入知识库之前
  ↓
文档如何切片
  ↓
文档如何存储
  ↓
用户问题如何处理
  ↓
检索器如何召回
  ↓
检索结果如何筛选
  ↓
检索结果如何重排序
  ↓
上下文如何拼接
  ↓
模型如何使用上下文

可以将 RAG 检索优化分为以下几个层面：

1. 文档预处理优化
2. 文本切片优化
3. 元数据优化
4. Embedding 模型优化
5. Top K 与相似度阈值优化
6. 混合检索优化
7. 重排序 Rerank 优化
8. 查询改写 Query Rewrite 优化
9. 多轮对话检索优化
10. 上下文组装优化
11. Prompt 约束优化
12. 检索评估与调试优化
13. 前端可视化与可配置化优化

下面结合当前项目逐一分析。

---
四、优化方向一：文档预处理优化

4.1 为什么要做文档预处理

RAG 的知识库不是把原始文件直接丢进去就可以。很多文件中会包含：

- 空行；
- 页眉页脚；
- 页码；
- 重复标题；
- 表格错乱文本；
- 无意义符号；
- 目录；
- 扫描件乱码；
- Excel 中的空列空行；
- PDF 中断裂的句子。

如果这些内容直接进入向量库，会影响 Embedding 表达，也会影响检索结果。

例如，一个 PDF 解析后可能变成：

第 1 页
公司制度说明
版权所有
1
第一章 总则
...
第 2 页
公司制度说明
版权所有
2
...

如果每一页都有重复页眉页脚，这些重复内容会被向量化并进入检索，导致用户检索时返回很多无意义片段。

---
4.2 当前项目情况

当前项目在 vector_stores.py 中已经支持多格式加载：

if ext == ".pdf":
    return self._load_pdf_langchain(file_path)
elif ext == ".docx":
    return self._load_docx_langchain(file_path)
elif ext in [".xlsx", ".xls"]:
    return self._load_excel_langchain(file_path)
elif ext in [".txt", ".md", ".json"]:
    return self._load_text_file(file_path)

但是加载后的文本没有经过统一清洗，直接进入：

split_docs = self.text_splitter.split_documents(documents)

所以可以增加一个预处理环节。

---
4.3 建议增加文本清洗方法

可以在 VectorStoreService 中增加：

import re

def clean_text(self, text: str) -> str:
    if not text:
        return ""

    # 统一换行
    text = text.replace("\r\n", "\n").replace("\r", "\n")

    # 去除多余空格
    text = re.sub(r"[ \t]+", " ", text)

    # 去除连续空行
    text = re.sub(r"\n{3,}", "\n\n", text)

    # 去除首尾空白
    text = text.strip()

    return text

然后在 add_documents() 中处理：

def add_documents(self, documents):
    if not documents:
        return 0

    cleaned_docs = []
    for doc in documents:
        cleaned_text = self.clean_text(doc.page_content)
        if cleaned_text:
            doc.page_content = cleaned_text
            cleaned_docs.append(doc)

    split_docs = self.text_splitter.split_documents(cleaned_docs)
    self.vector_store.add_documents(split_docs)
    return len(split_docs)

这样可以保证进入切片器之前的文本更干净。

---
4.4 进一步优化：针对不同文件类型清洗

对于不同文件类型，可以采用不同策略。

TXT / MD

保留标题、列表、段落结构。

PDF

重点清理：

- 页码；
- 页眉；
- 页脚；
- 多余换行；
- 断行句子。

Excel

建议不要简单使用：

df.to_string(index=False)

因为 Excel 表格直接转字符串可能造成列含义不清晰。

可以优化为：

def excel_rows_to_text(self, df):
    lines = []
    for index, row in df.iterrows():
        parts = []
        for col in df.columns:
            value = row[col]
            if str(value) != "nan":
                parts.append(f"{col}: {value}")
        if parts:
            lines.append("；".join(parts))
    return "\n".join(lines)

这样 Excel 中每一行都会转换成：

商品名称: T恤；适合体重: 150-180斤；推荐尺码: XL

比原来的表格字符串更适合 Embedding。

---
五、优化方向二：文本切片优化

5.1 文本切片对检索质量的影响

切片是 RAG 中非常关键的一步。

如果切片太大：

- 一个片段包含多个主题；
- 向量语义不够集中；
- 检索命中后上下文冗余；
- 大模型容易被无关内容干扰。

如果切片太小：

- 语义不完整；
- 缺少上下文；
- 用户问题匹配不到完整答案；
- 模型需要的信息被切断。

所以切片的目标是：

▎ 每个 chunk 尽量表达一个相对完整、独立、语义集中的知识点。

---
5.2 当前项目切片配置

当前配置是：

chunk_size = 1000
chunk_overlap = 100
separators = ["\n\n", "\n", "!", "。", "，", ",", "."]

当前切片器初始化代码是：

self.text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=config.chunk_size,
    chunk_overlap=config.chunk_overlap,
    separators=config.separators,
)

这个配置适合通用文本，但还可以进一步优化。

---
5.3 优化建议一：根据文档类型动态切片

不同类型文档适合不同切片策略。

例如：

当前项目所有文件都使用同一个切片器：

RecursiveCharacterTextSplitter(...)

可以改造成根据文件类型选择切片方式。

示例：

def get_text_splitter(self, file_ext=None):
    if file_ext == ".md":
        return RecursiveCharacterTextSplitter(
            chunk_size=800,
            chunk_overlap=100,
            separators=["\n## ", "\n### ", "\n\n", "\n", "。", "，"]
        )
    elif file_ext in [".xlsx", ".xls"]:
        return RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n", "；", "，"]
        )
    else:
        return RecursiveCharacterTextSplitter(
            chunk_size=config.chunk_size,
            chunk_overlap=config.chunk_overlap,
            separators=config.separators,
        )

然后在 add_file() 中根据文件后缀选择切片器。

---
5.4 优化建议二：调整 chunk_size

当前：

chunk_size = 1000
chunk_overlap = 100

对于“尺码推荐”这种规则型知识库来说，1000 字符可能偏大。因为尺码规则通常比较短，如果一个 chunk 中混合了多个尺码段，检索时可能返回不够精准。

例如一个 chunk 中同时包含：

体重 100-120斤推荐 M
体重 120-150斤推荐 L
体重 150-180斤推荐 XL
体重 180-210斤推荐 XXL

当用户问“180 斤穿什么尺码”时，这个 chunk 虽然相关，但它包含多个候选规则。模型可能需要进一步判断。

如果切片更细，例如每条规则一个 chunk：

体重 180-210斤推荐 XXL

检索会更精准。

因此对于当前项目，可以考虑：

chunk_size = 500
chunk_overlap = 50

或者针对规则文档使用更小切片：

chunk_size = 300
chunk_overlap = 30

---
5.5 优化建议三：保留标题上下文

很多文档的标题对理解内容非常重要。

例如：

## 男装尺码推荐

体重 180 斤建议 XXL。

如果切片后只剩：

体重 180 斤建议 XXL。

模型可能不知道这是男装还是女装。

所以可以在切片时保留标题上下文。

一种简单方案是：在加载 Markdown 或文本时，识别最近标题，将标题写入 metadata 或拼接到 chunk 前面。

例如：

metadata = {
    "source": file_path,
    "file_name": os.path.basename(file_path),
    "section_title": "男装尺码推荐"
}

或者 chunk 内容变成：

标题：男装尺码推荐
内容：体重 180 斤建议 XXL。

这样检索和生成都会更准确。

---
六、优化方向三：元数据 Metadata 优化

6.1 Metadata 的作用

在 RAG 中，metadata 非常重要。它不仅可以用于展示来源，还可以用于过滤检索。

例如：

metadata = {
    "source": "尺码推荐.txt",
    "file_name": "尺码推荐.txt",
    "file_type": "txt",
    "upload_time": "2026-06-15 10:00:00",
    "category": "尺码推荐",
    "chunk_index": 3
}

有了 metadata，就可以实现：

- 按文件过滤；
- 按业务类别过滤；
- 展示答案来源；
- 删除某个文件的所有向量；
- 调试检索结果；
- 对不同知识库分组检索。

---
6.2 当前项目元数据情况

当前 vector_stores.py 中在 add_file() 里有：

for doc in docs:
    if metadata:
        doc.metadata.update(metadata)
    doc.metadata.setdefault('source', file_path)
    doc.metadata.setdefault('file_name', os.path.basename(file_path))

这已经包含基础来源信息，但还不够完整。

---
6.3 建议增加更多 metadata

可以改为：

from datetime import datetime

def build_metadata(self, file_path, extra_metadata=None):
    _, ext = os.path.splitext(file_path)
    metadata = {
        "source": file_path,
        "file_name": os.path.basename(file_path),
        "file_type": ext.lower().replace(".", ""),
        "upload_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }

    if extra_metadata:
        metadata.update(extra_metadata)

    return metadata

然后在 add_file() 中使用：

base_metadata = self.build_metadata(file_path, metadata)

for doc in docs:
    doc.metadata.update(base_metadata)

切片后还可以给每个 chunk 增加编号：

split_docs = self.text_splitter.split_documents(documents)

for index, doc in enumerate(split_docs):
    doc.metadata["chunk_index"] = index
    doc.metadata["chunk_id"] = f"{doc.metadata.get('file_name', 'unknown')}_{index}"

这样后续调试时可以清楚知道每个片段来自哪个文件、哪个位置。

---
6.4 支持按文件或类别过滤检索

当前检索是全库检索：

self.vector_store.as_retriever(search_kwargs={"k": 2})

如果知识库变大，不同文件混在一起，可能会检索到无关资料。

可以增加 filter 参数，例如：

def get_retriever(self, top_k=None, filters=None):
    search_kwargs = {
        "k": top_k or config.retrieval_top_k
    }

    if filters:
        search_kwargs["filter"] = filters

    return self.vector_store.as_retriever(search_kwargs=search_kwargs)

调用时可以指定：

retriever = vector_service.get_retriever(
    top_k=5,
    filters={"file_type": "txt"}
)

或者：

filters={"category": "尺码推荐"}

这样可以显著提升大知识库场景下的检索准确率。

---
七、优化方向四：Top K 与相似度阈值优化

7.1 当前问题

当前配置：

similarity_threshold = 2

实际作为 Top K 使用：

search_kwargs={"k": config.similarity_threshold}

存在两个问题：

1. 命名不准确；
2. Top K 固定为 2，灵活性不足。

---
7.2 Top K 太小的问题

如果 k=2，只返回 2 个片段。

对于简单问题，比如：

我体重180斤穿什么码？

2 个片段可能够用。

但对于复杂问题，比如：

我身高178，体重180斤，喜欢宽松一点，应该怎么选尺码？

可能需要同时参考：

- 身高规则；
- 体重规则；
- 宽松版型规则；
- 商品类型规则；
- 男女款规则。

此时只返回 2 个 chunk 可能不够。

---
7.3 Top K 太大的问题

如果 k=10，返回太多片段，也可能带来问题：

- 上下文变长；
- 无关内容变多；
- 大模型注意力被干扰；
- 响应速度变慢；
- Token 成本变高。

所以 Top K 需要根据业务调优。

---
7.4 建议修改配置命名

可以将：

similarity_threshold = 2

改成：

retrieval_top_k = 4
similarity_score_threshold = 0.3

其中：

retrieval_top_k

表示最多返回多少个文档片段。

similarity_score_threshold

表示最低相似度阈值。

---
7.5 使用 similarity_score_threshold 检索

LangChain 的 Chroma retriever 支持不同 search_type。

可以改成：

def get_retriever(self, top_k=None, score_threshold=None):
    return self.vector_store.as_retriever(
        search_type="similarity_score_threshold",
        search_kwargs={
            "k": top_k or config.retrieval_top_k,
            "score_threshold": score_threshold or config.similarity_score_threshold
        }
    )

不过需要注意：不同向量库和 LangChain 版本对分数含义可能有差异，有的是相似度，有的是距离，需要实际测试。

---
7.6 更推荐使用 similarity_search_with_score 做可控过滤

为了更清楚控制，可以不用默认 retriever，而是自己写检索方法：

def similarity_search_with_scores(self, query, top_k=5):
    results = self.vector_store.similarity_search_with_score(query, k=top_k)
    return results

然后手动过滤：

def retrieve_with_threshold(self, query, top_k=5, max_distance=None):
    results = self.vector_store.similarity_search_with_score(query, k=top_k)

    filtered_docs = []
    for doc, score in results:
        doc.metadata["retrieval_score"] = score

        if max_distance is None or score <= max_distance:
            filtered_docs.append(doc)

    return filtered_docs

如果 Chroma 返回的是距离，分数越小越相关，就可以用：

score <= max_distance

这样比直接依赖 retriever 更方便调试。

---
八、优化方向五：混合检索 Hybrid Search

8.1 为什么需要混合检索

当前项目只使用向量检索。

向量检索擅长语义匹配，例如：

用户问：衣服大一点怎么选？
知识库：如果喜欢宽松版型，可以选择大一码。

即使用词不同，向量检索也能匹配。

但向量检索也有弱点：

1. 对数字、型号、编码不够敏感；
2. 对精确关键词不一定稳定；
3. 对短文本问题可能召回不准；
4. 对专有名词可能表现一般。

例如用户问：

SKU-7788 的尺码规则是什么？

这种问题更适合关键词检索。

所以更好的方案是混合检索：

向量检索 + 关键词检索

也就是：

用户问题
  ├── 向量检索：找语义相似内容
  └── 关键词检索：找关键词精确匹配内容
          ↓
       合并去重
          ↓
       重排序
          ↓
       送给大模型

---
8.2 当前项目可以怎么做

当前项目使用 Chroma，Chroma 本身偏向向量检索。可以额外引入 BM25 检索。

LangChain 中有：

BM25Retriever

可以基于文档文本做关键词检索。

---
8.3 增加 BM25 检索思路

需要在向量库服务中保存一份原始切片文档：

self.documents_cache = []

在添加文档时：

split_docs = self.text_splitter.split_documents(documents)
self.documents_cache.extend(split_docs)
self.vector_store.add_documents(split_docs)

然后构建 BM25：

from langchain_community.retrievers import BM25Retriever

def get_bm25_retriever(self, top_k=5):
    retriever = BM25Retriever.from_documents(self.documents_cache)
    retriever.k = top_k
    return retriever

再做混合检索：

def hybrid_search(self, query, vector_k=5, keyword_k=5):
    vector_docs = self.vector_store.similarity_search(query, k=vector_k)

    bm25_retriever = self.get_bm25_retriever(top_k=keyword_k)
    keyword_docs = bm25_retriever.invoke(query)

    merged = []
    seen = set()

    for doc in vector_docs + keyword_docs:
        key = doc.page_content[:100]
        if key not in seen:
            seen.add(key)
            merged.append(doc)

    return merged

这种方式可以同时兼顾：

- 语义相似；
- 关键词精确命中。

---
8.4 更工程化的混合检索

更正式的混合检索一般会使用权重融合，例如：

final_score = vector_score * 0.7 + keyword_score * 0.3

或者使用 RRF（Reciprocal Rank Fusion）算法。

RRF 的核心思想是：
不直接比较不同检索器的原始分数，而是根据排名融合。

公式可以简化理解为：

score = 1 / (k + rank)

排名越靠前，得分越高。

比如：

def rrf_fusion(result_lists, rrf_k=60):
    scores = {}
    docs = {}

    for results in result_lists:
        for rank, doc in enumerate(results):
            key = doc.page_content[:200]
            docs[key] = doc
            scores[key] = scores.get(key, 0) + 1 / (rrf_k + rank + 1)

    sorted_keys = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
    return [docs[key] for key in sorted_keys]

混合检索：

def hybrid_search_with_rrf(self, query, vector_k=5, keyword_k=5, final_k=5):
    vector_docs = self.vector_store.similarity_search(query, k=vector_k)

    bm25_retriever = self.get_bm25_retriever(top_k=keyword_k)
    keyword_docs = bm25_retriever.invoke(query)

    fused_docs = rrf_fusion([vector_docs, keyword_docs])
    return fused_docs[:final_k]

这样可以提升检索召回率。

---
九、优化方向六：Rerank 重排序

9.1 为什么需要 Rerank

基础向量检索通常只负责“召回”，但召回结果未必顺序最优。

例如向量检索返回了 8 个片段：

doc1 相关
doc2 一般相关
doc3 非常相关
doc4 无关
doc5 比较相关
...

如果直接把前 2 个给大模型，可能错过真正最相关的 doc3。

所以常见 RAG 架构会分两步：

第一步：召回 Recall
  从知识库中尽量多找一些候选文档，比如 Top 10

第二步：重排序 Rerank
  使用更精细的模型判断 query 和文档的相关性，选出 Top 3

---
9.2 当前项目没有 Rerank

当前项目是：

Retriever Top K
  ↓
format_document
  ↓
Prompt

没有中间重排序过程。

---
9.3 可以怎么加 Rerank

一种简单方案是让大模型做轻量重排序。

先召回更多文档：

retrieval_top_k = 8
rerank_top_k = 3

然后用一个函数筛选最相关文档。

简单规则版 Rerank

如果暂时不引入额外模型，可以根据关键词重叠做简单排序：

def simple_rerank(self, query, docs, top_k=3):
    query_terms = set(query)

    scored_docs = []
    for doc in docs:
        content = doc.page_content
        overlap = sum(1 for char in query_terms if char in content)
        scored_docs.append((doc, overlap))

    scored_docs.sort(key=lambda x: x[1], reverse=True)
    return [doc for doc, score in scored_docs[:top_k]]

这个方法很简单，但对中文短文本也能起到一定补充作用。

---
9.4 使用 DashScope 或其他 Rerank 模型

如果使用专门的重排序模型，效果会更好。

理想流程：

用户问题
  ↓
向量召回 Top 10
  ↓
Rerank 模型逐条判断相关性
  ↓
选出 Top 3
  ↓
交给大模型生成回答

伪代码：

def rerank_documents(self, query, docs, top_k=3):
    pairs = []
    for doc in docs:
        pairs.append({
            "query": query,
            "document": doc.page_content
        })

    # 调用 rerank 模型，返回相关性分数
    rerank_results = call_rerank_model(pairs)

    scored_docs = []
    for doc, score in zip(docs, rerank_results):
        doc.metadata["rerank_score"] = score
        scored_docs.append((doc, score))

    scored_docs.sort(key=lambda x: x[1], reverse=True)
    return [doc for doc, score in scored_docs[:top_k]]

即使暂时不接入真实 Rerank 模型，也建议在代码结构中预留 Rerank 层。

---
十、优化方向七：查询改写 Query Rewrite

10.1 当前多轮对话下的问题

当前项目已经通过 RunnableWithMessageHistory 支持历史对话。

但是检索时使用的是当前用户输入：

def temp1(value: dict) -> str:
    return value["input"]

也就是说，Retriever 只拿当前这句话去检索。

这在多轮对话中会有问题。

例如：

用户：我体重180斤，身高178
AI：建议选择 XXL
用户：那如果我想宽松一点呢？

第二个问题：

那如果我想宽松一点呢？

单独拿去检索，缺少“体重180斤、身高178、尺码推荐”这些上下文，检索可能不准。

---
10.2 解决方案：将追问改写成独立问题

应该先把用户追问改写成完整问题：

原问题：
那如果我想宽松一点呢？

改写后：
用户身高178cm、体重180斤，如果想穿得宽松一点，应该选择什么尺码？

然后用改写后的问题去检索。

这就是 Query Rewrite，也叫 Contextualize Question。

---
10.3 代码改造思路

可以在 RAGService 中增加一个问题改写链：

@property
def rewrite_prompt(self):
    return ChatPromptTemplate.from_messages(
        [
            ("system", "请根据历史对话，将用户最新问题改写成一个可以独立理解的完整问题。不要回答问题，只输出改写后的问题。"),
            MessagesPlaceholder("history"),
            ("user", "{input}")
        ]
    )

构建链：

def _get_rewrite_chain(self):
    return self.rewrite_prompt | self.chat_model | StrOutputParser()

在 RAG 检索前先改写：

standalone_question = self.rewrite_chain.invoke(
    {"input": prompt, "history": history}
)

然后用：

standalone_question

去检索。

---
10.4 当前项目可做的轻量改造

由于当前项目的 RAG Chain 已经使用 LangChain Runnable，可以把逻辑拆得更清晰一些：

def rewrite_query(self, prompt, history):
    if not history:
        return prompt

    rewrite_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", "请结合历史对话，把用户问题改写为独立完整的问题。只输出改写后的问题。"),
            MessagesPlaceholder("history"),
            ("user", "{input}")
        ]
    )

    chain = rewrite_prompt | self.chat_model | StrOutputParser()
    return chain.invoke({"input": prompt, "history": history})

然后在 chat_sync() 中手动组织：

history = get_history(session_id).messages
query = self.rewrite_query(prompt, history)
docs = self.vector_service.search(query)
context = self.format_documents(docs)

这种方式比完全写在 Runnable 中更容易调试。

---
十一、优化方向八：上下文组装优化

11.1 当前 context 格式

当前项目中：

def format_document(docs: list[Document]):
    if not docs:
        return "无相关参考资料"
    formatted_str = ""
    for doc in docs:
        formatted_str += f"文档片段:{doc.page_content}\n文档元数据:{doc.metadata}\n\n"
    return formatted_str

这个格式可用，但还可以优化。

---
11.2 当前格式的问题

现在会把完整 metadata 直接放进去：

文档元数据:{doc.metadata}

metadata 中可能包含：

source, file_name, file_type, upload_time, chunk_index

直接给模型可能会显得杂乱。

更好的方式是格式化为更清晰的引用块。

---
11.3 优化后的 context 格式

可以改为：

def format_documents(self, docs: list[Document]):
    if not docs:
        return "未检索到相关参考资料。"

    formatted = []
    for i, doc in enumerate(docs, start=1):
        source = doc.metadata.get("file_name", doc.metadata.get("source", "未知来源"))
        score = doc.metadata.get("retrieval_score", None)

        item = f"【资料{i}】\n来源：{source}\n内容：{doc.page_content}"

        if score is not None:
            item += f"\n检索分数：{score}"

        formatted.append(item)

    return "\n\n".join(formatted)

输出变成：

【资料1】
来源：尺码推荐.txt
内容：体重 180 斤的用户建议选择 XXL 尺码。

【资料2】
来源：尺码推荐.txt
内容：如果喜欢宽松版型，可以选择大一码。

这种格式更适合模型理解，也方便后续要求模型引用来源。

---
十二、优化方向九：Prompt 约束优化

12.1 当前 Prompt

当前项目的 Prompt 是：

("system", "以我提供的资料为主简洁和专业的回答用户问题.参考资料:{context}")

这个 Prompt 能用，但约束不够强。

它没有明确要求：

- 如果资料不足，不要编造；
- 优先使用检索资料；
- 回答要引用来源；
- 如果资料冲突，要说明；
- 如果问题与知识库无关，要提示用户。

---
12.2 优化后的 Prompt

可以改成：

@property
def prompt_template(self):
    return ChatPromptTemplate.from_messages(
        [
            (
                "system",
                """
你是一个专业的 RAG 智能客服助手。请严格基于【参考资料】回答用户问题。

要求：
1. 优先使用参考资料中的内容回答。
2. 如果参考资料中没有足够信息，请明确说明“知识库中没有找到足够信息”，不要编造。
3. 如果参考资料中存在多个可能答案，请说明判断依据。
4. 回答要简洁、准确、结构清晰。
5. 如有来源信息，请在回答末尾列出“参考来源”。

【参考资料】
{context}
"""
            ),
            ("system", "以下是历史对话："),
            MessagesPlaceholder("history"),
            ("user", "{input}")
        ]
    )

这样可以明显减少幻觉。

---
12.3 增加“无资料”处理逻辑

如果检索为空，现在是：

return "无相关参考资料"

但 Prompt 没有强制模型在无资料时拒答。

可以增加判断：

if not docs:
    return "未检索到相关参考资料。请明确告诉用户知识库中没有找到相关内容。"

同时 Prompt 中明确要求：

如果参考资料提示“未检索到相关参考资料”，请不要使用常识编造业务答案。

---
十三、优化方向十：检索调试与可观测性

13.1 为什么需要检索调试

RAG 项目最常见的问题是：

用户觉得回答不准

但原因可能有很多：

1. 文档没有入库；
2. 文档切片不合理；
3. Embedding 不准确；
4. Top K 太小；
5. 检索到了错误片段；
6. Prompt 没有限制模型；
7. 模型忽略了资料；
8. 历史对话干扰了问题。

如果没有调试信息，就很难知道问题出在哪里。

---
13.2 当前项目缺少检索可视化

当前 Streamlit 页面只显示最终回答，没有展示：

- 检索到了哪些片段；
- 每个片段来源；
- 检索分数；
- 使用了几个 chunk；
- 最终 Prompt 长什么样。

虽然 rag.py 中有：

def print_prompt(prompt):
    print("*" * 20)
    print(prompt.to_string())
    print("*" * 20)
    return prompt

但这只是打印到控制台，不适合页面调试。

---
13.3 建议在页面增加“检索调试模式”

可以在 app_qa.py 侧边栏增加：

debug_mode = st.toggle(
    "🔍 显示检索调试信息",
    value=False,
    help="开启后展示检索到的文档片段和来源"
)
st.session_state["debug_mode"] = debug_mode

然后 RAGService 提供一个带检索详情的方法：

def retrieve_debug(self, query, top_k=5):
    docs = self.vector_service.similarity_search_with_scores(query, top_k=top_k)

    debug_results = []
    for doc, score in docs:
        debug_results.append({
            "content": doc.page_content,
            "metadata": doc.metadata,
            "score": score
        })

    return debug_results

页面展示：

if st.session_state.get("debug_mode"):
    debug_results = rag.retrieve_debug(prompt)

    with st.expander("🔍 检索调试信息"):
        for i, item in enumerate(debug_results, start=1):
            st.markdown(f"### 片段 {i}")
            st.write(item["content"])
            st.json(item["metadata"])
            st.write(f"Score: {item['score']}")

这样可以非常直观地观察：

用户问题到底检索到了什么？
为什么模型这么回答？
是不是知识库没有命中？

---
十四、优化方向十一：知识库去重与更新机制

14.1 当前项目的问题

当前完整版本 online_chroma 中，文件上传时只在本次上传列表中判断重复文件名：

existing_files = set()
...
if file_name in existing_files:
    st.warning(f"⚠️ 跳过重复文件: {file_name}")
    continue

这只能避免一次上传中重复选择同名文件，不能避免：

- 同一个文件多次上传；
- 同内容不同文件名；
- 文件更新后旧 chunk 仍然存在；
- 删除文件后向量库中仍保留旧内容。

根目录早期 Demo knowledgebase.py 中有 MD5 去重：

def get_string_md5(input_str: str, encoding="utf-8"):
    str_bytes = input_str.encode(encoding=encoding)
    md5_obj = hashlib.md5()
    md5_obj.update(str_bytes)
    md5_hex = md5_obj.hexdigest()
    return md5_hex

这个思路可以迁移到完整版项目。

---
14.2 增加文件 MD5

可以在 vector_stores.py 中增加：

import hashlib

def get_file_md5(self, file_path):
    md5 = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            md5.update(chunk)
    return md5.hexdigest()

入库时：

file_md5 = self.get_file_md5(file_path)

metadata 中加入：

"file_md5": file_md5

---
14.3 防止重复入库

可以在入库前检查 Chroma 中是否已有相同 MD5：

def file_exists(self, file_md5):
    data = self.vector_store.get(where={"file_md5": file_md5})
    return bool(data.get("ids"))

然后：

if self.file_exists(file_md5):
    return 0

这样可以避免重复上传同一文件。

---
14.4 支持按文件删除旧向量

如果用户上传了新版文件，应该先删除旧版本。

可以通过 metadata 删除：

def delete_file_vectors(self, file_name):
    self.vector_store.delete(where={"file_name": file_name})

然后重新入库：

self.delete_file_vectors(os.path.basename(file_path))
self.add_file(file_path)

这对知识库更新非常重要。

---
十五、优化方向十二：多知识库与分类检索

15.1 当前项目是单 collection

当前：

collections_name = "rag"

所有文件都进入同一个 Chroma collection。

对于小项目没问题，但如果以后知识库扩展为：

- 尺码推荐；
- 售后规则；
- 商品介绍；
- 物流政策；
- 常见问题；
- 内部制度；

全部混在一个 collection 里，容易出现跨领域误检索。

---
15.2 优化方案一：metadata 分类

上传文件时让用户选择分类：

category = st.selectbox(
    "知识库分类",
    ["尺码推荐", "售后政策", "商品介绍", "物流说明", "其他"]
)

入库时传入：

metadata={"category": category}

检索时可以按分类过滤：

filters={"category": selected_category}

---
15.3 优化方案二：多 collection

也可以不同业务使用不同 collection：

collections_name = "size_recommendation"

或者动态创建：

def __init__(self, embedding, collection_name=None):
    self.vector_store = Chroma(
        collection_name=collection_name or config.collections_name,
        embedding_function=self.embedding,
        persist_directory=config.persist_directory
    )

适合知识库之间边界非常明确的场景。

---
十六、结合当前项目的推荐优化路线

对于当前项目，不建议一次性把所有高级能力都加上。更合理的是分阶段优化。

---
第一阶段：基础检索质量优化

目标：不改变整体架构，快速提升检索准确率。

建议修改：

1. 修改配置命名

在 config_data.py 中：

retrieval_top_k = 4
rerank_top_k = 3
similarity_score_threshold = None

替换原来的：

similarity_threshold = 2

2. 优化文档清洗

在 vector_stores.py 中增加：

clean_text()

3. 增强 metadata

增加：

file_type
upload_time
chunk_index
chunk_id

4. 优化 context 格式

在 rag.py 中优化 format_document()。

5. 优化 Prompt

让模型严格基于参考资料回答。

---
第二阶段：检索调试与可视化

目标：让开发者能看到检索过程。

建议修改：

1. 增加检索分数返回

在 vector_stores.py 中增加：

similarity_search_with_scores()

2. 页面增加调试开关

在 app_qa.py 中增加：

显示检索调试信息

3. 页面展示检索片段

展示：

- chunk 内容；
- 来源文件；
- metadata；
- score。

---
第三阶段：高级检索优化

目标：提升复杂问题和大知识库场景下的效果。

建议增加：

1. Query Rewrite；
2. 混合检索；
3. Rerank；
4. metadata filter；
5. 多知识库分类；
6. 文件去重与更新机制。

---
十七、推荐代码改造方案

下面给出一套适合当前项目的代码改造示例。

---
17.1 config_data.py 优化

建议改成：

import os
from dotenv import load_dotenv

load_dotenv()

collections_name = "rag"
persist_directory = "./chroma_db"

chunk_size = 800
chunk_overlap = 100
separators = ["\n\n", "\n", "。", "！", "？", "；", "，", ",", "."]

retrieval_top_k = 5
rerank_top_k = 3
similarity_score_threshold = None

embedding_model_name = "text-embedding-v4"
chat_model_name = "qwen3-max"

DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "")

主要变化：

- similarity_threshold 改为 retrieval_top_k；
- chunk_size 从 1000 调整为 800；
- 增加 rerank_top_k；
- 增加 similarity_score_threshold。

---
17.2 vector_stores.py 优化

增加文本清洗

import re

def clean_text(self, text: str) -> str:
    if not text:
        return ""

    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()

增加文件 MD5

import hashlib

def get_file_md5(self, file_path):
    md5 = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            md5.update(chunk)
    return md5.hexdigest()

优化 add_documents

def add_documents(self, documents):
    if not documents:
        return 0

    cleaned_docs = []
    for doc in documents:
        cleaned_text = self.clean_text(doc.page_content)
        if cleaned_text:
            doc.page_content = cleaned_text
            cleaned_docs.append(doc)

    split_docs = self.text_splitter.split_documents(cleaned_docs)

    for index, doc in enumerate(split_docs):
        doc.metadata["chunk_index"] = index
        doc.metadata["chunk_id"] = f"{doc.metadata.get('file_name', 'unknown')}_{index}"

    self.vector_store.add_documents(split_docs)
    return len(split_docs)

优化 get_retriever

def get_retriever(self, top_k=None, filters=None):
    search_kwargs = {
        "k": top_k or config.retrieval_top_k
    }

    if filters:
        search_kwargs["filter"] = filters

    return self.vector_store.as_retriever(search_kwargs=search_kwargs)

增加带分数检索

def similarity_search_with_scores(self, query, top_k=None):
    return self.vector_store.similarity_search_with_score(
        query,
        k=top_k or config.retrieval_top_k
    )

---
17.3 rag.py 优化

优化 format_document

def format_document(docs: list[Document]):
    if not docs:
        return "未检索到相关参考资料。"

    formatted = []
    for i, doc in enumerate(docs, start=1):
        source = doc.metadata.get("file_name", doc.metadata.get("source", "未知来源"))
        chunk_index = doc.metadata.get("chunk_index", "未知")

        formatted.append(
            f"【资料{i}】\n"
            f"来源：{source}\n"
            f"片段编号：{chunk_index}\n"
            f"内容：{doc.page_content}"
        )

    return "\n\n".join(formatted)

优化 Prompt

@property
def prompt_template(self):
    return ChatPromptTemplate.from_messages(
        [
            (
                "system",
                """
你是一个专业的 RAG 智能客服助手。请严格基于【参考资料】回答用户问题。

回答要求：
1. 优先依据参考资料回答。
2. 如果参考资料不足，请明确说明“知识库中没有找到足够信息”，不要编造。
3. 如果问题涉及推荐，请说明推荐依据。
4. 回答要简洁、专业、结构清晰。
5. 如果可以，请在末尾列出参考来源。

【参考资料】
{context}
"""
            ),
            ("system", "以下是历史对话："),
            MessagesPlaceholder("history"),
            ("user", "{input}")
        ]
    )

---
17.4 app_qa.py 优化

增加 Top K 配置

在侧边栏增加：

top_k = st.slider(
    "检索返回片段数 Top K",
    min_value=1,
    max_value=10,
    value=5
)
st.session_state["top_k"] = top_k

增加调试模式

debug_mode = st.toggle(
    "🔍 显示检索调试信息",
    value=False
)
st.session_state["debug_mode"] = debug_mode

展示检索结果

可以在用户提问后增加：

if st.session_state.get("debug_mode"):
    debug_results = rag.vector_service.similarity_search_with_scores(
        prompt,
        top_k=st.session_state.get("top_k", 5)
    )

    with st.expander("🔍 检索调试信息"):
        for i, (doc, score) in enumerate(debug_results, start=1):
            st.markdown(f"### 检索片段 {i}")
            st.write(doc.page_content)
            st.write(f"Score: {score}")
            st.json(doc.metadata)

这样可以帮助开发者直接看到检索效果。

---
十八、优化后的整体 RAG 流程

经过上述优化后，项目流程可以升级为：

用户上传文件
  ↓
文件解析
  ↓
文本清洗
  ↓
按文档类型切片
  ↓
补充 metadata
  ↓
计算 MD5 去重
  ↓
Embedding 向量化
  ↓
写入 Chroma
  ↓
用户提问
  ↓
结合历史对话进行 Query Rewrite
  ↓
向量检索 Top K
  ↓
可选：关键词检索 BM25
  ↓
合并召回结果
  ↓
可选：Rerank 重排序
  ↓
格式化上下文
  ↓
强约束 Prompt
  ↓
大模型生成答案
  ↓
展示答案 + 参考来源 + 检索调试信息

相比当前版本，优化后的系统具备：

- 更干净的知识库；
- 更合理的切片；
- 更丰富的 metadata；
- 更灵活的 Top K；
- 更强的检索可调试性；
- 更稳的 Prompt 约束；
- 更适合多轮对话；
- 更方便后续扩展混合检索和 Rerank。

---
十九、针对当前项目的优先级建议

如果按照投入产出比排序，建议优先做以下优化。

---
优先级一：马上值得做

1. 修改变量命名

将：

similarity_threshold

改为：

retrieval_top_k

原因：当前命名容易误导。

---
2. 优化 Prompt

让模型严格基于资料回答。

原因：改动小，收益大，可以减少幻觉。

---
3. 优化 context 格式

将检索结果格式化为：

【资料1】
来源：
内容：

原因：模型更容易理解，也方便输出引用。

---
4. 增加检索调试信息

在页面显示检索到的 chunk。

原因：RAG 项目调试必须知道“找到了什么”。

---
优先级二：较推荐做

1. 文本清洗

减少脏数据进入向量库。

2. metadata 增强

为后续来源引用、文件删除、分类检索做准备。

3. Top K 可配置

让用户或开发者能在页面调整检索数量。

4. 文件去重

避免重复上传影响检索效果。

---
优先级三：高级优化

1. Query Rewrite

适合多轮对话场景。

2. Hybrid Search

适合文档多、关键词强、规则多的知识库。

3. Rerank

适合知识库变大后提升最终排序质量。

4. 多知识库分类

适合业务模块增多后使用。

---
二十、总结

RAG 系统的效果并不只取决于大模型本身，更取决于检索链路是否能够稳定、准确地找到正确资料。

RAG 系统的效果并不只取决于大模型本身，更取决于检索链路是否能够稳定、准确地找到正确资料。

当前这个 RAG 智能客服项目已经实现了一个完整的基础闭环：

文件上传 -> 文档切片 -> 向量入库 -> 用户提问 -> 检索增强 -> 大模型回答

但从检索优化角度来看，还可以继续从以下方面提升：

1. 文档预处理：清洗无效文本，提升入库质量；
2. 文本切片：根据文档类型采用更合理的 chunk 策略；
3. 元数据管理：增加来源、类型、时间、chunk 编号等信息；
4. Top K 优化：将固定检索数量改为可配置；
5. 相似度阈值：避免低相关内容进入 Prompt；
6. 混合检索：结合向量检索和关键词检索；
7. Rerank 重排序：从召回结果中选出最相关内容；
8. Query Rewrite：解决多轮对话中的追问检索问题；
9. Prompt 优化：约束模型严格基于资料回答；
10. 检索调试：在页面展示检索片段、来源和分数。

如果要结合当前项目实际情况，最推荐先做四个改动：

1. similarity_threshold 改名为 retrieval_top_k
2. 优化 format_document 的上下文格式
3. 强化 prompt_template，要求模型不要编造
4. 在 Streamlit 页面增加检索调试信息展示

这四个优化改动相对较小，但能明显提高系统的可解释性、稳定性和回答质量。

后续如果知识库规模变大，再逐步加入：

Query Rewrite
Hybrid Search
Rerank
Metadata Filter
文件去重与更新机制