RAG 让你的 AI 更聪明

今天想跟大家分享一个让大语言模型（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 问答助手，直接看下面的完整代码就能上手。

# -*- coding: utf-8 -*-
"""
RAG_ChatPDF.py
- 读取 PDF -> 句子级 + 交叠式切块
- ChromaDB 建向量库
- 用 Ollama (llama3) 中文回答
"""import os
import json
import re
import requests
import fitz  # PyMuPDFimport chromadb
from chromadb.config import Settingsimport 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 = Nonedef _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:continueif 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()])continueexcept Exception:pass# 兜底sentences.extend(_fallback_sentence_split(p))return sentencesdef 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 = 0while i < len(sentences):# 当前块，从第 i 个句子开始chunk = sentences[i]# 计算向前的重叠部分（从 i-1 往回拼）overlap = ''prev = i - 1while prev >= 0 and len(sentences[prev]) + len(overlap) <= overlap_size:overlap = sentences[prev] + ' ' + overlapprev -= 1chunk = overlap + chunk# 再往后拼，直到到达 chunk_sizenext_idx = i + 1while next_idx < len(sentences) and len(sentences[next_idx]) + len(chunk) <= chunk_size:chunk = chunk + ' ' + sentences[next_idx]next_idx += 1chunks.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_fndef 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 = Nonedef _load_embedder():global _tokenizer, _modelif _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_dbself.n_results = n_resultsdef 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 promptdef 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👋 已退出。")breakif user_query.lower() in {"q", "quit", "exit"}:print("👋 已退出。")breakif not user_query:continueprint("\n—— 答复 ——")resp = bot.chat(user_query)print(resp)