RAG 让你的 AI 更聪明

推荐语

RAG技术让大模型突破知识局限，实现实时检索与精准回答，大幅提升AI应用价值。

核心内容：
1. RAG工作原理：向量数据库+实时检索的协同机制
2. 文档处理关键技巧：定长重叠切块法优化检索效果
3. 混合检索策略：关键词与向量搜索的智能结合

杨芳贤

53AI创始人/腾讯云(TVP)最具价值专家

今天想跟大家分享一个让大语言模型（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)

```