RAG（Retrieval-Augmented Generation，检索增强生成）是当前AI应用开发最核心的技术之一。今天用大白话讲清楚它的原理和实现。

为什么需要RAG？

大语言模型（LLM）有个致命缺陷：训练数据有截止日期，不知道最新信息。

比如：

• GPT-4的知识截止到2023年4月
• 问它"2025年10月的新闻"，它答不出来
• 问它"你们公司内部文档规定的报销流程"，它更不知道

RAG就是解决这个问题的：在调用大模型前，先从你的知识库里检索相关内容，然后一起喂给模型。

RAG的工作流程

传统方式（不用RAG）

用户提问 → 大模型 → 回答

问题：模型只能基于训练数据回答，无法获取新知识。

使用RAG

用户提问
  ↓
1. 检索相关文档（向量搜索）
  ↓
2. 把文档 + 问题一起发给模型
  ↓
3. 模型基于检索到的内容生成回答

核心技术：Embedding（向量化）

什么是Embedding？

把文本转成数字数组（向量），语义相近的文本，向量也相近。

举例：

"苹果很好吃" → [0.2, 0.8, 0.1, ...]
"香蕉味道不错" → [0.3, 0.7, 0.2, ...]  # 向量接近
"今天天气真好" → [0.9, 0.1, 0.8, ...]  # 向量很远

为什么用向量而不是关键词搜索？

关键词搜索的问题：

问题："如何提升销售额？"
文档A："增加营收的10个方法"  # 没有"销售额"关键词，搜不到
文档B："销售额统计表"  # 有关键词，但不相关，却被搜到

向量搜索：

• 理解语义：知道"销售额"和"营收"意思相近
• 更智能：找到真正相关的内容

实现RAG的5个步骤

步骤1：文档准备

# 支持多种格式
documents = [
    "公司内部文档.pdf",
    "产品手册.docx",
    "网页内容.html",
    "Notion笔记"
]

步骤2：文档分块（Chunking）

不能把整个文档都喂给模型（Token限制），需要切成小块。

分块策略：

• 按字符数：每1000个字符一块（简单但粗暴）
• 按段落：每个段落一块（保持完整性）
• 语义分块：用AI识别主题切换点（最智能）

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块1000字符
    chunk_overlap=200,    # 块之间重叠200字符（避免切断语义）
)

chunks = splitter.split_documents(documents)

为什么要overlap（重叠）：

块1：...产品的主要功能包括A、B、[C]
块2：[C]、D、E。这些功能可以...

重叠部分保证不会把"功能C的描述"切成两半。

步骤3：向量化存储

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

# 向量化
embeddings = OpenAIEmbeddings()

# 存入向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 本地存储
)

步骤4：检索相关内容

# 用户提问
query = "如何提升团队协作效率？"

# 检索Top-K个最相关的文档块
docs = vectorstore.similarity_search(query, k=3)

# 结果：
# [
#   "使用项目管理工具可以提升协作效率...",
#   "定期站会是保持团队同步的有效方式...",
#   "明确的分工和责任能减少重复劳动..."
# ]

步骤5：生成回答

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

# 构建QA链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4"),
    retriever=vectorstore.as_retriever(),
    return_source_documents=True  # 返回引用来源
)

# 提问
result = qa_chain({"query": "如何提升团队协作效率？"})

print(result["result"])
# → 基于检索到的内容生成的专业回答

print(result["source_documents"])
# → 显示引用了哪些文档（可追溯）

生产环境的优化技巧

优化1：选择合适的Embedding模型

我的建议：

• 测试阶段：OpenAI（方便）
• 生产环境：评估成本后决定（每1M Token约$0.10）

优化2：改进检索质量

方法1：Reranking（重排序）

检索出10个候选，用更强的模型重新排序，取Top-3。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

compressor = CohereRerank()
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# 检索质量显著提升
docs = compression_retriever.get_relevant_documents(query)

方法2：Hybrid Search（混合搜索）

向量搜索 + 关键词搜索，取并集。

from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever

# 向量检索器
vector_retriever = vectorstore.as_retriever()

# 关键词检索器
bm25_retriever = BM25Retriever.from_documents(chunks)

# 混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]  # 向量权重0.7，关键词权重0.3
)

方法3：Query Expansion（查询扩展）

把用户的问题改写成多个版本，分别检索，合并结果。

original_query = "如何提升销售额？"

# 用LLM生成3个改写版本
expanded_queries = [
    "如何提升销售额？",
    "增加营收的方法有哪些？",
    "提高销售业绩的策略"
]

# 分别检索，合并去重
all_docs = []
for q in expanded_queries:
    docs = vectorstore.similarity_search(q, k=2)
    all_docs.extend(docs)

# 去重
unique_docs = list(set(all_docs))

优化3：成本控制

缓存机制

import hashlib
import json

def get_cached_response(query, cache_dict):
    query_hash = hashlib.md5(query.encode()).hexdigest()
    if query_hash in cache_dict:
        return cache_dict[query_hash]
    return None

def cache_response(query, response, cache_dict):
    query_hash = hashlib.md5(query.encode()).hexdigest()
    cache_dict[query_hash] = response

只给关键块做Embedding

不是所有内容都需要向量化：

• ❌ 目录、页眉页脚、无关紧要的内容
• ✅ 核心内容、问答对、知识点

用更便宜的模型

# 检索用便宜模型
retriever_llm = ChatOpenAI(model="gpt-3.5-turbo")

# 生成回答用强模型
answer_llm = ChatOpenAI(model="gpt-4")

常见问题和解决方案

问题1：检索到的内容不相关

原因：

• 分块策略不合理（切得太碎）
• Embedding模型不适合你的领域
• 问题表述不清晰

解决：

• 调整chunk_size和overlap
• 尝试不同的Embedding模型
• 用LLM重写用户的问题（Query Rewriting）

问题2：回答不准确

原因：

• 检索到的内容太少或太多
• Prompt没有强调"基于检索内容回答"

解决：

prompt_template = """
请基于以下参考内容回答问题。如果参考内容中没有答案，明确说"我不知道"，不要编造。

参考内容：
{context}

问题：{question}

回答：
"""

问题3：速度太慢

原因：

• 向量数据库查询慢
• 检索的文档太多

解决：

• 用更快的向量数据库（Milvus、Qdrant）
• 减少检索数量（k=3而不是k=10）
• 用异步处理

真实案例：企业知识库助手

我给一家公司做的内部知识库系统：

需求：

• 1000+篇内部文档（规章制度、操作手册、历史案例）
• 员工随时提问，秒级响应
• 必须标注信息来源（可追溯）

技术方案：

• 向量数据库：Qdrant（开源、快）
• Embedding：OpenAI Ada-002
• LLM：GPT-3.5（成本考虑）
• 优化：Reranking + 缓存

效果：

• 问题解决率：85%
• 平均响应时间：3秒
• 月成本：$200（500名员工）

总结

RAG是AI应用的核心技术，掌握它你就能做：

• 企业知识库
• 个人笔记助手
• 客服机器人
• 代码文档问答

关键要点：

1. 合理的分块策略
2. 高质量的Embedding
3. 优化检索算法
4. 控制成本

下一步建议：自己动手实现一个简单的RAG系统，代码不超过100行。

推荐阅读：

• 《20年老程序员的AI学习路径》
• 《企业AI转型实践》

相关资源：

• 完整代码示例
• 向量数据库选型指南