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 问答助手，直接看下面的完整代码就能上手。

```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)

```