RAG 系统完整实现教程

RAG 工作流程

RAG = Retrieval（检索）+ Augmented（增强）+ Generation（生成）。核心是在生成答案前从知识库检索相关内容，再一起交给大模型生成更准确的答案。

技术栈选择

本教程使用：文档处理（langchain）、向量数据库（Chroma/Qdrant）、Embedding（ChinaWHAPI）、生成模型（ChinaWHAPI）。

• 文档加载：langchain document loaders
• 文本切分：RecursiveCharacterTextSplitter
• 向量数据库：Chroma（轻量）或 Qdrant（生产级）
• Embedding + 生成：ChinaWHAPI

步骤 1：文档加载与切分

from langchain.document_loaders import TextLoader, PDFPlumberLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

loader = PDFPlumberLoader("./document.pdf")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", " "]
)
texts = splitter.split_documents(documents)
print(f"切分完成，共 {len(texts)} 个 chunk")

步骤 2：Embedding 向量化

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

embeddings = OpenAIEmbeddings(
    openai_api_key="your_chinawhapi_key",
    openai_api_base="https://chinawhapi.com/v1",
    model="qwen3-vl-plus",  # 或使用专门的 embedding 模型
)

vectorstore = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./vectorstore"
)
vectorstore.persist()

步骤 3：向量检索

retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}  # 检索最相关的 5 个 chunk
)

query = "这份合同的主要风险点是什么？"
docs = retriever.get_relevant_documents(query)
for i, doc in enumerate(docs):
    print(f"\n--- Chunk {i+1} (相似度相关) ---")
    print(doc.page_content[:300])

步骤 4：RAG 生成

from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

llm = ChatOpenAI(
    model="qwen3.6-plus",
    openai_api_key="your_chinawhapi_key",
    openai_api_base="https://chinawhapi.com/v1",
    temperature=0.3,
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
)

result = qa_chain({"query": "这份合同的主要风险点是什么？"})
print(result["result"])

步骤 5：混合检索（可选）

结合向量检索和关键词检索（BM25）可以提升检索质量。

from langchain.retrievers import EnsembleRetriever

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
bm25_retriever = BM25Retriever.from_documents(texts)

ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 向量 60%，BM25 40%
)

生产环境优化

RAG 系统上生产需要关注的几个关键点：

• 检索质量：定期评估 precision@K 和 recall@K
• Embedding 更新：新文档加入时重新索引
• 成本控制：简单问题用快速模型（Qwen3.5 Flash）做路由
• 缓存：对相同问题做语义缓存，命中率可达 40-60%