RAG 检索增强生成：让 LLM 拥有私有知识

**AI/LLM 应用开发实战系列** 第 4 篇

上一篇：[LangChain 工作流](./2026-03-01-llm-series-03.md) | 下一篇：向量数据库选型与实战

什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种让 LLM 访问外部知识库的技术。

**核心思想**：在回答用户问题前，先从知识库中检索相关信息，然后将检索结果作为上下文提供给 LLM。

用户问题 → 检索相关知识 → 拼接成 Prompt → LLM 生成答案

为什么需要 RAG

直接使用 LLM 有以下局限：

| 问题 | 说明 | RAG 解决方案 |

|------|------|-------------|

| 知识截止 | GPT-4 只训练到 2024 年 4 月 | 检索最新信息 |

| 私有知识 | 无法访问企业内部文档 | 连接私有知识库 |

| 幻觉问题 | 可能编造不存在的信息 | 基于检索到的事实回答 |

| 上下文限制 | 无法放入所有知识 | 只检索相关内容 |

| 成本高昂 | 长上下文 token 费用高 | 只检索必要信息 |

RAG 工作流程

1. 文档加载 → 2. 文本分块 → 3. 向量化 → 4. 存储到向量数据库
                                              ↓
5. 用户提问 → 6. 问题向量化 → 7. 相似度检索 → 8. 拼接 Prompt → 9. LLM 生成

实战：构建文档问答系统

1. 安装依赖

pip install langchain langchain-openai chromadb pypdf

2. 加载 PDF 文档

from langchain_community.document_loaders import PyPDFLoader

# 加载 PDF
loader = PyPDFLoader("company_handbook.pdf")
documents = loader.load()

print(f"加载了 {len(documents)} 页文档")
print(f"第一页内容预览：{documents[0].page_content[:200]}")

3. 文本分块

直接将整篇文档放入 Prompt 会超出 token 限制。需要分块：

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块 1000 字符
    chunk_overlap=200,    # 重叠 200 字符，保持上下文连贯
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "！", "?", " ", ""]
)

chunks = text_splitter.split_documents(documents)
print(f"分成了 {len(chunks)} 个文本块")

4. 向量化并存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化存储
)

print(f"向量数据库已创建，包含 {vectorstore._collection.count()} 个向量")

5. 构建检索问答链

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

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

# 创建检索问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 将所有检索结果拼接到一个 Prompt
    retriever=vectorstore.as_retriever(
        search_type="similarity",  # 相似度搜索
        search_kwargs={"k": 3}     # 返回最相关的 3 个文本块
    ),
    return_source_documents=True   # 返回源文档（用于溯源）
)

# 提问
query = "公司的年假政策是什么？"
result = qa_chain.invoke({"query": query})

print(f"问题：{query}")
print(f"答案：{result['result']}")
print(f"\n参考来源:")
for doc in result['source_documents']:
    print(f"- {doc.metadata.get('source', 'unknown')}")

完整的 RAG 系统示例

让我们构建一个可以问答公司内部文档的完整系统：

import os
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# ============ 1. 配置 ============
os.environ["OPENAI_API_KEY"] = "your-api-key"
DATA_DIR = "./company_docs"  # 公司文档目录
DB_DIR = "./vector_db"       # 向量数据库目录

# ============ 2. 加载文档 ============
def load_documents():
    """加载多种格式的文档"""
    loaders = [
        DirectoryLoader(DATA_DIR, glob="**/*.pdf", loader_cls=PyPDFLoader),
        DirectoryLoader(DATA_DIR, glob="**/*.txt", loader_cls=TextLoader),
        DirectoryLoader(DATA_DIR, glob="**/*.md", loader_cls=TextLoader),
    ]
    
    documents = []
    for loader in loaders:
        documents.extend(loader.load())
    
    print(f"加载了 {len(documents)} 个文档")
    return documents

# ============ 3. 文本分块 ============
def split_documents(documents):
    """将文档分成小块"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["\n\n", "\n", "。", "！", "？", " ", ""]
    )
    chunks = text_splitter.split_documents(documents)
    print(f"分成了 {len(chunks)} 个文本块")
    return chunks

# ============ 4. 创建向量数据库 ============
def create_vectorstore(chunks):
    """创建或加载向量数据库"""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    if os.path.exists(DB_DIR):
        # 加载已有数据库
        vectorstore = Chroma(
            persist_directory=DB_DIR,
            embedding_function=embeddings
        )
    else:
        # 创建新数据库
        vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=embeddings,
            persist_directory=DB_DIR
        )
        vectorstore.persist()
    
    return vectorstore

# ============ 5. 创建问答链 ============
def create_qa_chain(vectorstore):
    """创建检索问答链"""
    llm = ChatOpenAI(model="gpt-4", temperature=0.7)
    
    # 自定义 Prompt 模板
    template = """基于以下参考信息回答问题。如果参考信息中没有答案，请说"根据现有文档，我无法回答这个问题"。

参考信息：
{context}

问题：{question}
答案："""

    prompt = PromptTemplate(
        template=template,
        input_variables=["context", "question"]
    )
    
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        return_source_documents=True,
        chain_type_kwargs={"prompt": prompt}
    )
    
    return qa_chain

# ============ 6. 主程序 ============
def main():
    # 初始化（首次运行需要）
    print("正在加载文档...")
    documents = load_documents()
    chunks = split_documents(documents)
    vectorstore = create_vectorstore(chunks)
    
    # 创建问答链
    qa_chain = create_qa_chain(vectorstore)
    
    # 交互式问答
    print("\n=== 文档问答系统已就绪 ===")
    print("输入问题开始提问，输入'quit'退出\n")
    
    while True:
        query = input("你：").strip()
        if query.lower() == 'quit':
            break
        
        result = qa_chain.invoke({"query": query})
        
        print(f"AI: {result['result']}")
        
        # 显示参考来源
        if result['source_documents']:
            print("\n参考来源:")
            for i, doc in enumerate(result['source_documents'], 1):
                source = doc.metadata.get('source', 'unknown')
                page = doc.metadata.get('page', '')
                print(f"  [{i}] {source}" + (f" 第{page}页" if page else ""))
            print()

if __name__ == "__main__":
    main()

优化技巧

1. 混合搜索（相似度 + 关键词）

retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance
    search_kwargs={
        "k": 5,         # 返回 5 个结果
        "fetch_k": 10,  # 先取 10 个，再多样性选择
        "lambda_mult": 0.5  # 相关性和多样性的平衡 (0-1)
    }
)

2. 多向量检索

from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import FAISS

# 结合多个检索器
faiss_retriever = faiss_vectorstore.as_retriever(search_kwargs={"k": 3})
chroma_retriever = chroma_vectorstore.as_retriever(search_kwargs={"k": 3})

ensemble_retriever = EnsembleRetriever(
    retrievers=[faiss_retriever, chroma_retriever],
    weights=[0.5, 0.5]
)

3. 添加对话历史

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    memory=memory
)

# 多轮对话
result = qa_chain({"question": "年假有多少天？"})
result = qa_chain({"question": "那病假呢？"})  # 会记住之前的对话

4. 流式输出

for chunk in qa_chain.stream({"query": "公司有哪些福利？"}):
    if 'result' in chunk:
        print(chunk['result'], end='', flush=True)

常见问题

Q: 检索结果不相关怎么办？

1. 调整分块大小（chunk_size）

2. 增加重叠（chunk_overlap）

3. 使用更好的嵌入模型（text-embedding-3-large）

4. 调整 k 值（返回结果数量）

Q: 回答质量不高怎么办？

1. 优化 Prompt 模板

2. 增加参考上下文数量

3. 降低 temperature（更确定性）

4. 在 Prompt 中明确要求"基于参考信息回答"

Q: 如何处理大量文档？

1. 使用增量更新（只添加新文档）

2. 使用分布式向量数据库（Milvus、Weaviate）

3. 实现文档版本管理

总结

RAG 让 LLM 可以：

• ✅ 访问最新信息
• ✅ 利用私有知识库
• ✅ 减少幻觉
• ✅ 降低 token 成本

• 下一篇我们将学习**向量数据库选型**，对比 FAISS、Chroma、Milvus 等主流方案。

  参考资源

• [LangChain RAG 文档](https://python.langchain.com/docs/use_cases/question_answering/)
• [RAG 论文](https://arxiv.org/abs/2005.11401)
• [Chroma 向量数据库](https://www.trychroma.com/)