第五章:RAG 检索增强——让 AI 拥有“长期记忆”
5.1 为什么 LLM 需要 RAG?
大语言模型(LLM) 虽然强大,但存在两个核心缺陷:
- 知识时效性:训练数据有截止日期(如 2024 年),无法知道之后的事件。
- 私有知识缺失:无法访问用户的私人文档、数据库、内部系统。
RAG(Retrieval Augmented Generation,检索增强生成) 通过**“检索 + 生成”**的闭环,解决了这两个问题:
- 检索:从外部知识库(如文档、数据库、互联网)中查找相关信息。
- 生成:将检索到的信息作为上下文,让 LLM 生成更准确的回答。
类比:
- 纯 LLM:就像让一个记忆力超群但无法查资料的人回答问题。
- RAG:就像让一个会查资料的人回答问题,先查再答。
5.2 RAG 的核心流程
┌─────────────────────────────────────────────────────────────┐
│ 用户提问 (User Query) │
│ "OpenClaw 的 MCP 协议是什么?" │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 检索阶段 (Retrieval) │
│ 1. 将问题转换为向量 (Embedding) │
│ 2. 在向量数据库中搜索相似文档片段 │
│ 3. 返回 Top-K 个最相关的片段 │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 增强阶段 (Augmentation) │
│ 将检索到的片段拼接到 Prompt 中: │
│ "已知信息:{检索结果} │
│ 请回答:{用户问题}" │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ 生成阶段 (Generation) │
│ LLM 基于增强后的 Prompt 生成回答 │
└─────────────────────────────────────────────────────────────┘
5.3 分块策略:如何切割文档?
RAG 的第一步是将长文档切割成小块(Chunks),以便检索。
5.3.1 固定分块(Fixed-size Chunking)
方法:按固定字符数切割(如每 500 字符一块)。
优点:简单、快速。
缺点:可能切断语义(如一句话被切成两半)。
示例:
原文:"OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。"
固定分块(每 20 字符):
块 1: "OpenClaw 是一个开源"
块 2: "的 AI 助手。它支持 MC"
块 3: "P 协议。"
5.3.2 语义分块(Semantic Chunking)
方法:按语义边界切割(如段落、句子)。
优点:保持语义完整,检索更准确。
缺点:实现复杂,需要 NLP 模型。
示例:
原文:"OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。"
语义分块:
块 1: "OpenClaw 是一个开源的 AI 助手。"
块 2: "它支持 MCP 协议。"
5.3.3 实战:Python 实现语义分块
import re
def semantic_chunking(text, max_length=500):
"""按段落和句子进行语义分块"""
chunks = []
current_chunk = ""
# 按段落分割
paragraphs = re.split(r'\n\s*\n', text)
for para in paragraphs:
# 按句子分割
sentences = re.split(r'(?<=[。!?.!?])\s*', para)
for sentence in sentences:
if len(current_chunk) + len(sentence) <= max_length:
current_chunk += sentence + " "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence + " "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
# 使用
text = """
OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。
MCP 协议定义了工具调用的标准格式。
"""
chunks = semantic_chunking(text)
for i, chunk in enumerate(chunks):
print(f"块 {i+1}: {chunk}")
输出:
块 1: OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。
块 2: MCP 协议定义了工具调用的标准格式。
5.4 向量索引:ChromaDB vs FAISS
分块后,需要将每个块转换为向量(Embedding),并存入向量数据库。
5.4.1 常用向量数据库
| 数据库 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| ChromaDB | 简单易用,Python 原生 | 性能一般 | 小型项目、快速原型 |
| FAISS | 性能高,支持大规模 | 配置复杂 | 生产环境、大规模数据 |
| Weaviate | 功能丰富,支持混合搜索 | 资源消耗大 | 企业级应用 |
| Qdrant | 速度快,支持过滤 | 学习曲线陡 | 高性能需求 |
5.4.2 实战:使用 ChromaDB 构建 RAG
import chromadb
from sentence_transformers import SentenceTransformer
# 1. 初始化
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(name="openclaw_docs")
model = SentenceTransformer("all-MiniLM-L6-v2") # 轻量级 Embedding 模型
# 2. 分块并嵌入
text = """
OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。
MCP 协议定义了工具调用的标准格式。
Agent Reach 让 AI 能够访问互联网。
"""
chunks = semantic_chunking(text)
embeddings = model.encode(chunks).tolist()
# 3. 存入数据库
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
collection.add(
ids=[f"doc_{i}"],
documents=[chunk],
embeddings=[embedding]
)
# 4. 查询
query = "什么是 MCP 协议?"
query_embedding = model.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=3
)
print("检索结果:")
for i, doc in enumerate(results["documents"][0]):
print(f"{i+1}. {doc}")
输出:
检索结果:
1. MCP 协议定义了工具调用的标准格式。
2. OpenClaw 是一个开源的 AI 助手。它支持 MCP 协议。
3. Agent Reach 让 AI 能够访问互联网。
5.5 实战:构建技术文档问答机器人
5.5.1 场景
用户提问:"OpenClaw 如何配置 MCP Server?"
系统流程:
- 检索:从 OpenClaw 文档中检索相关片段。
- 增强:将检索结果拼接到 Prompt。
- 生成:LLM 生成回答。
5.5.2 代码实现
import chromadb
from sentence_transformers import SentenceTransformer
import requests
class RAGBot:
def __init__(self):
self.client = chromadb.PersistentClient(path="./chroma_db")
self.collection = self.client.get_collection(name="openclaw_docs")
self.model = SentenceTransformer("all-MiniLM-L6-v2")
self.llm_api = "https://api.qwen.ai/v1/chat/completions"
def query(self, question):
# 1. 检索
query_embedding = self.model.encode([question]).tolist()
results = self.collection.query(
query_embeddings=query_embedding,
n_results=5
)
context = "\n".join(results["documents"][0])
# 2. 增强 Prompt
prompt = f"""
已知信息:
{context}
请根据以上信息回答以下问题:
{question}
如果信息不足,请说明"根据现有资料无法回答"。
"""
# 3. 生成
response = requests.post(
self.llm_api,
json={
"model": "qwen-max",
"messages": [{"role": "user", "content": prompt}]
}
)
answer = response.json()["choices"][0]["message"]["content"]
return answer
# 使用
bot = RAGBot()
answer = bot.query("OpenClaw 如何配置 MCP Server?")
print(answer)
输出示例:
根据 OpenClaw 文档,配置 MCP Server 的步骤如下:
- 编辑
~/.mcporter/mcporter.json文件。- 添加 MCP Server 配置,包括
command、args、transport等字段。- 保存后,使用
mcporter list验证配置。- 使用
mcporter call <server> <tool> <params>调用工具。
5.6 深度思考:RAG 能否解决知识时效性问题?
5.6.1 RAG 的优势
- 实时更新:只需更新向量数据库,无需重新训练 LLM。
- 低成本:检索比微调便宜得多。
- 可解释性:可以追溯答案来源(引用原文)。
5.6.2 RAG 的局限
- 检索质量:如果检索不到相关信息,RAG 无法生成正确答案。
- 上下文限制:LLM 的上下文长度有限,无法容纳太多检索结果。
- 幻觉问题:LLM 仍可能基于错误信息生成幻觉。
5.6.3 未来方向
- 混合搜索:结合关键词搜索(BM25)和向量搜索(Embedding)。
- 重排序(Re-ranking):对检索结果进行二次排序,提升相关性。
- 多跳检索:多次检索,逐步逼近答案。
- 实时索引:自动抓取最新文档,实时更新向量数据库。
结论: RAG 是解决 LLM 知识时效性的最佳实践,但并非完美。未来需要检索 + 生成 + 验证的闭环,才能真正实现“可靠的知识问答”。
(第五章完)
下一章预告:第六章将探讨容错与自愈,如何让 Agent 在出错时自动恢复。