RAG 实战：从手写 MVP 链路到生产级优化

EEva·二月 25, 2026·4 min read

摘要：从零手写 LangChain RAG 链路，解析切片策略、混合检索、重排序（Rerank）及向量数据库实战指南。

01 核心能力要求

在 RAG（检索增强生成）的工程落地中，我们不能只停留在"知道概念"的层面。一个合格的 RAG 工程师必须具备以下能力：

  • 基础链路闭环：能够手写出完整的：文档加载 -> 切片(Chunking) -> 向量化(Embedding) -> 存储(Vector DB) -> 检索 -> 提示词组装 -> 模型生成。
  • 精细化切片策略：不盲目按字数切分，掌握语义切分与 Markdown 标题层级切分。
  • 多路召回与重排序：理解为什么单一向量检索不够用，如何引入 Rerank 解决精度问题。
  • 混合检索 (Hybrid Search)：能够结合 Elasticsearch (BM25) 的关键词搜索与 Vector 的语义搜索。
  • 向量库实战：熟练掌握 Chroma 或 Milvus 的 CRUD 及索引配置。

02 标准 LangChain MVP 实现

切片 (Chunking)

# 策略：按字符递归切分，保留上下文
text_splitter = RecursiveCharacterTextSplitter(
chunk_size = 500,       # 每一块约500字符
chunk_overlap = 50,     # 重叠50字，保证句子不被腰斩
separators = ["

", "
", "。", "!", "?"]  # 优先按段落切
)
splits = text_splitter.split_documents(docs)

向量化 & 存储 (Embedding & Vector DB)

# 调用 OpenAI API 将文字转为向量 [0.1, -0.2, ...]
vectorstore = Chroma.from_documents(
documents = splits,
embedding = OpenAIEmbeddings(),
persist_directory = "./chroma_db"  # 持久化存储
)

检索 (Retrieval)

# 找出最相似的 Top 3 片段
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
question = "TSLA 2025 Q4的净利润率是多少？"
retrieved_docs = retriever.invoke(question)

提示词组装 & 模型生成 (Generation)

llm = ChatOpenAI(model="gpt-3.5-turbo")
chain = prompt | llm

# context 由检索出的 docs 拼接而成
response = chain.invoke({
"question": question,
"context": "

".join([doc.page_content for doc in retrieved_docs])
})

print(response.content)

03 进阶：RAG 效果不好怎么办？

这是面试和实战中最常见的问题。我们通常从以下三点入手优化：

  1. 优化切片策略 (Chunking Strategy)

痛点

死板地按 500 字切分，容易把"2025年营收："切在上一段，而把具体的"100亿"切在下一段。检索时上下文丢失，导致大模型幻觉。

解决方案

  • 语义切分 (Semantic Chunking)：利用 Embedding 计算前后句子的相似度，意思连贯时不切，突变时才切。
  • Markdown 标题切分：按照 # 一、财务摘要、## 1.1 营收 这种层级切分。检索时，内容会带上 财务摘要 > 营收 的元数据，极大地提高检索精度。
  1. 多路召回与重排序 (Rerank)
  • 粗排：向量检索召回 Top 50 相关片段（速度快，精度一般）。
  • 精排：使用 Cross-Encoder（如 bge-reranker）对这 50 条进行精细打分，选出 Top 5 给大模型。

效果：虽然增加了约 200ms 耗时，但准确率会有质的飞跃。

04 向量数据库 (Chroma) 实战速查

  1. 创建/读取集合

import chromadb
client = chromadb.PersistentClient(path="./db")
collection = client.get_or_create_collection(name="finance_reports")

  1. Upsert (更新或插入)

注意：必须指定唯一的 ids，否则数据会重复堆积。

collection.upsert(
documents=["苹果公司Q3营收上涨...", "特斯拉销量下跌..."],
metadatas=[{"source": "report1.pdf"}, {"source": "report2.pdf"}],
ids=["doc1", "doc2"]
)

  1. Query (查询)

results = collection.query(
query_texts=["特斯拉销量怎么样？"],
n_results=2
)

05 深度 Q&A：工程化避坑指南

Q1：PDF 里的表格怎么处理？

直接用 PyPDFLoader 加载表格会变成乱码，语义全毁。

实战解法：使用 pdfplumber 提取表格结构，保留行列关系后再切片。

Q2：如何评估 RAG 效果？

核心指标是"召回率 (Recall)"（即 Top 3 结果里是否包含正确答案）。只有召回率提升（如从 60% -> 80%）时，才上线新代码。

Q3：怎么解决"上下文断裂"问题？

比如"茅台营收"这句话，一半在 Chunk A 结尾，数字在 Chunk B 开头。

实战配置：利用 chunk_overlap（切片重叠）。配置 Chunk Size = 500, Overlap = 50~100。这样 Chunk B 的开头会重复 Chunk A 的结尾，保证关键信息（主语+数字）完整出现在至少一个 Chunk 中。

Q4：流程这么长，怎么优化 Latency？

如果用户等 10 秒，体验就崩了。

三层优化：

  • 体验层：全链路流式输出 (Streaming/SSE)。
  • 架构层：向量检索和 BM25 并行执行；Rerank 后只选 Top 3 给大模型（减少 Input Token）。
  • 兜底层：引入 Redis 语义缓存。如果问题被问过，直接返回缓存答案，耗时仅需 0.1s。
#RAG
首页→