LLM RAG架构设计与实现

引言

Retrieval-Augmented Generation（RAG）是目前将 LLM 与外部知识库结合的主流范式。相比微调，RAG 具有成本低、数据可更新、可溯源等优势。然而，构建一个生产级 RAG 系统远不止”检索+生成”这么简单——文档分块策略、Embedding 模型选择、检索算法、重排序、Prompt 工程以及评估体系都是需要精心设计的环节。本文将系统性地介绍 RAG 的完整架构和关键实现细节。

RAG 整体架构

graph TB
    subgraph "离线索引 (Indexing)"
        A[原始文档] --> B[文档解析<br/>PDF/HTML/Markdown]
        B --> C[文档分块<br/>Chunking]
        C --> D[Embedding 生成]
        D --> E[向量数据库<br/>存储索引]
    end

    subgraph "在线检索 (Retrieval)"
        F[用户查询] --> G[Query 重写/扩展]
        G --> H[Query Embedding]
        H --> I[向量相似度检索]
        E --> I
        I --> J[重排序 Re-ranking]
    end

    subgraph "生成 (Generation)"
        J --> K[Prompt 构建<br/>Context + Query]
        K --> L[LLM 生成回答]
        L --> M[引用溯源]
    end

    style A fill:#3498db,color:#fff
    style E fill:#e74c3c,color:#fff
    style L fill:#2ecc71,color:#fff

文档分块策略

分块质量直接影响检索精度和生成质量。不同的分块策略适用于不同场景：

固定大小分块

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Basic fixed-size chunking with overlap
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # Characters per chunk
    chunk_overlap=50,      # Overlap between chunks
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""],
)

chunks = text_splitter.split_text(document_text)

语义分块

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Semantic chunking: split at points where meaning shifts
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # or "standard_deviation"
    breakpoint_threshold_amount=95,
)

semantic_chunks = semantic_splitter.split_text(document_text)

层次化分块（Parent-Child）

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Parent chunks: larger context for generation
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# Child chunks: smaller for precise retrieval
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)

parent_chunks = parent_splitter.split_documents(documents)

for parent in parent_chunks:
    children = child_splitter.split_documents([parent])
    for child in children:
        child.metadata["parent_id"] = parent.metadata.get("id")
        # Store child for retrieval, but use parent for context

graph TD
    A[原始文档] --> B[Parent Chunk<br/>2000 tokens]
    B --> C1[Child 1<br/>400 tokens]
    B --> C2[Child 2<br/>400 tokens]
    B --> C3[Child 3<br/>400 tokens]

    C2 -->|检索命中| D[返回 Parent Chunk<br/>提供完整上下文]

    style C2 fill:#e74c3c,color:#fff
    style D fill:#2ecc71,color:#fff

分块策略对比

策略	优点	缺点	适用场景
固定大小	简单高效	可能切断语义	通用文档
按段落/标题	保持语义完整	长度不均	结构化文档
语义分块	语义边界准确	需要 Embedding 计算	高质量需求
层次化	检索精确+上下文丰富	实现复杂	长文档

Embedding 生成

from sentence_transformers import SentenceTransformer
import numpy as np

# Local embedding model
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# Generate embeddings for chunks
chunks = ["RAG是一种检索增强生成技术", "向量数据库存储文档的Embedding"]
embeddings = model.encode(chunks, normalize_embeddings=True)

# Similarity calculation
similarity = np.dot(embeddings[0], embeddings[1])
print(f"Cosine similarity: {similarity:.4f}")

# Batch processing for large document sets
def batch_embed(texts, batch_size=32):
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        batch_embeddings = model.encode(batch, normalize_embeddings=True)
        all_embeddings.extend(batch_embeddings)
    return np.array(all_embeddings)

Query Embedding 优化

# Instruction-tuned embedding for queries vs documents
# BGE models use different prefixes for queries and documents

query_prefix = "为这个句子生成表示以用于检索相关文章: "
doc_prefix = ""

query_embedding = model.encode(query_prefix + "什么是RAG?")
doc_embedding = model.encode(doc_prefix + "RAG是检索增强生成的缩写...")

# This asymmetric encoding often improves retrieval quality

向量检索

使用 ChromaDB

import chromadb
from chromadb.utils import embedding_functions

# Initialize client
client = chromadb.PersistentClient(path="./chroma_db")

# Create collection with embedding function
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-large-zh-v1.5"
)

collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=embedding_fn,
    metadata={"hnsw:space": "cosine"},
)

# Add documents
collection.add(
    documents=chunks,
    metadatas=[{"source": "doc1.pdf", "page": i} for i in range(len(chunks))],
    ids=[f"chunk_{i}" for i in range(len(chunks))],
)

# Query
results = collection.query(
    query_texts=["什么是向量数据库?"],
    n_results=5,
    where={"source": "doc1.pdf"},  # Metadata filter
)

混合检索（Hybrid Search）

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

# Sparse retriever (keyword-based)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# Dense retriever (vector-based)
vector_retriever = Chroma.from_documents(
    documents, embedding=embeddings
).as_retriever(search_kwargs={"k": 5})

# Ensemble: combine sparse + dense with weights
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7],  # Favor vector search
)

results = ensemble_retriever.invoke("RAG的工作原理是什么?")

graph LR
    A[用户查询] --> B[BM25 关键词检索]
    A --> C[向量相似度检索]

    B --> D[BM25 结果<br/>权重 0.3]
    C --> E[向量结果<br/>权重 0.7]

    D --> F[Reciprocal Rank Fusion<br/>结果合并]
    E --> F

    F --> G[Top-K 候选文档]

    style B fill:#f39c12,color:#000
    style C fill:#3498db,color:#fff
    style F fill:#9b59b6,color:#fff

重排序 (Re-ranking)

from sentence_transformers import CrossEncoder

# Cross-encoder for re-ranking
reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')

query = "RAG架构的核心组件有哪些?"
candidates = [doc.page_content for doc in retrieved_docs]

# Score each (query, candidate) pair
pairs = [[query, candidate] for candidate in candidates]
scores = reranker.predict(pairs)

# Sort by relevance score
ranked_results = sorted(
    zip(retrieved_docs, scores),
    key=lambda x: x[1],
    reverse=True,
)

# Take top-k after re-ranking
top_k_docs = [doc for doc, score in ranked_results[:3]]

Prompt 构建

from langchain.prompts import ChatPromptTemplate

RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的技术问答助手。基于以下提供的上下文信息回答用户的问题。

规则：
1. 只基于提供的上下文回答，不要使用训练数据中的知识
2. 如果上下文中没有相关信息，明确说明"根据提供的资料，我无法回答这个问题"
3. 在回答中引用信息来源，格式为 [来源: 文档名, 页码]
4. 保持回答简洁准确

上下文信息：
{context}"""),
    ("human", "{question}"),
])

def format_context(documents):
    """Format retrieved documents into context string."""
    context_parts = []
    for i, doc in enumerate(documents):
        source = doc.metadata.get('source', 'unknown')
        page = doc.metadata.get('page', 'N/A')
        context_parts.append(
            f"[文档{i+1}] 来源: {source}, 页码: {page}\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(context_parts)

# Build the prompt
context = format_context(top_k_docs)
prompt = RAG_PROMPT.format_messages(context=context, question=query)

完整 RAG Pipeline

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

# Initialize components
llm = ChatOpenAI(model="gpt-4o", temperature=0)
retriever = ensemble_retriever

# Build RAG chain
rag_chain = (
    {
        "context": retriever | format_context,
        "question": RunnablePassthrough(),
    }
    | RAG_PROMPT
    | llm
    | StrOutputParser()
)

# Invoke
answer = rag_chain.invoke("什么是RAG架构?")
print(answer)

Query 重写与扩展

# Multi-query: generate multiple query variations
from langchain.retrievers.multi_query import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vector_retriever,
    llm=llm,
)

# HyDE: Hypothetical Document Embeddings
def hyde_transform(query: str) -> str:
    """Generate a hypothetical answer, then use it for retrieval."""
    prompt = f"请写一段简短的文字来回答这个问题: {query}"
    hypothetical_answer = llm.invoke(prompt).content
    return hypothetical_answer

# Step-back prompting for abstract retrieval
def step_back_query(query: str) -> str:
    prompt = f"这个问题背后更基础的概念是什么: {query}"
    abstract_query = llm.invoke(prompt).content
    return abstract_query

评估指标

graph TD
    A[RAG 评估维度] --> B[检索质量]
    A --> C[生成质量]
    A --> D[端到端质量]

    B --> B1[Recall@K]
    B --> B2[MRR - Mean Reciprocal Rank]
    B --> B3[NDCG]

    C --> C1[Faithfulness 忠实度]
    C --> C2[Answer Relevancy 答案相关性]
    C --> C3[Hallucination 幻觉检测]

    D --> D1[RAGAS Score]
    D --> D2[Human Evaluation]

    style B fill:#3498db,color:#fff
    style C fill:#e74c3c,color:#fff
    style D fill:#2ecc71,color:#fff

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

# Evaluate RAG pipeline with RAGAS
result = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,       # Is the answer grounded in context?
        answer_relevancy,   # Is the answer relevant to the question?
        context_precision,  # Are retrieved docs relevant?
        context_recall,     # Are all relevant docs retrieved?
    ],
)

print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88,
#  'context_precision': 0.85, 'context_recall': 0.78}

生产优化建议

分块大小调优：根据文档类型和查询模式实验不同的 chunk_size（256-1024）
元数据过滤：利用文档元数据（来源、时间、类型）缩小检索范围
缓存层：对热门查询的检索结果和生成结果进行缓存
流式输出：使用 Streaming 提升用户感知速度
对话历史：将对话历史融入 query 重写，实现多轮对话
监控体系：记录检索延迟、LLM 延迟、用户反馈等指标

总结

RAG 架构的核心在于”让 LLM 基于准确的外部知识回答问题”。从文档分块到检索再到生成，每个环节都有优化空间。混合检索（BM25 + 向量）配合重排序通常能显著提升检索质量；层次化分块和 query 重写则能进一步提升端到端效果。最重要的是建立完善的评估体系（如 RAGAS），用数据驱动优化迭代。

踩坑记录

公司内部搭建技术知识库，把 5 年的技术文档和 Wiki 接入 LLM，供团队 Q&A 查询。第一版用了最简单的方案：chunk_size=500 tokens，top-k=5，直接把检索结果拼进 prompt。

上线后用户反馈「文不对题」——问 K8s 相关问题，回答里全是 Redis 内容。排查发现：向量检索只看语义相似度，「如何处理高并发」这类通用问法会召回各种技术主题的 chunk，质量完全不可控。

第一次优化：加了基于文档标签的 keyword filter 做预过滤，把候选集从全库缩小到相关技术域，召回率从 52% 提升到 68%。

第二次优化：引入 Rerank 模型（BGE-reranker-v2-m3），对 top-k=20 的候选结果重排序后取 top-5，召回率进一步提升到 81%，用户满意度评分从 3.2 升到 4.1（满分 5）。

目前还在迭代的问题：chunk 边界切割导致上下文断裂，比如一个代码示例被切成两个 chunk，单独看都看不懂。正在试验按语义段落切分而非固定 token 数。

实测结果

测试集：200 个内部技术问答，由 3 名工程师人工标注标准答案

方案	召回率	答案相关性（1-5）	幻觉率	检索 P99 延迟
基础 RAG（top-k=5）	52%	3.2	23%	120ms
+ keyword filter	68%	3.6	15%	135ms
+ Rerank（BGE）	81%	4.1	9%	380ms

延迟增加 260ms（Rerank 的代价），但准确率提升明显，用户接受度高。

我的看法

RAG 不是「接个向量数据库就完事了」。在多个项目里见过的最大误区是过分相信向量检索的语义理解——向量模型对领域专有名词（特定产品名、版本号、错误码）非常不敏感，纯语义检索在技术知识库场景表现很差。

混合检索（向量 + BM25 关键词）几乎在任何场景都比纯向量检索更稳，实现也不复杂，建议作为默认方案。Rerank 是质量保障的最后一道防线，加上之后效果提升往往超预期，不要因为多 200ms 延迟就省掉这一步。

另外，RAG 系统一定要从第一天就建立评估体系，哪怕只是一个 50 条问题的小测试集。没有量化指标，优化就是盲飞。