检索增强生成技术入门：从零构建你的第一个RAG应用

1次阅读

共计 2838 个字符，预计需要花费 8 分钟才能阅读完成。

检索增强生成（Retrieval-Augmented Generation，简称 RAG）是近年来自然语言处理领域的重要进展。与传统语言模型（如 GPT 系列）完全依赖预训练知识不同，RAG 在生成答案前会先检索相关文档片段作为参考依据。这种机制带来了三个显著优势：

知识可更新性 ：无需重新训练模型，通过更新检索库即可同步最新知识
结果可验证性 ：每个生成结果都能追溯到具体的参考文档
领域适应性 ：通过领域专用文档库快速适配垂直场景

典型应用场景包括：客服系统（基于产品文档生成回答）、法律咨询（援引法条生成建议）、教育领域（参考教材内容解答问题）等需要准确引用外部知识的场合。

构建 RAG 系统需要两类核心组件：

向量数据库 ：负责高效存储和检索文档向量
FAISS：Facebook 开源的轻量级库，适合本地开发和中小规模数据
Pinecone：全托管云服务，支持自动缩放和实时更新
Weaviate：支持混合检索（向量 + 关键词）的开源方案
语言模型 ：用于生成最终回答
GPT-3.5/4：生成质量高但 API 调用成本较高
Claude 2：长文本处理能力强
Llama 2：可本地部署的开源选择

对于初学者，建议从 FAISS+GPT-3.5 的组合开始，兼顾开发便利性和效果质量。

首先安装必要依赖（假设使用 Python 3.10）：

pip install openai faiss-cpu sentence-transformers pypdf

from typing import List
from sentence_transformers import SentenceTransformer
import PyPDF2

def extract_text_from_pdf(file_path: str) -> List[str]:
    """将 PDF 文档按页提取为文本段落"""
    with open(file_path, 'rb') as f:
        reader = PyPDF2.PdfReader(f)
        return [page.extract_text() for page in reader.pages]

def chunk_text(paragraphs: List[str], chunk_size: int = 500) -> List[str]:
    """将长文本分割为固定大小的片段"""
    chunks = []
    for text in paragraphs:
        words = text.split()
        for i in range(0, len(words), chunk_size):
            chunks.append(' '.join(words[i:i+chunk_size]))
    return chunks

import faiss
import numpy as np

encoder = SentenceTransformer('all-MiniLM-L6-v2')  # 轻量级嵌入模型

def build_vector_index(texts: List[str]) -> faiss.Index:
    """创建 FAISS 向量索引"""
    embeddings = encoder.encode(texts)
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(embeddings)
    return index

# 使用示例
pdf_text = extract_text_from_pdf('manual.pdf')
chunks = chunk_text(pdf_text)
index = build_vector_index(chunks)

import openai

def retrieve_relevant_chunks(query: str, index: faiss.Index, k: int = 3) -> List[str]:
    """检索最相关的 k 个文本片段"""
    query_embedding = encoder.encode([query])
    distances, indices = index.search(query_embedding, k)
    return [chunks[i] for i in indices[0]]

def generate_answer_with_context(question: str, context: List[str]) -> str:
    """调用 OpenAI API 生成回答"""
    prompt = f""" 基于以下信息回答问题：{''.join(context)}
    问题：{question}
    回答："""
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# 完整查询流程
def ask_question(question: str) -> str:
    relevant_chunks = retrieve_relevant_chunks(question, index)
    return generate_answer_with_context(question, relevant_chunks)

评估 RAG 系统时需要关注三个维度：