LLM应用架构：生产环境RAG系统设计

RAG（Retrieval-Augmented Generation）是当前LLM应用最主流的架构模式。我在多个企业知识库问答项目中实践了从文档处理到生产部署的完整流程，这篇文章分享生产环境RAG系统的设计要点和最佳实践。

RAG架构概览

一个完整的RAG系统包含以下核心模块：

1. 文档加载与预处理：从各种数据源加载文档，清洗和分块
2. 向量化（Embedding）：将文本转化为向量表示
3. 向量存储：将向量存入向量数据库
4. 检索：根据用户查询检索相关文档
5. 生成：将检索结果作为上下文，调用LLM生成回答

文档加载与预处理

文档预处理是RAG系统的基础，直接影响检索和生成效果。

1. 多格式文档加载

from langchain.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    UnstructuredFileLoader,
    TextLoader,
    WebBaseLoader
)

def load_document(file_path):
    if file_path.endswith('.pdf'):
        loader = PyPDFLoader(file_path)
    elif file_path.endswith('.docx'):
        loader = Docx2txtLoader(file_path)
    elif file_path.endswith('.txt'):
        loader = TextLoader(file_path)
    elif file_path.startswith('http'):
        loader = WebBaseLoader(file_path)
    else:
        loader = UnstructuredFileLoader(file_path)
    
    documents = loader.load()
    return documents

2. 文档分块策略

分块大小是RAG系统最关键的参数之一，需要根据文档类型和业务场景调整。

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    TokenTextSplitter,
    SentenceTransformersTokenTextSplitter
)

def split_documents(documents, chunk_size=512, chunk_overlap=50):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=['\n\n', '\n', '。', '！', '？', '.', '!', '?', ' ']
    )
    
    chunks = text_splitter.split_documents(documents)
    return chunks

3. 文档元数据增强

def enrich_metadata(documents, source_type='manual'):
    for doc in documents:
        doc.metadata['source_type'] = source_type
        doc.metadata['chunk_id'] = uuid.uuid4().hex[:8]
        doc.metadata['created_at'] = datetime.now().isoformat()
        doc.metadata['language'] = detect_language(doc.page_content)
    
    return documents

向量化与Embedding选型

Embedding模型的选择直接影响检索质量，需要根据业务场景权衡。

1. 开源Embedding模型

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')

def embed_text(texts, batch_size=32):
    embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        batch_embeddings = model.encode(batch)
        embeddings.extend(batch_embeddings)
    
    return np.array(embeddings)

2. 商用Embedding API

from openai import OpenAI

client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def embed_with_openai(texts):
    response = client.embeddings.create(
        input=texts,
        model="text-embedding-3-small"
    )
    
    embeddings = [item.embedding for item in response.data]
    return np.array(embeddings)

3. 国产Embedding模型

from langchain.embeddings import QianfanEmbeddingsEndpoint

embeddings = QianfanEmbeddingsEndpoint(
    qianfan_ak=os.getenv('QIANFAN_AK'),
    qianfan_sk=os.getenv('QIANFAN_SK'),
    model="text-embedding-v1"
)

模型	维度	优点	缺点
all-MiniLM-L6-v2	384	轻量、免费、速度快	语义理解能力有限
text-embedding-3-small	1536	质量高、API易用	付费、有调用限制
text-embedding-v1（千帆）	384/1024	中文支持好、国内部署	质量略逊于OpenAI

向量数据库选型与配置

1. Chroma（轻量级首选）

from langchain_community.vectorstores import Chroma

vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

vectorstore.add_documents(documents=chunks)

2. Pinecone（生产环境首选）

from langchain_pinecone import Pinecone

vectorstore = Pinecone.from_documents(
    documents=chunks,
    embedding=embeddings,
    index_name="knowledge-base"
)

3. Milvus（开源高性能）

from langchain_community.vectorstores import Milvus

vectorstore = Milvus.from_documents(
    documents=chunks,
    embedding=embeddings,
    connection_args={
        "host": "localhost",
        "port": "19530"
    },
    collection_name="knowledge_base"
)

检索策略优化

简单的相似度检索往往不够，需要多种检索策略组合。

1. 混合检索（Hybrid Search）

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain_community.retrievers import PineconeHybridSearchRetriever

bm25_retriever = BM25Retriever.from_documents(chunks)
vector_retriever = vectorstore.as_retriever(k=5)

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

2. 上下文压缩（Context Compression）

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=vector_retriever
)

3. 查询扩展（Query Expansion）

def expand_query(query, llm):
    prompt = f"""请帮我扩展以下查询，生成2-3个相关的查询变体：
    原查询：{query}
    
    扩展查询："""
    
    response = llm.invoke(prompt)
    expanded_queries = [query] + [q.strip() for q in response.content.split('\n') if q.strip()]
    
    return expanded_queries

expanded_queries = expand_query("如何申请退款？", llm)
all_results = []
for q in expanded_queries:
    results = vectorstore.similarity_search(q, k=3)
    all_results.extend(results)

LLM集成与Prompt设计

1. 基础RAG Prompt

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的知识问答助手。
    请根据以下参考文档回答用户的问题：
    
    {context}
    
    回答要求：
    1. 必须基于参考文档的内容，不要编造信息
    2. 如果文档中没有相关信息，请明确说明"文档中没有相关信息"
    3. 回答要简洁明了，直接针对用户的问题
    4. 引用文档中的关键点时，可以标注来源"""),
    ("human", "{question}")
])

2. 高级RAG模式：RAG-Fusion

def rag_fusion(query, retriever, llm, k=5):
    expanded_queries = expand_query(query, llm)
    
    all_results = []
    for q in expanded_queries:
        results = retriever.get_relevant_documents(q)
        all_results.extend(results)
    
    fused_results = rerank_with_llm(query, all_results, k=k)
    
    context = "\n\n".join([doc.page_content for doc in fused_results])
    answer = llm.invoke(f"基于以下内容回答问题：{context}\n\n问题：{query}")
    
    return answer.content

3. 流式输出（Streaming）

from langchain_core.callbacks import StreamingStdOutCallbackHandler

llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

生产部署架构

生产环境的RAG系统需要考虑性能、可靠性和可扩展性。

1. 异步向量化流水线

import asyncio
from celery import Celery

app = Celery('rag_pipeline', broker='redis://localhost:6379/0')

@app.task
def process_document(file_path):
    documents = load_document(file_path)
    chunks = split_documents(documents)
    chunks = enrich_metadata(chunks)
    
    vectorstore.add_documents(chunks)
    
    return {"status": "success", "chunks_count": len(chunks)}

2. API服务封装

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="RAG Knowledge Base API")

class QueryRequest(BaseModel):
    question: str
    k: int = 5
    use_hybrid: bool = True

class QueryResponse(BaseModel):
    answer: str
    sources: list
    latency: float

@app.post("/query", response_model=QueryResponse)
async def query_knowledge_base(request: QueryRequest):
    start_time = time.time()
    
    if request.use_hybrid:
        retriever = ensemble_retriever
    else:
        retriever = vector_retriever
    
    docs = retriever.get_relevant_documents(request.question, k=request.k)
    
    context = "\n\n".join([doc.page_content for doc in docs])
    answer = llm.invoke(prompt.format(context=context, question=request.question))
    
    latency = time.time() - start_time
    
    return QueryResponse(
        answer=answer.content,
        sources=[doc.metadata.get('source', '') for doc in docs],
        latency=round(latency, 2)
    )

3. 缓存策略

from langchain_core.caches import InMemoryCache
from langchain_community.cache import RedisCache
import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)
llm.cache = RedisCache(redis_client=redis_client)

@lru_cache(maxsize=1000)
def cached_retrieve(query: str, k: int = 5):
    return retriever.get_relevant_documents(query, k=k)

RAG评估与优化

上线后的RAG系统需要持续评估和优化。

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

result = evaluate(
    dataset=test_dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision
    ],
    llm=llm,
    embeddings=embeddings
)

实际项目踩的坑

坑一：文档分块不当。分块太大导致上下文冗余，分块太小导致信息碎片化。解决方案：根据文档类型调整分块大小，使用语义分块。

坑二：检索结果不相关。用户问的是A，检索回来的是B。解决方案：使用混合检索、查询扩展、Rerank。

坑三：LLM幻觉。回答中包含文档中没有的信息。解决方案：在Prompt中明确要求基于文档回答，使用RAGAS评估faithfulness。

坑四：性能瓶颈。大文档向量化慢，查询响应时间长。解决方案：异步处理文档、添加缓存、使用更轻量的Embedding模型。

总结

RAG系统的核心是"好的文档处理 + 高质量检索 + 合理的Prompt设计"。一个优秀的RAG系统需要：

1. 精心设计的文档预处理流程：分块、元数据增强、格式转换
2. 合适的Embedding模型：根据业务场景选择开源或商用模型
3. 多种检索策略组合：向量检索 + 关键词检索 + Rerank
4. 清晰的Prompt指令：明确要求基于文档回答，避免幻觉
5. 完善的评估体系：使用RAGAS等工具持续评估和优化

记住：RAG不是一次性工程，而是需要持续迭代的系统。随着文档积累和用户反馈，系统会越来越好。