高质量 RAG 问答系统设计学习指南

基于 LangChain + Milvus 的渐进式构建路线 当前项目进度: 基础 RAG + 混合检索(向量+BM25) + RRF融合 + Agent 问答

目录

1. 起步: 最基础的 RAG 问答系统
2. 升级一: 引入 Milvus 向量数据库
3. 升级二: 混合检索 -- 向量 + BM25
4. 升级三: RRF 结果融合
5. 升级四: 查询重写与扩展
6. 升级五: 重排序 Re-ranking
7. 升级六: 完整架构与持续优化

一、最基础的 RAG 问答系统

1.1 RAG 是什么

RAG(Retrieval-Augmented Generation) = 检索 + 生成。

用户问题 → 检索相关文档 → 把文档 + 问题一起喂给 LLM → LLM 生成回答

核心价值: 让 LLM 基于你私有知识库中的真实数据回答,而不是靠训练记忆。

1.2 最小实现

一个最基础的 RAG 只需要三件事:

1. 文档加载与切分 -- 把 PDF/TXT/MD 切成小块(chunk)
2. 向量化存储 -- 用 embedding 模型把文本转为向量,存入向量库
3. 检索 + 生成 -- 用户提问时,用同样的 embedding 模型把问题转为向量,检索最相似的 chunk,交给 LLM 生成回答

文档 → 切分 → 向量化 → 存入向量库
                              ↑
用户问题 → 向量化 → 检索相似chunk → LLM → 回答

1.3 基础代码结构

文档加载与切分 (rag_loadDocuments.py):

from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_documents(directory: str):
    """遍历目录,加载所有支持的文档"""
    all_docs = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            file_path = os.path.join(root, file)
            ext = file.lower().split(".")[-1]
            if ext in ("txt", "md"):
                loader = TextLoader(file_path, encoding="utf-8")
                all_docs.extend(loader.load())
            elif ext == "pdf":
                loader = PyPDFLoader(file_path)
                all_docs.extend(loader.load())
    return all_docs

def split_documents(docs, chunk_size=300, chunk_overlap=30):
    """把文档切成小块"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    return splitter.split_documents(docs)

为什么 chunk_size=300?

• 太小: 上下文断裂,信息不完整
• 太大: 噪声多,LLM 注意力分散,token 消耗大
• 300 是经验起点,根据实际内容类型调整

Agent + 检索 Tool (rag_chain.py 基础版):

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings

# 初始化向量库
embeddings = OpenAIEmbeddings(...)
vector_store = Milvus(
    embedding_function=embeddings,
    connection_args={"uri": "http://192.168.56.10:19530"},
    collection_name="college_rag",
)

@tool
def retrieval_context(query: str):
    """搜索校园知识库"""
    docs = vector_store.similarity_search(query=query, k=5)
    return "\n".join([d.page_content for d in docs])

# 创建 Agent
agent = create_agent(model=llm, tools=[retrieval_context])

这是最初阶段 -- 纯向量语义检索,Agent 自动调用 tool 获取上下文后回答。

1.4 这个阶段的局限

接下来一步步解决这些问题。

二、引入 Milvus 向量数据库

2.1 为什么要用 Milvus

基础阶段可能用 FAISS(内存向量库)或 Chroma(轻量级),但它们有局限:

Milvus 的核心优势:

• 混合检索: 同时支持向量检索 + BM25 关键词检索
• 标量过滤: filter="doc_type == 'pdf' AND year >= 2024"
• 分区管理: 按租户/业务线隔离数据
• 生产级: 支持分布式部署、水平扩展

2.2 创建 Collection 与写入数据

from pymilvus import MilvusClient
from pymilvus.milvus_client.index import IndexParams

client = MilvusClient(uri="http://192.168.56.10:19530", db_name="lfl_college")

# 创建 collection
client.create_collection(
    collection_name="campus_details",
    dimension=1536,  # text-embedding-v2 的输出维度
    auto_id=True,    # 自动生成主键
)

# 创建向量索引
index_params = IndexParams()
index_params.add_index(
    field_name="vector",
    index_type="IVF_FLAT",    # 适合中小数据集
    metric_type="L2",
    params={"nlist": 128},
)
client.create_index(collection_name="campus_details", index_params=index_params)

# 写入数据
data = [
    {"text": "校园开放时间为...", "source": "校规手册.txt", "vector": [0.1, ...]},
]
client.insert(collection_name="campus_details", data=data)
client.flush("campus_details")         # 强制落盘
client.load_collection("campus_details") # 加载到内存

2.3 Schema 设计演进

最初版本(只有文本 + 向量):

fields = [
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1536),
]

加入元数据(支持来源追溯):

fields = [
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=500),
    FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=10),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1536),
]

加入 BM25 稀疏向量(支持混合检索):

fields = [
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=500),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema(name="sparse_vector", dtype=DataType.SPARSE_FLOAT_VECTOR),
]

# BM25 Function: 自动从 text 生成 sparse_vector
bm25_func = Function(
    name="bm25_text_embedding",
    function_type=FunctionType.BM25,
    input_field_names=["text"],
    output_field_names=["sparse_vector"],
)
schema = CollectionSchema(fields, functions=[bm25_func])

关键点: BM25 Function 会在 insert 时自动把 text 分词生成 sparse_vector,不需要手动处理。

三、混合检索 -- 向量 + BM25

3.1 为什么需要混合检索

纯向量检索的短板:

BM25(Best Matching 25)是传统搜索引擎用的关键词匹配算法:

• 对 text 做分词
• 计算每个词在文档中的 TF-IDF 权重
• 按 query 词匹配度打分

两者互补:

• 向量检索: 擅长"语义相近但用词不同" -- "学校怎么样" 匹配 "校园简介"
• BM25: 擅长"精确关键词匹配" -- "qwen3.5" 精确匹配包含该词的文档

3.2 实现方式

方案 A: Milvus 原生 Hybrid Search(推荐,项目中已采用)

from pymilvus import AnnSearchRequest, WeightedRanker

# 1. 向量检索
query_vector = _get_embedding(query)
vector_results = collection.search(
    data=[query_vector],
    anns_field="vector",
    param={"metric_type": "L2", "params": {"nprobe": 16}},
    limit=10,
    output_fields=["text", "source"],
)

# 2. BM25 关键词检索
bm25_req = AnnSearchRequest(
    data=[query],  # 原始文本,Milvus 内部自动分词
    anns_field="sparse_vector",
    param={"metric_type": "BM25"},
    limit=10,
    output_fields=["text", "source"],
)
bm25_results = collection.hybrid_search(
    reqs=[bm25_req],
    rerank=WeightedRanker(0.5, 0.5),  # 两路权重各占一半
    limit=10,
    output_fields=["text", "source"],
)

方案 B: 代码层两路独立检索 + 手动融合

如果 Milvus 版本不支持 BM25 Function,可以退化:

def keyword_search(query: str, all_docs: list, top_k: int = 10):
    """退化的关键词匹配: 统计 query 词在 doc 中的出现"""
    from collections import Counter
    import jieba
    query_words = set(jieba.lcut(query))
    scored = []
    for doc in all_docs:
        doc_words = Counter(jieba.lcut(doc.text))
        score = sum(doc_words.get(w, 0) for w in query_words)
        scored.append((doc, score))
    scored.sort(key=lambda x: x[1], reverse=True)
    return [d for d, _ in scored[:top_k]]

四、RRF 结果融合

4.1 为什么需要融合

两路检索各有排名:

• 向量检索: ["文档A", "文档B", "文档C"]
• BM25检索: ["文档C", "文档A", "文档D"]

直接拼接会丢失排名信息。RRF(Reciprocal Rank Fusion) 是一种无参数、无需归一化的排名融合算法。

4.2 RRF 公式

RRF_score(doc) = Σ 1 / (k + rank(doc在各路结果中的排名))

k 通常取 60,作用是降低排名差异的影响(第1名和第2名的差距不要太大)。

4.3 代码实现

from collections import defaultdict

def _rrf_fusion(vector_results, bm25_results, k=60):
    scores = defaultdict(float)
    texts = {}
    sources = {}

    # 向量检索排名
    for rank, hits in enumerate(vector_results, 1):
        for hit in hits:
            pk = hit.id
            scores[pk] += 1.0 / (k + rank)
            texts[pk] = hit.entity.get("text")
            sources[pk] = hit.entity.get("source", "unknown")

    # BM25 检索排名
    for rank, hits in enumerate(bm25_results, 1):
        for hit in hits:
            pk = hit.id
            scores[pk] += 1.0 / (k + rank)
            texts[pk] = hit.entity.get("text")
            sources[pk] = hit.entity.get("source", "unknown")

    # 按 RRF 分数排序
    sorted_pks = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
    return [{"text": texts[pk], "source": sources[pk], "score": scores[pk]}
            for pk in sorted_pks]

为什么两路都有就用 RRF,而不是 WeightedRanker?

• Milvus 的 WeightedRanker 依赖内部距离分数,不同 metric 类型(L2 vs BM25)的分数范围不一致,直接加权可能失衡
• RRF 只关心 排名,不关心分数绝对值,更加鲁棒

4.4 当前项目的检索流程

用户查询
    ├── 向量检索(L2, nprobe=16, limit=10)
    └── BM25检索(limit=10)
            ↓
        RRF 融合(k=60)
            ↓
        取 top 5,带来源输出
            ↓
        Agent 基于上下文生成回答

五、查询重写与扩展

5.1 问题

用户提问往往简短模糊:

• 输入: "福师大简介"
• 期望: 匹配到 "福建师范大学学校简介"、"福师大基本情况"

5.2 方案

用 LLM 将短查询扩展为 2-3 个变体:

输入: "福师大图书馆怎么用?"

LLM 重写:
1. "福建师范大学图书馆使用方法"
2. "图书馆借阅流程和开放时间"
3. "福师大图书馆怎么借书"

对每个变体分别检索,所有结果去重后取 top。

5.3 实现

def rewrite_query(query: str) -> list[str]:
    """用 LLM 将查询重写为多个变体"""
    prompt = f"""
请将以下查询重写为3个语义相近但表述不同的查询变体。
要求:
1. 第一个是更正式/完整的表述
2. 第二个是同义词替换
3. 第三个是从不同角度提问

原始查询: {query}
只返回3行,每行一个查询,不要序号。
"""
    response = llm.invoke(prompt)
    variants = [line.strip() for line in response.content.strip().split("\n") if line.strip()]
    return [query] + variants[:3]  # 保留原始查询

然后在检索中并行处理:

def expanded_search(query: str, top_k: int = 5):
    variants = rewrite_query(query)
    all_results = {}
    for v in variants:
        results = hybrid_search(v)  # 调用已有的混合检索
        for r in results:
            if r["text"] not in all_results:
                all_results[r["text"]] = r
    return sorted(all_results.values(), key=lambda x: x.get("score", 0), reverse=True)[:top_k]

六、重排序 Re-ranking

6.1 问题

检索 10-20 条后直接取 top 3 给 Agent,但排名靠前不一定等于最相关。低质量 chunk 的噪声会干扰 Agent 生成。

6.2 方案选择

项目中推荐先用 LLM 打分,成本低且效果好:

def rerank(query: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
    """对检索结果做 relevance scoring,返回 top_k 最相关的"""
    scored = []
    for chunk in chunks:
        prompt = f"""判断以下文档片段是否与用户问题相关,打分 0-10:
问题: {query}
文档: {chunk['text'][:500]}
只返回一个数字(0-10)。"""
        resp = llm.invoke(prompt)
        try:
            score = int(resp.content.strip())
        except ValueError:
            score = 5  # 解析失败给中间分
        scored.append({**chunk, "relevance_score": score})

    scored.sort(key=lambda x: x["relevance_score"], reverse=True)
    return scored[:top_k]

成本控制:

• 只在检索结果 > 5 条时触发
• prompt 中文更省 token
• 只取 chunk 前 500 字,减少输入

七、完整架构与持续优化

7.1 最终架构

                    ┌─────────────────────────────────┐
                    │           用户问题               │
                    └────────────┬────────────────────┘
                                 │
                    ┌────────────▼────────────────────┐
                    │     查询重写/扩展                 │
                    │   短查询 → 多版本变体             │
                    └────────────┬────────────────────┘
                                 │
              ┌──────────────────▼──────────────────┐
              │    混合检索 + RRF 融合                │
              │  向量检索 ←──→ BM25 关键词检索        │
              └──────────────────┬──────────────────┘
                                 │
                    ┌────────────▼────────────────────┐
                    │     重排序                        │
                    │   LLM relevance scoring          │
                    └────────────┬────────────────────┘
                                 │
                    ┌────────────▼────────────────────┐
                    │     结构化输出                    │
                    │   带来源标记的上下文              │
                    └────────────┬────────────────────┘
                                 │
                    ┌────────────▼────────────────────┐
                    │     Agent + LLM 生成回答          │
                    └─────────────────────────────────┘

7.2 实施路线图

已完成 ────────────────────────────→ 待做
┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
│ 基础RAG   │→│ Milvus   │→│ 混合检索  │→│ 查询重写  │→│ 重排序   │
│ 向量检索  │  │ 入库     │  │ + RRF    │  │ + 扩展   │  │ + 精排   │
└──────────┘  └──────────┘  └──────────┘  └──────────┘  └──────────┘
                                                         │
                                              ┌──────────▼──────────┐
                                              │  持续优化方向         │
                                              │ • chunk 策略调优      │
                                              │ • 多模态支持          │
                                              │ • 检索缓存            │
                                              │ • A/B 测试评估        │
                                              │ • 用户反馈闭环         │
                                              └─────────────────────┘

7.3 测试验证

准备测试问题集,每个阶段实施后用同一组问题测试:

7.4 每个阶段解决的问题

7.5 持续优化方向

• Chunk 策略调优: 不同文档类型用不同 chunk_size(PDF 可能需要按段落切分)
• 检索缓存: 常见 query 缓存检索结果,减少延迟和 API 调用
• 多路检索: 增加第三路,如基于 metadata 过滤的精确检索
• 评估体系: 用 RAGAS 等框架量化评估检索质量(上下文 precision/recall,回答忠实度)
• 用户反馈: 收集 "这个回答有帮助/无帮助" 信号,持续优化检索策略

本文档随项目迭代持续更新。当前对应代码版本: 混合检索 + RRF 融合 + Agent 问答已实现。