RAG检索增强生成原理：从向量搜索到答案合成的完整技术解析

一、背景：为什么RAG是LLM落地的核心基础设施

1. 知识过时：训练数据有截止日期，无法获取最新信息
2. 幻觉问题：模型会编造听起来合理但完全错误的内容
3. 私有知识缺失：企业的内部文档、规章制度、客服记录等都不在模型知识范围内
4. 推理不可控：无法限定模型只能基于特定资料回答

RAG（Retrieval-Augmented Generation，检索增强生成）正是为解决这些问题而生的。它的核心思想是：不依赖模型自身知识，而是让模型"先检索再回答"——先把私有知识存起来，问答时先去检索相关知识，再让模型基于检索到的内容生成答案。

本文从技术原理层面剖析RAG的每个环节：Embedding、向量化、向量索引、相似度检索、答案生成，手写代码带你理解每个细节。

二、方案：RAG完整技术原理与代码实现

2.1 整体架构

RAG系统的数据流如下：

知识文档
    │
    ▼
┌─────────────────────┐
│  文档解析与切分      │  ← Text Splitter（文本切分器）
│  DocumentLoader     │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  向量化嵌入          │  ← Embedding Model（嵌入模型）
│  Text → Vector      │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  向量数据库存储      │  ← Vector DB（ChromaDB/Qdrant/Milvus）
│  (id, vector, meta) │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  用户查询            │
│  Query               │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  查询向量化          │  ← 同上Embedding Model
│  Query → Vector     │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  相似度检索          │  ← Vector Search（余弦相似度/内积/欧氏距离）
│  Top-K Retrieval    │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  上下文构建          │  ← Prompt Template（提示词模板）
│  Context Builder    │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  LLM答案生成         │  ← LLM（GPT-4/Claude/本地模型）
│  Answer Generation  │
└─────────┴───────────┘
          │
          ▼
      用户答案

2.2 文档解析与切分

知识文档通常以PDF、Word、HTML等格式存在。首先需要把文档解析成纯文本，然后按语义切分成小块。

import re
from pathlib import Path

class TextSplitter:
    """
    文本切分器：将长文本按语义切分成适合检索的小块
    
    切分策略有两种：
    1. 固定长度切分：简单但破坏语义完整性
    2. 语义切分：按句子/段落切分，保留语义完整性（推荐）
    """
    
    def __init__(self, chunk_size=500, chunk_overlap=50):
        """
        参数:
            chunk_size: 每个文本块的目标字符数
            chunk_overlap: 相邻块之间的重叠字符数（避免信息被切割开）
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_by_sentence(self, text: str) -> list[dict]:
        """
        按句子切分（推荐方法）
        
        原理：
        1. 先用正则把文本分成句子（Sentence Tokenization）
        2. 把句子组织成段落
        3. 按段落累积，当达到chunk_size时切一刀
        4. 相邻块之间保留chunk_overlap的重复内容，保证上下文连贯
        """
        # 句子分割正则：匹配常见句子结束符
        sentence_endings = r'[.!?。！？]+'
        sentences = re.split(sentence_endings, text)
        
        chunks = []
        current_chunk = ""
        current_size = 0
        
        for sentence in sentences:
            sentence = sentence.strip()
            if not sentence:
                continue
                
            sentence_size = len(sentence)
            
            # 如果当前块加上这个句子超过chunk_size，先保存当前块，再开新块
            if current_size + sentence_size > self.chunk_size and current_chunk:
                # 保存当前块（带ID和元数据）
                chunks.append({
                    'text': current_chunk.strip(),
                    'length': current_size,
                    'start_idx': len(text) - len(current_chunk) - len(sentence),
                })
                
                # 开新块：保留chunk_overlap的重复内容（衔接上下文）
                overlap_text = current_chunk[-self.chunk_overlap:] if len(current_chunk) > self.chunk_overlap else current_chunk
                current_chunk = overlap_text + " " + sentence
                current_size = len(current_chunk)
            else:
                current_chunk += " " + sentence
                current_size += sentence_size + 1  # 加1是空格
        
        # 处理最后一块
        if current_chunk.strip():
            chunks.append({
                'text': current_chunk.strip(),
                'length': current_size,
                'start_idx': 0,
            })
        
        return chunks

# 使用示例
splitter = TextSplitter(chunk_size=500, chunk_overlap=50)
sample_text = """
人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，致力于开发能够完成通常需要人类智能才能完成的任务的系统。
机器学习是人工智能的一个子集，它使用统计技术使计算机系统能够"学习"而无需明确编程。
深度学习是机器学习的子集，使用多层神经网络来分析各种形式的原始数据。
RAG（检索增强生成）结合了检索系统和语言模型的优势，用于知识密集型任务。
向量数据库是RAG系统中存储和检索嵌入向量的关键技术基础设施。
"""

chunks = splitter.split_by_sentence(sample_text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i+1} (长度:{chunk['length']}) ---")
    print(chunk['text'][:200] + "..." if len(chunk['text']) > 200 else chunk['text'])
    print()

2.3 Embedding向量嵌入原理

Embedding是RAG最核心的技术。它的作用是把文字转换成数学向量，让语义相似的文字在向量空间中距离相近。

2.3.1 词向量基础：从One-Hot到Transformer

最早的词向量是One-Hot编码：每个词表示为一个维度等于词表大小的向量，只有对应词的位置是1，其他都是0。这种表示丢失了语义信息——"猫"和"狗"的One-Hot向量在欧式空间中的距离和"猫"和"汽车"是一样的。

后来有了Word2Vec、GloVe等分布式表示，让语义相似的词在向量空间中靠近。但这些方法无法处理一词多义和上下文依赖。

现代RAG系统使用的是Transformer-based Embedding模型，典型代表是OpenAI的text-embedding-3-small、Google的Gemini Embedding、国产的Embedding模型如BGE等。

2.3.2 Transformer Embedding的数学原理

import numpy as np

def self_attention(Q, K, V, dk=64):
    """
    简化版自注意力机制实现
    
    原理：
    1. QK^T 计算Query和Key的相似度（点积）
    2. 除以√d_k 进行缩放，防止梯度消失
    3. softmax归一化得到注意力权重
    4. 权重乘以V得到加权和
    """
    # 步骤1: 计算点积相似度
    scores = np.dot(Q, K.T)  # (seq_len, seq_len)
    
    # 步骤2: 缩放（防止点积过大导致softmax梯度为0）
    scaled_scores = scores / np.sqrt(dk)
    
    # 步骤3: softmax归一化（得到注意力权重）
    exp_scores = np.exp(scaled_scores - np.max(scaled_scores, axis=-1, keepdims=True))
    attention_weights = exp_scores / np.sum(exp_scores, axis=-1, keepdims=True)
    
    # 步骤4: 加权求和
    output = np.dot(attention_weights, V)
    
    return output, attention_weights

# 示例：计算"猫爱玩耍"和"狗喜欢运动"的注意力
# 实际Transformer中，这些向量是通过学习得到的，这里用简化示例
seq_len = 4  # 假设序列长度为4
d_model = 8  # 向量维度

# 模拟两个句子的词向量
sentence1 = np.random.randn(seq_len, d_model)  # "猫爱玩耍"
sentence2 = np.random.randn(seq_len, d_model)  # "狗喜欢运动"

# 计算注意力
output1, attn1 = self_attention(sentence1, sentence1, sentence1)
output2, attn2 = self_attention(sentence2, sentence2, sentence2)

print("句子1注意力权重形状:", attn1.shape)
print("句子2注意力权重形状:", attn2.shape)

2.3.3 余弦相似度检索

两个向量在语义上是否相似，用余弦相似度来衡量：

import numpy as np

def cosine_similarity(vec_a: np.ndarray, vec_b: np.ndarray) -> float:
    """
    计算两个向量的余弦相似度
    
    公式：cos(θ) = (A·B) / (||A|| × ||B||)
    """
    # 点积
    dot_product = np.dot(vec_a, vec_b)
    
    # L2范数（向量的模长）
    norm_a = np.linalg.norm(vec_a)
    norm_b = np.linalg.norm(vec_b)
    
    # 避免除零
    if norm_a == 0 or norm_b == 0:
        return 0.0
    
    return dot_product / (norm_a * norm_b)

def search_by_similarity(query_vector: np.ndarray, 
                          document_vectors: dict[int, np.ndarray], 
                          top_k: int = 5) -> list[tuple[int, float]]:
    """
    向量检索：给定查询向量，从文档向量库中找出最相似的top_k个
    
    参数:
        query_vector: 查询的嵌入向量 (embedding_dim,)
        document_vectors: 文档ID到向量的映射 {doc_id: vector}
        top_k: 返回前k个最相似的结果
    
    返回:
        [(doc_id, similarity_score), ...]，按相似度降序排列
    """
    similarities = []
    
    for doc_id, doc_vector in document_vectors.items():
        sim = cosine_similarity(query_vector, doc_vector)
        similarities.append((doc_id, sim))
    
    # 按相似度降序排序
    similarities.sort(key=lambda x: x[1], reverse=True)
    
    return similarities[:top_k]

# 示例使用
np.random.seed(42)

# 模拟文档向量库：4个文档，嵌入维度768（实际常用）
doc_vectors = {
    1: np.random.randn(768),
    2: np.random.randn(768),
    3: np.random.randn(768),
    4: np.random.randn(768),
}

# 模拟查询向量（假设与文档1语义最接近）
query = doc_vectors[1] + np.random.randn(768) * 0.1  # 加噪声模拟微调

# 检索
results = search_by_similarity(query, doc_vectors, top_k=4)

print("检索结果（doc_id, similarity）：")
for rank, (doc_id, score) in enumerate(results, 1):
    print(f"  Top {rank}: doc_{doc_id} = {score:.4f}")

2.4 完整RAG Pipeline实现

结合上面的原理，下面是一个完整的RAG Pipeline代码：

from typing import Optional
from dataclasses import dataclass
import numpy as np

@dataclass
class Document:
    """文档数据结构"""
    id: str
    content: str
    metadata: dict

@dataclass  
class检索Result:
    """检索结果数据结构"""
    document: Document
    similarity: float
    rank: int

class SimpleRAG:
    """
    简化版RAG Pipeline实现（不含真实Embedding调用）
    
    实际项目中推荐使用：
    - Embedding: OpenAI text-embedding-3-small / BGE-large-zh
    - Vector DB: ChromaDB / Qdrant / Milvus
    - LLM: GPT-4 / Claude-3 / 本地Ollama
    """
    
    def __init__(self, embedding_dim: int = 768):
        self.embedding_dim = embedding_dim
        self.documents: dict[str, Document] = {}  # doc_id -> Document
        self.vectors: dict[str, np.ndarray] = {}  # doc_id -> vector
        
        # 提示词模板（实际项目中应该提取到配置）
        self.prompt_template = """你是一个智能助手。请根据以下参考材料回答用户问题。

参考材料：
{context}

用户问题：{question}

回答要求：
1. 只基于参考材料回答，不要编造信息
2. 如果参考材料中没有相关信息，请如实说"我不知道"
3. 回答要具体、有条理
"""
    
    def add_documents(self, docs: list[Document], embeddings: list[np.ndarray]):
        """
        添加文档到知识库
        
        实际调用时：
        1. 文档先切分
        2. 调用Embedding API获取每个块的向量
        3. 存入向量数据库
        """
        for doc, emb in zip(docs, embeddings):
            self.documents[doc.id] = doc
            self.vectors[doc.id] = emb
            
        print(f"✅ 已添加 {len(docs)} 个文档到知识库")
    
    def retrieve(self, query: str, query_embedding: np.ndarray, 
                 top_k: int = 3, similarity_threshold: float = 0.7) -> list[检索Result]:
        """
        检索相关文档
        
        参数:
            query: 查询文本
            query_embedding: 查询的嵌入向量
            top_k: 返回前k个结果
            similarity_threshold: 相似度阈值，低于此值的结果被过滤
        
        流程：
        1. 计算查询向量与所有文档向量的相似度
        2. 按相似度降序排列
        3. 取top_k，同时过滤低于阈值的结果
        """
        results = []
        
        for doc_id, doc_vector in self.vectors.items():
            sim = cosine_similarity(query_embedding, doc_vector)
            
            if sim >= similarity_threshold:
                results.append(检索Result(
                    document=self.documents[doc_id],
                    similarity=sim,
                    rank=0
                ))
        
        # 按相似度排序
        results.sort(key=lambda x: x.similarity, reverse=True)
        
        # 添加排名
        for i, result in enumerate(results[:top_k], 1):
            result.rank = i
            
        return results[:top_k]
    
    def generate_context(self, query: str, retrieved_results: list[检索Result]) -> str:
        """
        构建提示词上下文
        
        将检索到的文档内容拼接成LLM可读的格式
        """
        if not retrieved_results:
            return "没有找到相关的参考材料。"
        
        context_parts = []
        for result in retrieved_results:
            source = result.document.metadata.get('source', '未知来源')
            context_parts.append(
                f"[来源：{source}，相似度：{result.similarity:.2f}]\n"
                f"{result.document.content}"
            )
        
        context = "\n\n---\n\n".join(context_parts)
        return context
    
    def build_prompt(self, question: str, context: str) -> str:
        """
        构建完整的提示词
        """
        return self.prompt_template.format(
            question=question,
            context=context
        )
    
    def answer(self, question: str, query_embedding: np.ndarray, 
               top_k: int = 3) -> dict:
        """
        完整的RAG问答流程
        
        流程：
        1. retrieve：检索相关文档
        2. generate_context：构建上下文
        3. build_prompt：构建提示词
        4. LLM生成答案（这里用占位符，实际要调用LLM API）
        """
        # Step 1: 检索
        retrieved = self.retrieve(question, query_embedding, top_k=top_k)
        
        # Step 2: 构建上下文
        context = self.generate_context(question, retrieved)
        
        # Step 3: 构建提示词
        prompt = self.build_prompt(question, context)
        
        # Step 4: LLM生成（这里打印，实际项目要调用真实API）
        # answer = llm_api.call(prompt)
        answer = f"[模拟答案] 根据检索到的{len(retrieved)}个参考资料回答：..."
        
        return {
            'answer': answer,
            'prompt': prompt,
            'retrieved_docs': [
                {
                    'id': r.document.id,
                    'content_preview': r.document.content[:100] + '...',
                    'similarity': r.similarity,
                    'rank': r.rank
                }
                for r in retrieved
            ],
            'context': context
        }

# 使用示例（使用随机向量模拟Embedding）
rag = SimpleRAG(embedding_dim=768)

# 添加文档
docs = [
    Document(id="doc1", content="Python是一种高级编程语言，由Guido van Rossum于1989年创建。",
             metadata={"source": "Python官方文档"}),
    Document(id="doc2", content="JavaScript是一种脚本语言，主要用于Web前端开发，现在也可用于服务端(Node.js)。",
             metadata={"source": "MDN文档"}),
    Document(id="doc3", content="Go语言由Google开发，于2009年发布，主要用于服务端编程和云计算。",
             metadata={"source": "Go官方文档"}),
]

# 模拟Embedding（实际应调用API）
embeddings = [np.random.randn(768) for _ in docs]
rag.add_documents(docs, embeddings)

# 模拟查询（与doc1最相关）
query_emb = np.random.randn(768) * 0.1 + embeddings[0]  # 加小噪声

# 执行RAG问答
result = rag.answer("Python是谁创建的？", query_emb, top_k=2)
print("问题：Python是谁创建的？")
print(f"答案：{result['answer']}")
print(f"检索到的文档数：{len(result['retrieved_docs'])}")
for doc in result['retrieved_docs']:
    print(f"  - doc_{doc['id']}: 相似度={doc['similarity']:.4f}")

2.5 主流Embedding模型对比

2.6 主流向量数据库对比

三、效果：RAG系统效果评估指标

3.1 检索质量评估

3.2 端到端效果对比

四、总结：RAG实战常见问题与优化方向