RAG 完全指南 2026：检索增强生成 — 架构、实现与最佳实践

检索增强生成（RAG）是构建基于真实数据的 LLM 应用的主流架构。RAG 不依赖模型训练时的记忆，而是在查询时从知识库检索相关文档并作为上下文传递给 LLM。本指南涵盖完整的 RAG 流水线 — 从文档加载、向量化、向量检索到高级策略、RAGAS 评估和生产部署，包含 LangChain + OpenAI 的 Python 代码示例。

要点概括：RAG = 检索相关文档 + 增强提示词 + 生成有依据的答案。消除幻觉、保持实时性、避免昂贵的微调。流程：加载文档、分块、用 text-embedding-3-small 嵌入、存入向量数据库（Pinecone、Qdrant、pgvector），查询时检索 top-k 结果传给 LLM。高级技术包括 HyDE、父子分块、混合搜索和重排序。用 RAGAS 评估（忠实度、相关性、上下文召回率）。

核心要点

• RAG 将 LLM 回答锚定在真实数据上，与纯提示相比可减少 80-95% 的幻觉。
• 根据内容选择分块大小：Q&A 用 512 tokens，摘要用 1024 tokens，保持 10-20% 重叠。
• text-embedding-3-small 以 $0.02/百万 tokens 的价格提供最优性价比。
• 混合搜索（稠密向量 + 稀疏 BM25）始终优于纯向量相似度搜索。
• 上线前必须用 RAGAS 评估 — 测量忠实度、答案相关性和上下文召回率。
• 初始检索后用交叉编码器重排序可显著提升答案质量，延迟成本极小。

什么是 RAG，为什么重要

大语言模型在静态数据集上训练，有知识截止日期。它们无法访问您的内部文档、最新 API 文档或实时数据。当被问到训练数据之外的信息时，它们要么拒绝回答，要么更糟 — 产生幻觉，给出听起来自信但实际错误的答案。

RAG 通过在生成之前引入检索步骤来解决这个问题。在查询时，系统在知识库中搜索相关文档，将其注入提示作为上下文，并指示 LLM 基于该上下文回答。LLM 成为数据上的推理引擎，而非记忆引擎。

2020 年 Meta 的 Lewis 等人首次提出 RAG，目前已成为企业 AI 应用、客服机器人、文档助手等需要事实性可验证答案的系统的标准架构。

RAG vs 微调 vs 提示工程

在构建 RAG 系统之前，了解何时使用哪种方法：

实际上，大多数生产系统三者结合：提示工程控制格式，RAG 提供知识，微调调整专业行为。但如果你在构建知识密集型应用，从 RAG 开始 — 它以最小投入带来最大收益。

RAG 架构：完整流水线

RAG 系统有两个主要阶段：离线索引流水线处理文档，在线查询流水线处理用户问题。以下是完整架构：

RAG Architecture Overview
========================

INDEXING PIPELINE (Offline)
--------------------------
                                                         
  [Documents]    [Text Splitter]    [Embedding]    [Vector Store]
  PDF, HTML  -->  Chunk into    -->  Convert to -->  Store in
  Markdown       256-512 tokens     dense vectors   Pinecone/
  Database       with overlap       (1536-dim)      Qdrant/PG
                                                         
QUERY PIPELINE (Online)
-----------------------
                                                         
  [User Query]   [Embed Query]   [Retrieve]   [Re-rank]   [Generate]
  "How do I  --> Convert to  --> Top-k     --> Cross-   --> LLM with
   deploy?"     query vector    similar      encoder      context
                                chunks       scoring      + prompt
                   |                            |            |
                   v                            v            v
              Same embedding              Filter &      Grounded
              model as indexing           compress      answer with
                                         context       citations

索引流水线运行一次（或定时运行）处理文档。查询流水线对每个用户问题运行，端到端通常在 1-3 秒内完成。

第一步：文档加载

第一步是将源文档加载为流水线可处理的格式。LangChain 提供了几乎所有格式的加载器：

from langchain_community.document_loaders import (
    PyPDFLoader,
    UnstructuredHTMLLoader,
    UnstructuredMarkdownLoader,
    TextLoader,
    CSVLoader,
    DirectoryLoader,
    WebBaseLoader,
)

# Load a single PDF
pdf_docs = PyPDFLoader("report.pdf").load()

# Load all markdown files from a directory
md_docs = DirectoryLoader(
    "./docs/", glob="**/*.md",
    loader_cls=UnstructuredMarkdownLoader
).load()

# Load a web page
web_docs = WebBaseLoader("https://docs.example.com/api").load()

# Load CSV data
csv_docs = CSVLoader("products.csv").load()

print(f"Loaded {len(pdf_docs)} pages from PDF")
print(f"First doc metadata: {pdf_docs[0].metadata}")

每个加载器返回 Document 对象列表，包含 page_content（文本）和 metadata（来源、页码等）。元数据对引用至关重要 — 务必保留。

第二步：文本分割（分块）

原始文档太大，无法直接嵌入或放入提示。你需要将它们分割成块。分块大小直接影响检索质量 — 这是最需要调优的参数。

常用分割策略：

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
)

# Default: RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = text_splitter.split_documents(pdf_docs)
print(f"Split into {len(chunks)} chunks")

# Markdown-aware splitting
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]
)
md_chunks = md_splitter.split_text(md_content)

分块大小指南：Q&A 系统用 256-512 tokens，50-100 token 重叠。摘要用 1024-2048 tokens。代码按函数/类边界分割。务必测试多种大小并用 RAGAS 衡量检索准确率。

第三步：嵌入模型

嵌入将文本块转换为捕获语义含义的稠密数值向量。语义相似的文本产生在向量空间中接近的向量，从而实现语义搜索。

from langchain_openai import OpenAIEmbeddings

# OpenAI embeddings (recommended for most cases)
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    # dimensions=512  # optional: reduce dimensions for cost savings
)

# Embed a single query
query_vector = embeddings.embed_query("How to deploy a RAG app?")
print(f"Vector dimensions: {len(query_vector)}")  # 1536

# Embed multiple documents (batched automatically)
doc_vectors = embeddings.embed_documents(
    [chunk.page_content for chunk in chunks]
)

建议：原型阶段从 text-embedding-3-small 开始，它以最低成本覆盖大多数场景。只有在评估指标显示需要时才切换到专用模型。

第四步：向量数据库

向量数据库为嵌入建立索引以支持快速相似度搜索。选择取决于规模、基础设施和查询模式。

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Create and persist a vector store
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="my_docs"
)

# Load an existing vector store
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_docs"
)

# Basic similarity search
results = vectorstore.similarity_search("deployment guide", k=4)
for doc in results:
    print(doc.page_content[:100])

推荐：原型阶段用 ChromaDB 或 FAISS。生产环境 Pinecone 运维负担最低。如果已用 PostgreSQL，pgvector 避免引入新服务。混合搜索需求选 Weaviate 或 Qdrant。

第五步：检索策略

检索质量决定 RAG 质量。完美的 LLM 也无法从不相关的上下文生成正确答案。以下是关键检索策略：

对大多数生产系统，最优组合是：混合搜索（稠密+BM25）+ 交叉编码器重排序。在各种基准测试中，这始终优于纯向量相似度搜索。

第六步：基于上下文生成

最后一步将检索到的上下文与精心设计的提示一起传递给 LLM。提示模板至关重要 — 它告诉 LLM 仅使用提供的上下文并引用来源。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_template(
    """Answer the question based ONLY on the following context.
If the context does not contain the answer, say "I don't have 
enough information to answer this question."

Context:
{context}

Question: {question}

Answer (cite sources):"""
)

# Build the RAG chain
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

def format_docs(docs):
    return "\n\n".join(
        f"[Source: {d.metadata.get('source', 'unknown')}]\n"
        + d.page_content for d in docs
    )

rag_chain = (
    {"context": retriever | format_docs,
     "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("How do I deploy to production?")
print(answer)

RAG 提示工程要点：(1) 明确指示模型仅从上下文回答。(2) 包含来源元数据以供引用。(3) 告诉模型在上下文不足时回答"我不知道"。(4) 使用系统消息保持一致行为。

高级 RAG 技术

基础 RAG 流水线运行后，以下技术可显著提升质量：

HyDE（假设文档嵌入）：不直接嵌入用户查询，而是先让 LLM 生成假设答案，然后嵌入该答案进行检索。这弥合了问题式查询和文档式内容之间的差距，在许多基准上提升 10-25% 的召回率。

父子分块：索引小块（256 tokens）以实现精确检索，但返回父块（1024-2048 tokens）作为上下文。在保持检索精确度的同时给 LLM 更多上下文。

多查询检索：用 LLM 生成 3-5 个查询改写，对每个运行检索并合并结果。捕获单一查询表述可能遗漏的相关文档。

上下文压缩：检索后用小型 LLM 从每个块中仅提取相关句子。减少噪音，在上下文窗口中放入更多有用信息。

查询路由：分类查询类型（事实查找、比较、操作指南等），路由到不同的检索策略或知识库。简单的分类器可大幅提升结果。

完整 RAG 实现（Python）

以下是使用 LangChain、OpenAI 和 ChromaDB 的完整 RAG 流水线代码，涵盖文档加载、分块、嵌入、索引和查询：

# Complete RAG pipeline: pip install langchain langchain-openai
# pip install langchain-chroma unstructured pypdf

import os
os.environ["OPENAI_API_KEY"] = "sk-..."

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. Load documents
docs = PyPDFLoader("knowledge_base.pdf").load()

# 2. Split into chunks
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=50
)
chunks = splitter.split_documents(docs)

# 3. Create vector store with embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    chunks, embeddings, persist_directory="./db"
)

# 4. Build RAG chain
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_template(
    "Answer based ONLY on this context:\n{context}\n\n"
    "Question: {question}\nAnswer:"
)

rag_chain = (
    {"context": retriever | (lambda docs: "\n".join(
        d.page_content for d in docs)),
     "question": RunnablePassthrough()}
    | prompt | llm | StrOutputParser()
)

# 5. Query
print(rag_chain.invoke("What is the deployment process?"))

以上代码用不到 50 行创建了完整的 RAG 流水线。生产环境需添加错误处理、流式传输、缓存，并将 ChromaDB 换为托管向量存储。

进阶：混合搜索与重排序

本示例添加 BM25 混合搜索和交叉编码器重排序，显著提升检索质量：

# pip install rank-bm25 sentence-transformers

from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder
import numpy as np

# BM25 sparse retriever
corpus = [doc.page_content for doc in chunks]
tokenized = [doc.split() for doc in corpus]
bm25 = BM25Okapi(tokenized)

def hybrid_search(query, k=10):
    # Dense search via vector store
    dense_results = vectorstore.similarity_search(query, k=k)
    
    # Sparse search via BM25
    bm25_scores = bm25.get_scores(query.split())
    top_bm25_idx = np.argsort(bm25_scores)[-k:][::-1]
    sparse_results = [chunks[i] for i in top_bm25_idx]
    
    # Merge and deduplicate
    seen = set()
    merged = []
    for doc in dense_results + sparse_results:
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            merged.append(doc)
    return merged[:k]

# Cross-encoder re-ranking
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query, docs, top_k=4):
    pairs = [[query, d.page_content] for d in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), reverse=True)
    return [doc for _, doc in ranked[:top_k]]

# Usage: hybrid search + re-rank
candidates = hybrid_search("How to handle auth?", k=10)
final_docs = rerank("How to handle auth?", candidates, top_k=4)

用 RAGAS 评估 RAG 质量

不能衡量就无法改进。RAGAS（检索增强生成评估）是评估 RAG 流水线的标准框架，测量四个关键指标：

# pip install ragas

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Prepare evaluation dataset
eval_data = {
    "question": [
        "How do I deploy to production?",
        "What authentication methods are supported?",
    ],
    "answer": [  # generated by your RAG pipeline
        rag_chain.invoke("How do I deploy to production?"),
        rag_chain.invoke("What auth methods are supported?"),
    ],
    "contexts": [  # retrieved chunks for each question
        [d.page_content for d in retriever.invoke(
            "How do I deploy to production?")],
        [d.page_content for d in retriever.invoke(
            "What auth methods are supported?")],
    ],
    "ground_truth": [  # human-written correct answers
        "Deploy using Docker with the provided Dockerfile...",
        "OAuth 2.0, API keys, and JWT are supported...",
    ],
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy,
             context_precision, context_recall],
)
print(results)
# {faithfulness: 0.92, answer_relevancy: 0.89,
#  context_precision: 0.85, context_recall: 0.78}

在 50-100 个有标准答案的问题测试集上运行 RAGAS。忠实度低于 0.8 说明 RAG 系统在产生幻觉。上下文召回率低于 0.7 说明检索器遗漏了相关文档。

生产环境注意事项

从原型到生产需要解决以下问题：

缓存：缓存相同查询的嵌入结果。缓存相同查询+上下文组合的 LLM 响应。使用 Redis 或简单内存 LRU 缓存。典型工作负载可减少 40-60% 成本。

流式传输：逐 token 流式传输 LLM 响应。用户看到部分响应时会感觉系统更快。LangChain 和 OpenAI 都原生支持流式传输。

成本优化：除非评估需要，否则用 text-embedding-3-small。批量嵌入请求。积极缓存。简单查询用小模型（GPT-4o-mini），复杂查询路由到 GPT-4o。

监控：记录每个查询、检索块和生成答案。跟踪检索延迟、生成延迟和用户反馈。设置忠实度下降告警。工具：LangSmith、Phoenix、W&B。

安全：对文档实施访问控制 — 用户只能检索有权查看的文档。清洗用户查询以防止提示注入。不要在 UI 中未经过滤地暴露原始检索块。

常见陷阱及避免方法

以下是大多数 RAG 实现中常犯的错误：

RAG 框架对比

多个框架简化了 RAG 开发，以下是主要选项对比：

推荐：从 LangChain 开始，凭借其生态广度和文档质量。如果你的场景以文档为主且需要复杂索引，评估 LlamaIndex。TypeScript Web 应用推荐 Vercel AI SDK。

RAG 真实应用场景

RAG 不是理论概念 — 它支撑着数千个生产应用：

成本估算指南

了解 RAG 成本有助于预算规划。以下是典型知识库（10,000 篇文档，平均 2,000 tokens）的成本分解：

关键洞察：嵌入成本可忽略不计。LLM 生成模型选择主导你的账单。80% 查询用 GPT-4o-mini，仅将复杂查询路由到 GPT-4o。这种"模型路由"策略可减少 60-70% 成本。

监控与可观测性

没有监控的 RAG 系统是等待静默失败的系统。你需要对流水线每个阶段的可见性：

LangSmith、Phoenix（Arize）和 Weights & Biases Traces 是 LLM 应用的领先可观测性工具。LangSmith 与 LangChain 无缝集成，开箱即提供追踪可视化、评估和成本跟踪。

生产部署：pgvector + PostgreSQL

如果团队已经使用 PostgreSQL，pgvector 是最简单的生产路径。无需新基础设施、无需新供应商 — 只是现有数据库上的一个扩展：

-- Enable the extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create a table for document chunks
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}',
    embedding vector(1536),  -- matches text-embedding-3-small
    created_at TIMESTAMP DEFAULT NOW()
);

-- Create an HNSW index for fast similarity search
CREATE INDEX ON documents
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- Similarity search query
SELECT id, content, metadata,
    1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE 1 - (embedding <=> $1::vector) > 0.7
ORDER BY embedding <=> $1::vector
LIMIT 5;

# Python: LangChain + pgvector
# pip install langchain-postgres psycopg2-binary

from langchain_postgres import PGVector
from langchain_openai import OpenAIEmbeddings

CONNECTION = "postgresql://user:pass@localhost:5432/mydb"

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = PGVector.from_documents(
    documents=chunks,
    embedding=embeddings,
    connection=CONNECTION,
    collection_name="my_docs",
)

# Query with metadata filtering
results = vectorstore.similarity_search(
    "deployment guide",
    k=4,
    filter={"source": "docs/deploy.md"}
)

pgvector 支持 IVFFlat 和 HNSW 索引进行快速近似最近邻搜索。100 万向量以内无需分片即可良好运行。超过此规模，考虑专用向量存储。

RAG vs 智能体 RAG

传统 RAG 遵循固定流水线：先检索后生成。智能体 RAG 让 LLM 控制检索过程本身 — 它可以决定何时搜索、搜索什么、是否再次搜索以及何时有足够上下文回答。

在智能体 RAG 系统中，LLM 作为推理智能体可以调用检索作为工具。如果首次检索未返回相关结果，智能体会重新构造查询并重试。它还可以组合多次搜索的信息、跨来源验证事实，并在无法回答时优雅退出。

智能体 RAG 对需要多步推理的复杂问题特别强大："比较产品 X、Y 和 Z 的定价"需要三次独立检索，固定流水线无法优雅处理。LangGraph 和 CrewAI 等框架让构建智能体 RAG 系统变得简单。

多模态 RAG：超越文本

现代 RAG 不限于文本。多模态 RAG 可以索引和检索图像、表格和结构化数据：

多模态 RAG 增加了复杂性但显著提升了覆盖率。从纯文本 RAG 开始，然后根据用户问了但找不到答案的内容添加模态。

文档预处理技巧

输入文档的质量直接决定 RAG 质量。垃圾进，垃圾出。以下是预处理最佳实践：

快速入门清单

按此清单从零构建你的第一个 RAG 流水线：

流式 RAG 响应

用户期望实时响应。流式传输在 token 生成时即刻交付，而非等待完整响应。以下是 LangChain 实现流式传输的方法：

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o", temperature=0, streaming=True)

prompt = ChatPromptTemplate.from_template(
    "Answer based ONLY on this context:\n{context}\n\n"
    "Question: {question}\nAnswer:"
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

rag_chain = (
    {"context": retriever | (lambda docs: "\n".join(
        d.page_content for d in docs)),
     "question": RunnablePassthrough()}
    | prompt | llm | StrOutputParser()
)

# Synchronous streaming
for chunk in rag_chain.stream("How do I set up auth?"):
    print(chunk, end="", flush=True)

# Async streaming (for web frameworks)
async def stream_response(question: str):
    async for chunk in rag_chain.astream(question):
        yield chunk  # Send to client via SSE or WebSocket

# FastAPI example
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/ask")
async def ask(q: str):
    return StreamingResponse(
        stream_response(q),
        media_type="text/event-stream"
    )

流式传输将感知延迟从 3-5 秒降低到首 token 500ms 以内。面向用户的应用必须实现流式传输 — 对用户体验的提升是巨大的。

常见问题

RAG 是构建需要有依据、准确、最新答案的 LLM 应用最实用、最有影响力的架构。从简单流水线开始（LangChain + OpenAI + ChromaDB），用 RAGAS 测量，然后迭代。需要更好检索时添加混合搜索和重排序。生态已成熟，工具已就绪 — 关键差异在于你如何针对自己的数据和用例调优分块、检索和提示。