RAG 检索增强生成:原理、局限与优化路径
RAG 是当前落地 LLM 应用的主流架构,但实践中召回不准、上下文稀释、幻觉依然存在。本文从向量检索原理出发,覆盖 Chunk 策略、PGVector、HTTPS 混合搜索与 RRF 融合。
RAG 的工作原理
RAG(Retrieval-Augmented Generation)由 Facebook AI 在 2020 年提出,核心思想是:不让模型自己生成所有知识,而是先把外部知识存起来,推理时检索最相关的片段,作为上下文注入模型。
RAG 能解决"知识不足"问题,但解决不了"推理能力不足"问题。如果模型本身就不擅长多步骤逻辑推理,给它再好的上下文也救不了。RAG 的召回质量乘以模型的阅读理解质量,才是最终答案质量的上限。
RAG Pipeline 的 5 个环节
1. 文档解析
把原始文档(PDF、Word、网页)解析成纯文本。PDF 最麻烦——表格、多栏布局、脚注都能让解析质量大幅下降。
import pdfplumber
import PyMuPDF
import marker
parsers = {
"pdfplumber": {"strength": "表格提取好", "weakness": "多栏布局支持差"},
"PyMuPDF": {"strength": "速度快", "weakness": "表格支持弱"},
"marker": {"strength": "端到端 PDF 转 Markdown", "weakness": "速度慢,需要 GPU"},
}
print("解析器选择建议:")
for tool, info in parsers.items():
print(f" {tool}: {info['strength']}")2. 分块(Chunking)
把长文档切成小块,每块作为一个独立的检索单元。Chunk size 决定召回精度——太大则噪声多,太小则上下文不完整。
def recursive_chunk(text, max_chars=500, overlap=50):
if len(text) <= max_chars:
return [text]
chunks = []
start = 0
while start < len(text):
end = start + max_chars
chunk = text[start:end]
chunks.append(chunk)
start = end - overlap
return chunks
chunks = recursive_chunk(long_document)
print(f"生成了 {len(chunks)} 个 chunk")3. 向量嵌入与存储
用 Embedding 模型把每个 Chunk 转成向量,存入支持向量检索的数据库。推荐用 instruction-based 模型(如 bge-large)而非 base 模型,跨语言召回效果更好。
import numpy as np
from supabase import create_client
def cosine_similarity(a, b):
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def retrieve(query_embedding, chunks, k=5):
scores = [cosine_similarity(query_embedding, chunk['embedding']) for chunk in chunks]
top_k = sorted(zip(scores, chunks), reverse=True)[:k]
return [chunk for _, chunk in top_k]
query_emb = embed_text(user_question)
retrieved = retrieve(query_emb, chunks, k=5)
print(f"召回 {len(retrieved)} 个相关片段")4. 检索结果重排序
Top-K 召回后,用重排序模型(如 bge-reranker)做二次排序,比单纯向量相似度更准。
import numpy as np
from collections import defaultdict
def bm25_score(query, document, k1=1.5, b=0.75):
# BM25 算法
pass
def rrf_scores(hybrid_results, k=60):
# Reciprocal Rank Fusion
fused = defaultdict(float)
for source, ranked_list in hybrid_results.items():
for rank, item in enumerate(ranked_list):
fused[item] += 1.0 / (k + rank + 1)
return sorted(fused.items(), key=lambda x: x[1], reverse=True)
results = rrf_scores({'vector': vector_top10, 'bm25': bm25_top10})
print("融合后排序结果:", results[:5])5. 生成与幻觉控制
把检索到的片段注入 Prompt,让 LLM 基于给定上下文回答问题。关键:Prompt 里要明确说"只能根据提供的文档回答,不要编造"。
def build_rag_prompt(question, retrieved_chunks):
context = "\n\n".join([f"[文档{i+1}]: {chunk}" for i, chunk in enumerate(retrieved_chunks)])
return f"""你是一个专业的跨境电商客服助手。请基于以下文档回答用户问题。
只根据提供的文档内容回答,不要编造。
文档:
{context}
用户问题:{question}
"""
prompt = build_rag_prompt(user_question, retrieved_chunks)
response = llm.invoke(prompt)
print(response)RAG 的核心局限
RAG 不是银弹。它的局限主要来自三个方面:
1. 召回质量决定生成上限:如果向量数据库里没有相关内容,模型再怎么好也答不出来。Embedding模型的选取、Chunk size的大小、Top-K 的数量都会影响召回。
2. 上下文稀释:上下文窗口不是无限的。当检索片段超过 10 个时,有效信息被稀释,模型容易"眉毛胡子一把抓"。
3. 幻觉仍然存在:即使给了正确的上下文,模型仍可能在推理过程中产生幻觉。RAG 能降低幻觉频率,但不能根除。
实践建议:在生产环境里,建议加一个"置信度检测"环节:让模型在回答后输出"信心分数",低于阈值就走人工。
三种向量数据库横向对比
-- PGVector 建表语句
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
id bigserial primary key,
content text not null,
embedding vector(1024) not null,
metadata jsonb
);
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops);