今天想跟大家分享一个让大语言模型（LLM）“开挂”的技巧
——RAG：检索增强生成（Retrieval-Augmented Generation）。
如果你用过 ChatGPT 或其他大模型，一定遇到过这些问题：
🔍 知识过时：模型可能只知道训练数据截止前的事情。
🔒 无法访问私有资料：比如公司的内部文档、最新的 PDF 报告。
这就是 LLM 的天然局限。而 RAG 的目标，就是给模型实时查资料的能力，让它不仅靠“记忆”，还能像我们一样去翻文档、查数据库，然后再回答问题。
1️⃣ RAG 的工作原理
简单来说，RAG 的流程是：
1、载入文档（PDF、Word、网页等），切成可处理的小块。
2、把每一块转成向量（Embedding），存进一个专门的“向量数据库”。
3、用户提问 → 系统先去数据库里找最相关的内容。
4、把找到的内容 + 用户问题一起发给大模型，模型就能在最新资料的基础上作答。
可以理解为：
“模型记忆”+“实时检索” = 更准确、更专业的回答。
2️⃣ 文档处理的关键技巧
很多人做 RAG 的第一步就是“切文本”，但如果切得太随意，会导致模型回答不完整或者缺乏上下文。
1、粒度太大：模型一次读不完，检索不精准。
2、粒度太小：上下文被打断，模型回答不连贯。
3、改进方法：定长 + 重叠切块。
比如我们用 chunk_size=500 字符，overlap_size=120 字符，保证每一块之间有部分重叠，这样模型在回答时能更完整地理解上下文。
chunks = split_text(paragraphs, chunk_size=500, overlap_size=120)
3️⃣ 检索的两种主流方式
1、关键词检索（Keyword Search）
类似于搜索引擎，精确匹配词语。适合查找专有名词、缩写等。
2、向量检索（Vector Search）
通过文本向量化后计算语义相似度，即使不完全匹配关键词，也能找到语义相近的内容。
常用的相似度算法包括 余弦相似度（cosine similarity） 和 欧氏距离。
👉 实际应用中，混合检索效果最好：关键词搜索精准，向量搜索智能，把两者结合往往能比单独用一种更好。
4️⃣ 向量数据库的选择
向量数据库是 RAG 系统的“记忆库”。常见的有：
5️⃣ 大模型接入
检索只是“找资料”，最终的回答还是要靠大模型来生成。
在示例代码里，我用的是本地部署的 Ollama (llama3) 来回答中文问题：
answer = get_completion_ollama(prompt, model="llama3")
如果你用的是 OpenAI API，只要改成 gpt-4 或 gpt-4o 就能直接用。
6️⃣ 实战：ChatPDF 项目
为了方便大家理解，我写了一个完整的 RAG_ChatPDF.py，它可以：
1、自动读取 PDF 文档。
2、用“定长+重叠”的方式切分文本。
3、用 ChromaDB 建立向量库。
4、用本地的 LLaMA3 模型回答问题。
启动后，你只需要输入问题，就能用自己的 PDF 文档训练的“知识型 ChatGPT”回答你。
🌟 总结
RAG 让大模型可以“读最新文档+实时查知识”，特别适合企业内部知识库、论文问答、客户支持等场景。
核心流程是：切文档 → 向量化 → 建库检索 → 调用大模型回答。
通过合理的切块、混合检索和合适的向量数据库，可以显著提高回答准确度和实用性。
如果你想自己动手做一个 PDF 问答助手，直接看下面的完整代码就能上手。
```python
# -*- coding: utf-8 -*-
"""
RAG_ChatPDF.py
- 读取 PDF -> 句子级 + 交叠式切块
- ChromaDB 建向量库
- 用 Ollama (llama3) 中文回答
"""
import os
import json
import re
import requests
import fitz # PyMuPDF
import chromadb
from chromadb.config import Settings
import torch
from transformers import AutoTokenizer, AutoModel
# ========== 1) 读取 PDF ==========
def extract_text_from_pdf(pdf_path: str) -> str:
doc = fitz.open(pdf_path)
text = []
for page_num in range(doc.page_count):
page = doc.load_page(page_num)
text.append(page.get_text())
return "\n".join(text)
# ========== 2) 更鲁棒的“定长+交叠”切块 ==========
# 英文句子切分（可选）
try:
from nltk.tokenize import sent_tokenize
except Exception:
sent_tokenize = None
def _fallback_sentence_split(p: str):
"""兜底的简单分句：英文按 .!?; 中文按。！？；，"""
# 先用英文界定
parts = re.split(r"(?<=[\.\!\?;])\s+", p.strip())
out = []
for seg in parts:
# 再粗分中文标点，防止太长
out.extend([s for s in re.split(r"(?<=[。！？；，])", seg) if s and s.strip()])
return [s.strip() for s in out if s.strip()]
def sentence_tokenize_paragraphs(paragraphs):
"""把多段文本切成句子列表。优先用 nltk.sent_tokenize（英文），否则用兜底。"""
sentences = []
for p in paragraphs:
p = p.strip()
if not p:
continue
if sent_tokenize:
try:
ss = sent_tokenize(p)
if ss and len(" ".join(ss)) >= 3:
sentences.extend([s.strip() for s in ss if s.strip()])
continue
except Exception:
pass
# 兜底
sentences.extend(_fallback_sentence_split(p))
return sentences
def split_text(paragraphs, chunk_size=300, overlap_size=100):
"""
- 先句子化
- 前向把句子拼到 chunk_size
- 追加时，从“上一个 chunk 的末尾”回溯 overlap_size 字符，作为 overlap
"""
sentences = [s.strip() for p in paragraphs for s in sentence_tokenize_paragraphs([p])]
chunks = []
i = 0
while i < len(sentences):
# 当前块，从第 i 个句子开始
chunk = sentences[i]
# 计算向前的重叠部分（从 i-1 往回拼）
overlap = ''
prev = i - 1
while prev >= 0 and len(sentences[prev]) + len(overlap) <= overlap_size:
overlap = sentences[prev] + ' ' + overlap
prev -= 1
chunk = overlap + chunk
# 再往后拼，直到到达 chunk_size
next_idx = i + 1
while next_idx < len(sentences) and len(sentences[next_idx]) + len(chunk) <= chunk_size:
chunk = chunk + ' ' + sentences[next_idx]
next_idx += 1
chunks.append(chunk.strip())
i = next_idx # 跳到下一个起点
return chunks
# ========== 3) ChromaDB 连接器 ==========
class MyVectorDBConnector:
def __init__(self, collection_name, embedding_fn):
# 允许 reset，便于反复运行
self.chroma_client = chromadb.Client(Settings(allow_reset=True))
self.chroma_client.reset()
self.collection = self.chroma_client.get_or_create_collection(collection_name)
self.embedding_fn = embedding_fn
def add_documents(self, documents):
self.collection.add(
embeddings=self.embedding_fn(documents),
documents=documents,
ids=[f"id{i}" for i in range(len(documents))]
)
def search(self, query, top_n=3):
return self.collection.query(
query_embeddings=self.embedding_fn([query]),
n_results=top_n
)
# ========== 4) 向量（embedding） ==========
# 注：你本机已成功使用 all-MiniLM-L6-v2；中文也能用，但多语更好（若联网可换 paraphrase-multilingual-MiniLM-L12-v2）
_EMB_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
_tokenizer = None
_model = None
def _load_embedder():
global _tokenizer, _model
if _tokenizer is None or _model is None:
_model = AutoModel.from_pretrained(_EMB_MODEL)
_tokenizer = AutoTokenizer.from_pretrained(_EMB_MODEL)
def get_embeddings(texts):
_load_embedder()
inputs = _tokenizer(texts, return_tensors="pt", padding=True, truncation=True)
with torch.no_grad():
embs = _model(**inputs).last_hidden_state.mean(dim=1).cpu().numpy()
return embs
# ========== 5) 调用 Ollama (与 get_completion_ollama.py 一致) ==========
def get_completion_ollama(prompt: str, model: str = "llama3"):
url = "http://localhost:11434/api/chat"
headers = {"Content-Type": "application/json"}
data = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"stream": False
}
resp = requests.post(url, headers=headers, json=data, timeout=120)
resp.raise_for_status()
result = resp.json()
return result["message"]["content"]
# ========== 6) RAG 机器人（中文提示词） ==========
class RAG_Bot:
def __init__(self, vector_db, n_results=3):
self.vector_db = vector_db
self.n_results = n_results
def build_prompt(self, context_docs, query):
# 把检索到的若干片段做成中文提示词
ctx = "\n\n".join([f"【片段{i+1}】\n{doc}" for i, doc in enumerate(context_docs)])
prompt = (
"你是一个严谨的中文助手。请仅依据下面提供的参考片段，用简洁中文回答用户问题；"
"如果资料里没有答案，请明确说明“未在资料中找到直接答案”。\n\n"
f"{ctx}\n\n"
f"用户问题：{query}\n\n"
"请用中文作答。"
)
return prompt
def chat(self, user_query: str):
search_results = self.vector_db.search(user_query, self.n_results)
docs = search_results.get("documents", [[]])[0]
prompt = self.build_prompt(docs, user_query)
try:
answer = get_completion_ollama(prompt, model="llama3")
except Exception as e:
answer = f"调用本地大模型失败：{e}"
return answer
# ========== 7) 主程序 ==========
if __name__ == "__main__":
# ---- 修改为你的 PDF 路径 ----
pdf_path = "/Users/axia/Documents/Alex/ai_2025/research_manufacturing_industry.pdf"
if not os.path.exists(pdf_path):
print(f"❌ 找不到 PDF：{pdf_path}")
exit(1)
print("正在读取 PDF ...")
text = extract_text_from_pdf(pdf_path)
print("正在切分文本 ...")
# 先按换行粗分为段；再做“定长+交叠”的切块
paragraphs = [p.strip() for p in text.split("\n") if p.strip()]
chunks = split_text(paragraphs, chunk_size=500, overlap_size=120)
print(f"共得到切块：{len(chunks)}")
print("加载/构建向量库 ...")
vector_db = MyVectorDBConnector("pdf_collection_overlap", get_embeddings)
vector_db.add_documents(chunks)
# 交互式问答
print("✅ 准备就绪。开始提问吧！输入 q / quit / exit 退出。")
bot = RAG_Bot(vector_db, n_results=3)
while True:
try:
user_query = input("\n你的问题> ").strip()
except (EOFError, KeyboardInterrupt):
print("\n👋 已退出。")
break
if user_query.lower() in {"q", "quit", "exit"}:
print("👋 已退出。")
break
if not user_query:
continue
print("\n—— 答复 ——")
resp = bot.chat(user_query)
print(resp)
```
