1. 为什么需要自己动手实现RAG系统
第一次接触RAG（Retrieval-Augmented Generation）这个概念时，我被它的设计理念深深吸引。传统的语言模型在生成内容时，完全依赖预训练阶段学到的知识，这导致两个明显问题：一是无法获取训练数据之外的新知识，二是容易产生"幻觉"（hallucination）——即编造看似合理但实际错误的信息。
RAG通过引入检索机制完美解决了这两个痛点。它的核心思想很简单：当用户提出问题时，先从外部知识库中检索相关文档，然后将这些文档和问题一起输入生成模型，让模型基于最新、最相关的信息生成回答。这种架构既保持了语言模型的强大生成能力，又弥补了它在事实准确性方面的不足。
但市面上的RAG实现方案大多封装得太好，作为开发者很难真正理解其内部运作机制。这就是为什么我决定从零开始手写一个RAG系统——只有亲手实现每个组件，才能真正掌握这项技术的精髓。下面我就把这个实现过程完整分享出来，包括所有关键设计决策和踩过的坑。
2. 系统架构设计
2.1 核心组件拆解
一个完整的RAG系统包含三个主要模块：
- 文档处理流水线：负责将原始文档转换为可检索的向量表示
- 检索系统：根据查询找到最相关的文档片段
- 生成模型：基于检索结果生成最终回答
这三个模块看似简单，但每个都有大量工程细节需要考虑。我们先从最基础的文档处理开始。
2.2 文档处理方案选型
文档处理的核心目标是将非结构化的文本转换为结构化的向量表示。这里有几个关键决策点：
分块策略：
- 固定长度分块（如每256个token一块）
- 基于语义的分块（使用句子边界或段落边界）
- 重叠分块（相邻块有部分重叠内容）
经过测试，我选择了256个token的固定长度分块，重叠部分设为64个token。这种配置在检索准确性和计算效率之间取得了良好平衡。
提示：分块大小需要根据文档类型调整。技术文档适合较小的块（128-256），而连贯性强的文章可能需要更大的块（512+）。
嵌入模型选择：
- OpenAI的text-embedding-ada-002
- 开源的all-MiniLM-L6-v2
- 更大的bge-large-en-v1.5
考虑到本地部署的需求，我最终选择了all-MiniLM-L6-v2。虽然性能略逊于商业API，但768维的嵌入向量已经能提供不错的检索质量，而且完全可以在消费级GPU上运行。
3. 实现细节解析
3.1 构建文档向量库
python复制from sentence_transformers import SentenceTransformer
import numpy as np
import pickle
# 初始化嵌入模型
embedder = SentenceTransformer('all-MiniLM-L6-v2')
# 文档分块示例
def chunk_text(text, chunk_size=256, overlap=64):
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size-overlap):
chunk = ' '.join(words[i:i+chunk_size])
chunks.append(chunk)
return chunks
# 处理文档并保存向量
def process_documents(docs):
all_chunks = []
for doc in docs:
chunks = chunk_text(doc)
all_chunks.extend(chunks)
embeddings = embedder.encode(all_chunks)
# 保存向量和元数据
with open('vector_store.pkl', 'wb') as f:
pickle.dump({
'chunks': all_chunks,
'embeddings': embeddings
}, f)
这个实现有几个值得注意的细节：
- 使用简单的空格分词进行分块，实际项目中可能需要更智能的分词器
- 嵌入向量使用pickle序列化，生产环境建议用专业向量数据库
- 没有处理超长文档，实际应用需要添加文档分割逻辑
3.2 实现检索模块
检索的核心是计算查询向量与文档向量的相似度。我们使用余弦相似度作为度量标准：
python复制from sklearn.metrics.pairwise import cosine_similarity
def retrieve(query, top_k=3):
# 加载向量库
with open('vector_store.pkl', 'rb') as f:
data = pickle.load(f)
# 编码查询
query_embedding = embedder.encode(query)
# 计算相似度
similarities = cosine_similarity(
[query_embedding],
data['embeddings']
)[0]
# 获取top-k结果
top_indices = np.argsort(similarities)[-top_k:][::-1]
return [data['chunks'][i] for i in top_indices]
这里的一个性能优化点是使用FAISS等专用库替代sklearn，特别是当向量库很大时。在我的测试中，对于包含10万条向量的库，FAISS能将检索时间从几百毫秒降低到个位数毫秒。
3.3 集成生成模型
检索到相关文档后，我们需要将它们与原始问题一起喂给语言模型：
python复制from transformers import pipeline
generator = pipeline('text-generation', model='gpt2')
def generate_answer(question, retrieved_docs):
context = "\n\n".join(retrieved_docs)
prompt = f"""基于以下上下文回答问题：
{context}
问题：{question}
答案："""
result = generator(
prompt,
max_length=512,
temperature=0.7,
do_sample=True
)
return result[0]['generated_text']
这个简单实现有几个明显问题：
- 没有处理上下文长度限制（GPT-2只有1024个token的上下文窗口）
- 提示词设计过于简单
- 使用基础GPT-2模型，生成质量有限
在实际项目中，我升级到了Llama 2-7B，并实现了更复杂的提示模板：
code复制你是一个知识渊博的AI助手，请严格根据提供的上下文信息回答问题。
上下文：
{context}
问题：{question}
请按照以下要求回答：
1. 只使用上下文中的信息
2. 如果上下文不包含答案，明确说明"根据现有信息无法回答"
3. 保持回答简洁专业
这种结构化提示显著提高了回答的准确性和可靠性。
4. 性能优化实战
4.1 检索质量提升技巧
原始实现中的检索效果有几个明显瓶颈：
- 词汇不匹配问题：查询中的术语可能与文档使用不同的表达方式
- 语义模糊问题：简单相似度检索可能忽略深层次语义关联
- 相关性排序问题：top-k结果中可能包含冗余信息
我通过以下方法显著提升了检索质量：
查询扩展：
python复制def expand_query(query):
# 使用同义词扩展
synonyms = {
"python": ["Python编程", "Python语言"],
"机器学习": ["ML", "machine learning"]
}
for term, syns in synonyms.items():
if term in query:
query += " " + " ".join(syns)
return query
重排序策略：
python复制def rerank(query, chunks, embeddings):
# 第一轮：基于嵌入相似度
query_embedding = embedder.encode(query)
sim_scores = cosine_similarity([query_embedding], embeddings)[0]
# 第二轮：基于关键词重叠
query_words = set(query.lower().split())
overlap_scores = []
for chunk in chunks:
chunk_words = set(chunk.lower().split())
overlap = len(query_words & chunk_words)
overlap_scores.append(overlap)
# 综合评分
combined = 0.7 * sim_scores + 0.3 * np.array(overlap_scores)
return np.argsort(combined)[::-1]
4.2 生成质量优化
生成环节最常见的两个问题是：
- 忽略检索到的上下文（仍然依赖预训练知识）
- 过度依赖上下文（逐字复制而不理解）
我通过以下方法改善了生成质量：
上下文压缩：
python复制def compress_context(chunks, query):
# 提取每块中最相关的句子
relevant_sentences = []
for chunk in chunks:
sentences = chunk.split('.')
for sent in sentences:
if query.lower() in sent.lower():
relevant_sentences.append(sent)
return " ".join(relevant_sentences[:5]) # 最多5句
生成参数调优：
python复制generation_config = {
"temperature": 0.3, # 降低随机性
"top_p": 0.9,
"repetition_penalty": 1.2,
"max_new_tokens": 256,
"do_sample": True
}
5. 常见问题与解决方案
5.1 检索相关
问题1：检索结果不相关
- 检查嵌入模型是否适合你的领域（技术文档可能需要专门训练的嵌入模型）
- 尝试调整分块大小，太小会丢失上下文，太大可能引入噪声
- 添加查询扩展和重排序步骤
问题2：检索速度慢
- 对于超过1万条记录的库，务必使用FAISS或Annoy等近似最近邻搜索
- 考虑量化嵌入向量（如从float32降到int8）
- 实现缓存机制，对相同查询直接返回缓存结果
5.2 生成相关
问题1：模型忽略检索到的上下文
- 强化提示词设计，明确要求模型只使用提供的上下文
- 在上下文中添加明显标记（如"重要证据："前缀）
- 尝试不同的上下文拼接位置（前/后/前后都有）
问题2：生成内容不连贯
- 调整temperature参数（0.3-0.7通常较好）
- 添加典型的语言模型约束（重复惩罚、长度惩罚等）
- 考虑实现后处理步骤，如语法校正
6. 进阶优化方向
完成基础实现后，我探索了几个进阶优化方向：
混合检索策略：
结合传统的BM25关键词检索和向量检索，取两者之长。BM25对精确术语匹配效果更好，而向量检索擅长捕捉语义相似性。
python复制from rank_bm25 import BM25Okapi
# 初始化BM25
tokenized_corpus = [doc.split() for doc in chunks]
bm25 = BM25Okapi(tokenized_corpus)
# 混合评分
def hybrid_search(query, alpha=0.5):
# 向量检索部分
vector_scores = cosine_similarity(...)
# BM25部分
tokenized_query = query.split()
bm25_scores = bm25.get_scores(tokenized_query)
# 归一化
vector_scores = (vector_scores - np.min(vector_scores)) / (np.max(vector_scores) - np.min(vector_scores))
bm25_scores = (bm25_scores - np.min(bm25_scores)) / (np.max(bm25_scores) - np.min(bm25_scores))
# 混合
combined = alpha * vector_scores + (1-alpha) * bm25_scores
return combined
迭代检索-生成：
第一轮检索生成初步答案，然后基于这个答案发起第二轮检索，最后综合两轮结果生成最终回答。这种方法特别适合复杂问题。
细粒度引用：
让生成模型明确标注回答中的每一部分来自哪个具体文档块，极大增强了可验证性。实现方法是在提示词中要求模型以特殊格式（如[1]）标注引用来源。
经过这些优化，我的RAG系统在技术问答任务上的准确率从最初的58%提升到了82%，已经接近一些商业API的水平。最重要的是，通过这个手写实现过程，我真正理解了RAG系统每个组件的设计考量和实现细节，这种深入理解是单纯调用API永远无法获得的。