RAG系统构建 - 检索增强生成

前置知识：需要先掌握大模型API应用

本文重点：理解RAG原理，掌握向量数据库与检索系统构建

一、RAG概述

1.1 什么是RAG

RAG (Retrieval-Augmented Generation) 是一种结合外部知识检索与大模型生成的技术。

RAG工作流程：
1. 文档处理 → 分块 → 向量化 → 存入向量库
2. 用户查询 → 向量化 → 相似度检索 → 获取相关文档
3. 相关文档 + 查询 → 大模型 → 生成回答
优势：
- 解决知识时效性问题
- 减少幻觉
- 可解释性强
- 支持私有数据

1.2 RAG vs 微调

特性	RAG	微调
知识更新	实时更新	需要重新训练
成本	低	高
可解释性	高	低
适用场景	事实性问答	风格/任务定制

二、向量数据库

2.1 文本嵌入

from langchain_openai import OpenAIEmbeddings
from langchain.embeddings import HuggingFaceEmbeddings
import numpy as np
# OpenAI Embeddings
embeddings_openai = OpenAIEmbeddings()
# 开源 Embeddings (更经济)
embeddings_local = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
# 计算嵌入
text = "这是一段测试文本"
vector = embeddings_local.embed_query(text)
print(f"向量维度: {len(vector)}")
print(f"向量示例: {vector[:5]}")
# 批量嵌入
texts = ["文本1", "文本2", "文本3"]
vectors = embeddings_local.embed_documents(texts)
print(f"批量向量形状: {len(vectors)}x{len(vectors[0])}")

2.2 相似度计算

def cosine_similarity(a, b):
    """余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def similarity_search(query_embedding, doc_embeddings, top_k=3):
    """相似度搜索"""
    similarities = [
        (i, cosine_similarity(query_embedding, doc_emb))
        for i, doc_emb in enumerate(doc_embeddings)
    ]
    similarities.sort(key=lambda x: x[1], reverse=True)
    return similarities[:top_k]
# 示例
docs = [
    "机器学习是人工智能的分支",
    "深度学习使用神经网络",
    "今天天气很好",
    "Python是一种编程语言"
]
doc_embeddings = embeddings_local.embed_documents(docs)
query = "什么是AI"
query_embedding = embeddings_local.embed_query(query)
results = similarity_search(query_embedding, doc_embeddings)
print(f"查询: {query}")
print(f"最相关文档:")
for idx, score in results:
    print(f"  [{score:.3f}] {docs[idx]}")

2.3 向量数据库选型

"""
主流向量数据库对比：
1. FAISS (Meta开源)
   - 本地运行，无需服务器
   - 适合中小规模数据
   - 支持GPU加速
2. Chroma
   - 轻量级，嵌入式
   - LangChain默认支持
   - 适合原型开发
3. Pinecone
   - 云托管服务
   - 自动扩展
   - 生产环境推荐
4. Milvus
   - 开源分布式
   - 支持大规模数据
   - 企业级应用
5. Weaviate
   - 原生支持多模态
   - GraphQL API
   - 语义搜索增强
"""
# 使用Chroma示例
from langchain_community.vectorstores import Chroma
texts = ["文档1内容", "文档2内容", "文档3内容"]
vectorstore = Chroma.from_texts(
    texts=texts,
    embedding=embeddings_local,
    persist_directory="./chroma_db"
)
# 检索
results = vectorstore.similarity_search("查询内容", k=2)
for doc in results:
    print(doc.page_content)

三、文档处理Pipeline

3.1 文档加载

from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    WebBaseLoader
)
# PDF加载
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()
# Markdown加载
md_loader = UnstructuredMarkdownLoader("document.md")
md_docs = md_loader.load()
# 网页加载
web_loader = WebBaseLoader("https://example.com/article")
web_docs = web_loader.load()
print(f"PDF页数: {len(pdf_docs)}")
print(f"第一页内容预览: {pdf_docs[0].page_content[:200]}")

3.2 文本分割

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)
# 递归分割（推荐）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 块大小
    chunk_overlap=200,      # 重叠部分
    length_function=len,
    separators=["\n\n", "\n", "。", ".", " ", ""]
)
chunks = text_splitter.split_text(pdf_docs[0].page_content)
print(f"分割后块数: {len(chunks)}")
# 按Token分割
token_splitter = TokenTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
token_chunks = token_splitter.split_text(pdf_docs[0].page_content)
print(f"Token分割块数: {len(token_chunks)}")

3.3 元数据管理

from langchain.schema import Document
# 创建带元数据的文档
docs = [
    Document(
        page_content="这是文档内容...",
        metadata={
            "source": "document.pdf",
            "page": 1,
            "author": "张三",
            "date": "2024-01-01"
        }
    )
]
# 分割时保留元数据
split_docs = text_splitter.split_documents(docs)
for doc in split_docs:
    print(f"内容: {doc.page_content[:50]}...")
    print(f"元数据: {doc.metadata}")

四、完整RAG系统

4.1 构建索引

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema import StrOutputParser
# 准备文档
documents = [
    "机器学习是人工智能的一个分支，它使计算机能够从数据中学习。",
    "深度学习是机器学习的子领域，使用多层神经网络。",
    "自然语言处理(NLP)是AI的重要应用领域。",
    "Transformer架构是现代NLP的基础。",
    "RAG结合检索和生成，提高回答准确性。"
]
# 创建向量存储
vectorstore = Chroma.from_texts(
    texts=documents,
    embedding=OpenAIEmbeddings(),
    persist_directory="./rag_db"
)
# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)
print(f"已索引 {len(documents)} 个文档")

4.2 RAG问答链

# 定义Prompt模板
template = """
你是一个专业的问答助手。请根据以下上下文回答问题。
上下文：
{context}
问题：{question}
请根据上下文提供准确、详细的回答。如果上下文中没有相关信息，请明确说明。
"""
prompt = ChatPromptTemplate.from_template(template)
# 创建模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 格式化检索结果
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
# 构建RAG链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
# 提问
question = "什么是深度学习？"
answer = rag_chain.invoke(question)
print(f"问题: {question}")
print(f"回答: {answer}")

4.3 高级RAG技术

# ===== 1. 多查询检索 =====
from langchain.retrievers.multi_query import MultiQueryRetriever
# 自动生成多个相关查询，提高召回率
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever,
    llm=llm
)
# ===== 2. 上下文压缩 =====
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)
# ===== 3. 重排序 =====
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# 混合检索：语义+关键词
bm25_retriever = BM25Retriever.from_texts(documents)
bm25_retriever.k = 3
ensemble_retriever = EnsembleRetriever(
    retrievers=[retriever, bm25_retriever],
    weights=[0.5, 0.5]
)
# ===== 4. 带来源的问答 =====
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
question_answering_prompt = ChatPromptTemplate.from_messages([
    ("system", "根据以下上下文回答问题:\n\n{context}"),
    ("user", "{input}")
])
question_answer_chain = create_stuff_documents_chain(llm, question_answering_prompt)
rag_chain = create_retrieval_chain(retriever, question_answer_chain)
response = rag_chain.invoke({"input": "什么是Transformer？"})
print(f"回答: {response['answer']}")
print(f"来源: {[doc.metadata for doc in response['context']]}")

参考资源

LangChain RAG教程 - 官方RAG教程

Pinecone RAG指南 - RAG原理详解

Chroma文档 - 向量数据库文档

FAISS Wiki - FAISS使用指南

Advanced RAG Techniques - 高级RAG技术

LlamaIndex - 另一个RAG框架

RAG评测框架 - RAGAS评测工具

上一篇：大模型API应用 下一篇：AI工具链返回：RAG系统 最后更新: 2026年4月17日

访问 --

搜索文章、标签、项目线索

RAG系统构建 - 检索增强生成

一、RAG概述

1.1 什么是RAG

1.2 RAG vs 微调

二、向量数据库

2.1 文本嵌入

2.2 相似度计算

2.3 向量数据库选型

三、文档处理Pipeline

3.1 文档加载

3.2 文本分割

3.3 元数据管理

四、完整RAG系统

4.1 构建索引

4.2 RAG问答链

4.3 高级RAG技术

参考资源

讨论与反馈