RAG Refresher Notebook:在 Jupyter Notebook 里从零跑通 RAG 实战链路
如果你正在做 RAG 知识库,却对“文档加载、切分、嵌入、检索、生成、评估”这条链路没有一个全局认知,那这个 RAG Refresher Notebook 就是一个很适合拿来“刷一遍”的项目。它不是一个全新的服务端框架,而是一组面向 RAG 实践的 Notebook 实验,帮助你把 RAG 的每个环节拆开看,从输入 PDF 到最终生成回答,每一步都可视化、可修改、可复现。它的价值不在于跑出一个多强的问答机器人,而在于让你用最短的时间理解 RAG 的工程细节:文档怎么解析、Chunk 怎么切、Embedding 怎么选、检索结果为什么不准、评估指标到底看什么、Retrieval 和 Generation 之间怎么联动。
如果你也是那种“不想一上来就部署 Dify、FastGPT、RAGFlow,想先用 Notebook 把原理跑通”的人,这一篇文章可以直接收藏。下文会按“环境准备 → 文档加载 → Chunk 切分 → 向量化 → 检索 → 生成 → 评估 → 排错”的顺序,给出完整的实操路线和通用代码模板,并重点讲清楚 RAG 知识库中常见的坑:上下文丢失、检索召回不准、指标看不懂、长文档处理失败等。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目类型 | RAG 教学/实验型 Jupyter Notebook 工程,不依赖重型服务端 |
| 运行环境 | Windows / Linux / macOS 均可,Jupyter Notebook 或 JupyterLab |
| 硬件需求 | 纯 CPU 可跑通,推荐 8GB 内存以上;使用本地 Embedding 模型时建议 4GB 以上显存(可选) |
| 主要功能 | 文档加载、文本切分、向量化、向量检索、上下文组装、LLM 生成、RAG 指标评估 |
| 技术组件 | PDF 解析器、文本切分器、Embedding 模型、向量数据库、LLM 接口 |
| 启动方式 | 命令行启动 Jupyter,逐 Cell 执行 |
| 是否支持 API | 支持通过 LLM API 接入生成层,也支持本地 Ollama 接口 |
| 是否支持批量任务 | 支持批量处理文档目录和多 Query 评估 |
| 适合读者 | 刚接触 RAG 的开发者、准备搭建知识库的工程人员、做 RAG 指标分析的算法同学 |
这里要单独说明一句:这个 Notebook 并不是一个开箱即用的、带 Web UI 的 RAG 系统,它更像是一套“操作说明书 + 可运行实验”。你可以在里面替换自己的文档、模型和参数,跑完一遍后,再把经验迁移到正式框架中。
2. RAG 知识库的核心链路:这个 Notebook 在讲什么
很多 RAG 项目看起来功能很多,但核心链路就那么几步:
- 文档加载与解析:把 PDF、Word、Markdown、网页转成纯文本。
- 文本切分 Chunk:把长文本切成合适大小的块。
- Embedding 向量化:把每个 Chunk 转成向量。
- 向量存储:把向量写入向量数据库或内存索引。
- 检索召回:根据用户问题召回 TopK 相关 Chunk。
- 生成回答:把问题和召回内容拼进 Prompt,交给 LLM 生成。
RAG Refresher Notebook 的价值,就是把上面六个步骤全部变成 Notebook 中的可见 Cell。你每一步都能打印中间结果,比如切分后的 Chunk 长什么样,召回结果的相关度有多高,Prompt 最终拼接成了什么样子。
对比直接用 Dify 这类工具,Notebook 方式最大的不同是“可控性”。框架把你的操作封装成了黑盒,而 Notebook 把每一层都暴露给你。这对于排查 RAG 效果不好的问题尤其重要:很多知识库回答不准,问题并不在模型,而在 Chunk 切得不合适或者检索 TopK 取错了。
如果你正在做 RAG 实战,建议先用 Notebook 把链路跑通,然后记录每个环节的参数,再转到一个正式的 RAG 框架中做服务化。
3. 环境准备与前置条件
这一节是按通用环境写的。你本机的 Python 版本、CUDA 状态会影响具体命令,但检查思路是一致的。
3.1 基础软件
RAG Refresher Notebook 基于 Jupyter 运行,所以先准备 Python 和 Jupyter 环境。推荐用 Anaconda 管理环境,因为 RAG 生态里很多包(例如 LangChain、sentence-transformers、faiss-cpu)在 conda 环境中安装比较省心。
# 创建独立的 Python 环境,避免污染系统 Python conda create -n rag_notebook python=3.10 -y conda activate rag_notebook这里使用 Python 3.10 是比较保守的选择。RAG 相关的向量库、PDF 解析库到 2025 年基本都对 3.10 和 3.11 做了适配,不容易踩编译坑。
# 安装 Jupyter Notebook / JupyterLab pip install notebook jupyterlab启动 Jupyter:
jupyter notebook如果你习惯 JupyterLab,运行jupyter lab即可。两者在 Notebook 文件上完全兼容。JupyterLab 在查看 Markdown 标题大纲和多个文件并排时更好用;经典 Notebook 启动更快,界面也更简单。
3.2 Python 依赖库
下面这套依赖覆盖了从文档解析到向量检索再到 LLM 调用的完整链路。具体版本不写死,以你实际安装时的最新稳定版为准。
# 文档解析与文本处理 pip install pypdf pip install python-docx pip install markdown # 向量化与向量检索 pip install sentence-transformers pip install faiss-cpu # LangChain 生态(可选,但推荐) pip install langchain pip install langchain-community # LLM 接入:OpenAI 协议、Ollama 本地模型 pip install openai pip install ollama # 数据处理与评估 pip install numpy pip install pandas如果你的电脑是 NVIDIA 显卡,并且想用 GPU 跑 Embedding 模型,需要额外安装与你的 CUDA 版本匹配的 PyTorch。注意 faiss-gpu 的版本和 CUDA 版本有严格对应关系,如果安装失败,直接用 faiss-cpu 就够了,个人学习场景完全够用。
3.3 模型资源准备
RAG 链路里有两类模型:
第一类是 Embedding 模型,负责把文本转换成向量。可以选择:
- 在线 API:OpenAI Embedding、阿里云 DashScope Embedding 等。
- 本地模型:
BAAI/bge-small-zh-v1.5、BAAI/bge-large-zh-v1.5、text2vec系列等。
本地模型通过sentence-transformers加载,首次运行会从 HuggingFace 或 ModelScope 下载模型。中国大陆网络环境下,优先配置 ModelScope 镜像或使用modelscope下载,速度更稳定。
第二类是生成模型 LLM,负责根据检索结果生成回答。可以选择:
- 在线 API:OpenAI、通义千问、DeepSeek、Kimi 等。
- 本地模型:Ollama 部署的
qwen2.5:7b、llama3.1:8b等。
如果只是验证 RAG 链路,建议先用一个在线 API 或 Ollama 本地模型,不要一开始就追求大模型。
3.4 前置检查清单
| 检查项 | 检查方式 | 达标标准 |
|---|---|---|
| Python 版本 | python --version | 3.10 或 3.11 |
| Jupyter 可访问 | 浏览器打开http://127.0.0.1:8888 | 能看到 Notebook 列表 |
| Embedding 模型可下载 | 执行from sentence_transformers import SentenceTransformer并加载模型 | 不报网络或显存错误 |
| LLM API Key 可调用 | 用requests或openai库发一个请求 | 返回正常文本 |
| 测试文档可读取 | pypdf读取一个测试 PDF | 输出前几行文本 |
4. 文档加载与解析:让 PDF、Word、Markdown 变成可用的纯文本
进入 Notebook 之后,第一步是文档加载。RAG 系统处理最多的文件类型是 PDF,但 PDF 的解析难度被大多数人低估了:不是所有 PDF 都能直接抽出干净文本。
4.1 PDF 文本抽取
pypdf是最轻量的 PDF 文本抽取库,适合文本型 PDF。它会直接提取文本层内容。
from pypdf import PdfReader pdf_path = "./docs/rag_intro.pdf" reader = PdfReader(pdf_path) full_text = [] for page in reader.pages: text = page.extract_text() if text: full_text.append(text) print(f"--- Page {reader.pages.index(page) + 1} ---") print(text[:200]) raw_text = "\n".join(full_text) print(f"总文本长度: {len(raw_text)} 字符")这是最基础的加载方式。它的优点是简单;缺点也很明显:
- 扫描版 PDF 没有文本层,
extract_text()返回空字符串。 - 双栏 PDF 的文本顺序可能被打乱。
- 表格内容会变成无序文本。
- 复杂公式会丢失结构。
如果你遇到上述情况,需要引入 OCR 方案或使用更专业的文档解析引擎。常见方案包括:
paddleocr:适合中文扫描版 PDF,配合paddleocr的版面分析可以输出带位置信息的文本。unstructured库:支持分区解析 PDF,能识别标题、正文、表格。- 商业级解析服务:适合正式项目,但 Notebook 场景可以先用轻量方案。
4.2 Markdown 和 Word 加载
import markdown from bs4 import BeautifulSoup md_text = open("./docs/rag_guide.md", encoding="utf-8").read() html = markdown.markdown(md_text) soup = BeautifulSoup(html, "html.parser") plain_text = soup.get_text() print(plain_text[:500])Word 文件使用python-docx:
from docx import Document doc = Document("./docs/rag_notes.docx") word_text = "\n".join([para.text for para in doc.paragraphs]) print(word_text[:500])4.3 加载后必须做清洗
加载完成不是终点,文本清洗这一步在 RAG 知识库中非常关键。常见清洗操作包括:
import re def clean_text(text: str) -> str: # 合并连续空行 text = re.sub(r"\n{3,}", "\n\n", text) # 去除多余空格 text = re.sub(r"[ \t]+", " ", text) # 去除乱码字符 text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text) # 去除 URL 中的跟踪参数(测试场景可选) text = re.sub(r"\?utm_[^\s]+", "", text) return text.strip() clean_text(raw_text)注意:清洗规则不要做得太激进。比如把换行全部删除,会导致英文单词粘连、中文段落丢失边界。清洗的目标是去除噪音,不是重排文本结构。
5. 文本切分 Chunk:决定 RAG 效果的第一道闸门
很多人在做 RAG 时发现检索结果不准,第一反应是换 Embedding 模型,但很多时候问题出在 Chunk 切分上。
5.1 为什么 Chunk 大小很关键
Chunk 太大会导致向量表示过于笼统,检索时召回结果主题漂移;Chunk 太小会丢失上下文,生成阶段拿不到足够的背景信息。Chunk 的合理大小取决于你的文档类型和检索场景。
一般规律:
| Chunk 类型 | 适合场景 | 大致大小 |
|---|---|---|
| 小 Chunk | 问答型、关键词相关性高 | 200-400 字符 |
| 中 Chunk | 百科、说明文 | 500-800 字符 |
| 大 Chunk | 长上下文分析 | 1000-2000 字符 |
但这不是绝对的。中文和英文的“字符”概念不同,标点密度也不同。更靠谱的做法是:切完 Chunk 后随机抽 20 个 Chunk 人眼检查,看语义是否完整。
5.2 使用 LangChain 的 RecursiveCharacterTextSplitter
LangChain 的RecursiveCharacterTextSplitter是目前最实用的切分器之一。它的思路是递归地按分隔符列表切分,先按段落、再按句子、最后按字符。
from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""], length_function=len, ) chunks = splitter.split_text(cleaned_text) print(f"切分得到 {len(chunks)} 个 Chunk") for i, chunk in enumerate(chunks[:3]): print(f"\n=== Chunk {i} ===") print(chunk[:300])这里的重点参数是chunk_overlap。它让相邻 Chunk 之间有重叠,避免一句话被从中间截断,导致语义不完整。重叠值一般设置为 Chunk 大小的 10% 到 20%。
5.3 按语义结构切分
如果你的文档有明确的 Markdown 标题或 PDF 章节结构,强烈建议先按结构切分,再按长度二次切分。这样能保证每个 Chunk 都属于同一个主题。
from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "H1"), ("##", "H2"), ("###", "H3"), ] md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) md_splits = md_splitter.split_text(md_text) for split in md_splits[:3]: print(split.metadata) print(split.page_content[:100]) print("---")这种做法特别适合做精准 RAG。你可以在 Metadata 中保留章节路径,后续检索时可以对章节进行过滤。
5.4 Chunk 质量的检查方法
切分后不要急着向量化,先做人眼抽检:
- 看是否有 Chunk 在句子中间截断。
- 看是否有一个 Chunk 包含多个无关主题。
- 看 Chunk 长度分布是否极端。
- 看 Metadata 是否能追溯到原文档位置。
如果这四项都 OK,再进行下一步。这一步在 Notebook 里可以做成一个统计 Cell:
import matplotlib.pyplot as plt lengths = [len(c) for c in chunks] plt.hist(lengths, bins=20) plt.title("Chunk Length Distribution") plt.xlabel("Length") plt.ylabel("Count") plt.show() print(f"最短: {min(lengths)} 字符") print(f"最长: {max(lengths)} 字符") print(f"平均: {sum(lengths) / len(lengths):.1f} 字符")6. Embedding 向量化与向量库构建
Chunk 切好之后,就要把每一段文本转换成向量。这是 RAG 知识库检索能力的基础。
6.1 加载 Embedding 模型
这里以bge-small-zh-v1.5为例。它是中英文通用的轻量级 Embedding 模型,在 CPU 上也能跑,适合 Notebook 学习场景。
from sentence_transformers import SentenceTransformer embedding_model = SentenceTransformer("BAAI/bge-small-zh-v1.5") print(f"模型维度: {embedding_model.get_sentence_embedding_dimension()}")如果下载速度慢,可以配置 ModelScope 镜像:
# 使用 ModelScope 下载 bge 模型 from modelscope import snapshot_download model_dir = snapshot_download("AI-ModelScope/bge-small-zh-v1.5") embedding_model = SentenceTransformer(model_dir)这里要注意:不同 Embedding 模型的向量维度不同,词表不同,语义空间也不同。如果后续要切换模型,需要重新向量化整个语料库,不能混用。
6.2 向量化 Chunk
import numpy as np chunk_texts = [c for c in chunks] batch_embeddings = embedding_model.encode( chunk_texts, batch_size=16, normalize_embeddings=True, show_progress_bar=True, ) embeddings = np.array(batch_embeddings) print(f"向量矩阵形状: {embeddings.shape}")normalize_embeddings=True会让所有向量归一化到单位长度,这样后续用余弦相似度计算时,内积就等于余弦相似度,性能更高。
6.3 构建 Faiss 索引
Faiss 是 Meta 开源的向量检索库,支持 CPU 和 GPU。这里使用IndexFlatIP,这是最精确的暴力检索索引,适合数据量在百万级以下的学习场景。
import faiss dim = embeddings.shape[1] index = faiss.IndexFlatIP(dim) index.add(embeddings.astype("float32")) print(f"Faiss 索引中向量数量: {index.ntotal}")如果你的数据量很大,比如超过十万条 Chunk,可以换成IndexIVFFlat或IndexHNSWFlat来降低检索延迟。但在 Notebook 学习阶段,IndexFlatIP足够准确,也更容易理解。
6.4 保存索引与 Chunk 映射
向量库和原文必须配套保存。索引只保存向量,无法反查文本,所以需要单独维护 Chunk 列表。
import pickle # 保存索引 faiss.write_index(index, "./outputs/faiss_index.bin") # 保存 Chunk 文本和元数据 with open("./outputs/chunks.pkl", "wb") as f: pickle.dump({"chunks": chunk_texts}, f) print("索引与 Chunk 已保存")7. 检索测试:先不要急着上 LLM
RAG 实战最容易犯的错误是跳过检索评估,直接把 Chunk 交给 LLM 生成答案。如果检索召回的内容本身就不相关,LLM 再强也只能“一本正经地胡说八道”。
7.1 检索单条 Query
query = "什么是 RAG?" query_vec = embedding_model.encode([query], normalize_embeddings=True) query_vec = np.array(query_vec).astype("float32") top_k = 3 scores, indices = index.search(query_vec, top_k) for rank, (score, idx) in enumerate(zip(scores[0], indices[0])): print(f"\nRank {rank + 1} | Score: {score:.4f}") print(chunk_texts[idx][:300])这一步关键看两个东西:
- 分数是否合理。归一化向量后余弦相似度一般在 0 到 1 之间,如果普遍低于 0.5,说明 query 和语料语义差距较大。
- 召回内容是否真的相关。不要只看分数,要读内容。如果召回结果完全不相关,后面生成效果一定差。
7.2 检索失败的常见原因
检索结果不准,按优先级排查:
- Chunk 切得太大或太小,语义主题不集中。
- Embedding 模型不适合当前语言的文档。比如英文文档用了一个弱中文模型,效果会打折。
- Query 本身太短或太模糊,例如“帮我写个报告”这种 query 很难召回精准结果。
- 没有做文本清洗,噪声文本干扰了向量表示。
- 文档主题交叉严重,一个 Chunk 里包含多个主题,检索时只能命中一个主题。
7.3 检索优化小技巧
在 Notebook 里,你可以快速实验以下做法:
- 对 Query 做改写,添加业务上下文。
- 检索 TopK 后做一次重排序,用
bge-reranker或 CrossEncoder 对召回结果打分。 - 在向量检索之前加一层关键字过滤,基于 Metadata 的章节信息或标签缩小范围。
- 对 Chunk 内容做摘要后再向量化,保留原文用于生成。
重排模型的接入方式如下:
from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-base") pairs = [[query, chunk_texts[idx]] for idx in indices[0]] rerank_scores = reranker.predict(pairs) # 按重排分数重新排序 sorted_indices = [ idx for _, idx in sorted( zip(rerank_scores, indices[0]), reverse=True ) ]重排能显著提高检索精度。如果你的 RAG 应用对答案准确性要求高,不要省略这一步。
8. 生成链路:把检索结果交给 LLM
检索通过后,下一步是把召回结果和用户问题拼接到 Prompt 中,送到 LLM 生成答案。
8.1 使用 OpenAI 兼容接口
下面代码兼容 OpenAI、DeepSeek、通义千问、Ollama 等所有 OpenAI 兼容 API。
from openai import OpenAI client = OpenAI( base_url="http://127.0.0.1:11434/v1", # Ollama 为例 api_key="ollama", # Ollama 不校验 key,随便填 ) def build_rag_prompt(query: str, contexts: list[str]) -> str: context_text = "\n\n---\n\n".join(contexts) prompt = f"""你是一个知识库问答助手。请根据提供的资料回答问题。 如果资料中没有相关信息,请直接说明“资料中未找到相关信息”,不要编造。 资料: {context_text} 问题: {query} 回答:""" return prompt prompt = build_rag_prompt(query, [chunk_texts[idx] for idx in indices[0]]) response = client.chat.completions.create( model="qwen2.5:7b", messages=[ {"role": "system", "content": "你是严谨的知识库问答助手。"}, {"role": "user", "content": prompt}, ], temperature=0.3, max_tokens=1024, ) print(response.choices[0].message.content)如果你用在线 API,把base_url和api_key替换成对应服务的配置即可。注意:永远不要把 API Key 硬编码到 Notebook 里提交到公开仓库。建议用环境变量或.env文件管理。
8.2 Prompt 模板设计的关键
RAG 的 Prompt 模板比普通对话要严格。核心要求是:
- 明确告诉模型只能使用资料中的信息。
- 当资料中没有答案时,允许模型说“不知道”。
- 资料之间用清晰的分隔符隔开。
- 问题放在最后,避免长上下文干扰模型注意力。
如果你发现生成结果失真,优先检查 Prompt 是否设置了边界,而不是立刻换更大参数的模型。
8.3 多轮对话场景
如果 RAG 应用需要多轮对话,可以把历史对话拼入 Prompt,并对每一轮引用到的文档来源做记录。但在 Notebook 实验阶段,建议先跑通单轮问答,再扩展多轮。多轮对话会让 Prompt 长度快速膨胀,增加成本并且可能超出上下文窗口。
9. RAG 评估:用指标量化知识库效果
很多人问 RAG 知识库指标有哪些。在 Notebook 中,我们可以做最核心的三类指标评估:
| 指标 | 含义 | 评估对象 |
|---|---|---|
| 召回相关度 | 检索到的文档与问题是否相关 | 检索模块 |
| 生成忠实度 | 答案是否忠于资料,是否幻觉 | 生成模块 |
| 答案有用性 | 答案是否完整、准确 | 整体 |
9.1 检索评估:Recall@K
如果有一套带标准答案的测试集,可以计算 Recall@K:即在 TopK 召回结果中,包含正确文档的比例。
def recall_at_k(relevant_doc_ids, retrieved_doc_ids, k): retrieved_set = set(retrieved_doc_ids[:k]) relevant_set = set(relevant_doc_ids) if not relevant_set: return 0 return len(retrieved_set & relevant_set) / len(relevant_set) # 示例:对于一个 Query,假设正确文档 id 是 [2, 5],Top3 召回 [2, 8, 9] recall = recall_at_k([2, 5], [2, 8, 9], k=3) print(f"Recall@3: {recall:.2f}")9.2 生成评估:忠实度
忠实度评估在纯 Notebook 中可以做人工打标,也可以使用 LLM-as-Judge 的方式:让一个强 LLM 阅读“资料 + 答案”,判断答案是否基于资料。
judge_prompt = f"""请判断下面的回答是否完全基于提供的资料,不包含资料外的信息。 只输出“忠实”或“不忠实”,不要输出其他内容。 资料: {context_text} 回答: {answer}""" judge_response = client.chat.completions.create( model="qwen2.5:7b", messages=[{"role": "user", "content": judge_prompt}], temperature=0, ) print(judge_response.choices[0].message.content)这种评估方式有主观性,但可以快速发现严重的幻觉问题。
9.3 构建测试集
评估 RAG 必须有一套测试集。最简单的构造方式:
- 从文档中选择 20-50 个重要知识点。
- 每个知识点写一个自然语言问题。
- 记录对应答案所在的原文位置。
- 将问题和原文片段保存为 JSON 文件。
{ "test_cases": [ { "question": "RAG 中 Chunk 大小一般设置为多少合适?", "reference_ids": [3, 4], "expected_answer": "取决于文档类型,一般 500 到 800 字符较为常见" }, { "question": "RAG 检索不准时第一步应该检查什么?", "reference_ids": [7], "expected_answer": "先检查切分后的 Chunk 质量,确认语义是否完整" } ] }测试集建好后,用循环批量执行检索和生成,然后计算整体指标。这其实就是最简化的 RAG 批量任务。
9.4 指标如何理解
很多新手看到指标下降就慌。这里给一个经验判断:
- 检索指标低,说明召回链路有问题,先优化 Chunk 和 Embedding。
- 生成忠实度低,说明 Prompt 边界不够强,或 LLM 本身太激进。
- 答案有用性低,说明可能是生成模型能力不足,或者上下文窗口不够。
指标只是一个信号,关键是能定位到具体环节。
10. 接口 API 与批量任务扩展
Notebook 本身不是服务,但你可以把验证通过的 RAG 链路封装成一个可调用的函数,再通过 FastAPI 暴露成 API。这对后续集成到小工具或业务系统有帮助。
# 将一个完整的 RAG 查询封装成函数 def rag_query(query: str, top_k: int = 3, use_rerank: bool = True): # 1. Query 向量化 query_vec = embedding_model.encode([query], normalize_embeddings=True) query_vec = np.array(query_vec).astype("float32") # 2. 向量检索 scores, indices = index.search(query_vec, top_k) retrieved = [chunk_texts[idx] for idx in indices[0]] # 3. 可选:重排 if use_rerank: pairs = [[query, doc] for doc in retrieved] rerank_scores = reranker.predict(pairs) retrieved = [doc for _, doc in sorted(zip(rerank_scores, retrieved), reverse=True)] # 4. 组装 Prompt 并调用 LLM prompt = build_rag_prompt(query, retrieved) response = client.chat.completions.create( model="qwen2.5:7b", messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=1024, ) answer = response.choices[0].message.content # 5. 返回结果和召回片段,便于调试 return {"answer": answer, "references": retrieved}用 FastAPI 暴露成接口:
# api_server.py from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): query: str top_k: int = 3 @app.post("/api/rag") def rag_endpoint(req: QueryRequest): result = rag_query(req.query, req.top_k) return result启动接口服务:
uvicorn api_server:app --host 127.0.0.1 --port 8000调用测试:
curl -X POST http://127.0.0.1:8000/api/rag \ -H "Content-Type: application/json" \ -d '{"query": "什么是RAG知识库指标?"}'批量任务方面,在 Notebook 中可以直接循环遍历测试集。注意两点:一是控制并发,避免 LLM API 限流;二是保存中间结果,方便失败重跑。建议每处理一条写一行日志或保存到 JSON,失败时记下 query,不中断整体流程。
11. 资源占用与性能观察
RAG Refresher Notebook 的负载主要集中在三处:PDF 解析、Embedding 编码、LLM 推理。
PDF 解析是纯 CPU 任务,大 PDF 可能耗时几十秒。如果文本量很大,建议先对 PDF 进行页数抽样,不要一次性全量解析。
Embedding 编码部分,使用bge-small-zh-v1.5这类模型做 CPU 推理时,500 个 Chunk 大约需要十几秒到几十秒,取决于 CPU 性能和文本长度。如果显存不足或不想占用 GPU,CPU 完全可以跑完,只是速度慢一些。观察显存占用可以用nvidia-smi查看,也可以用下面的代码获取 PyTorch 显存占用值:
import torch if torch.cuda.is_available(): print(f"显存占用: {torch.cuda.memory_allocated() / 1024**3:.2f} GB") else: print("当前使用 CPU 推理")LLM 推理的负载差异巨大。如果使用 Ollama 本地模型,qwen2.5:7b在 CPU 上单轮回答可能需要 10 秒以上,GPU 上通常 1 到 3 秒。在线 API 基本不受本机性能影响,但受网络延迟和限流影响。
为了降低资源占用,建议:
- Embedding 采用 batch 方式编码,不要一条一条循环。
- Chunk 长度不要设置过大,文本越长,Embedding 和 LLM 耗时越长。
- 测试阶段把 LLM 的
max_tokens调低到 512。 - 不需要重排时,跳过重排模型加载,省出内存。
12. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
pip install报错 | Python 版本不兼容或依赖冲突 | 查看错误堆栈,确认是编译错误还是版本冲突 | 升级到 Python 3.10/3.11,或使用 conda 安装 |
| 模型下载失败 | HuggingFace 网络不稳定 | 检查是否超时 | 改用 ModelScope 镜像下载 |
| NVIDIA 相关报错 | 显存不足或 CUDA 版本不匹配 | 执行nvidia-smi查看驱动和显存 | 改用 CPU 版 PyTorch 或调小 batch_size |
| PDF 解析出空文本 | 扫描版 PDF 无文本层 | 抽取第一页看是否有文字 | 改用 OCR 方案(PaddleOCR) |
| Chunk 切分后语义断裂 | separators未覆盖标点或chunk_overlap过小 | 打印相邻 Chunk 拼接处观察 | 调整分隔符顺序,增大 overlap |
| 检索结果完全不相关 | Embedding 模型选型不对或 Chunk 过大 | 打印 Chunk 内容检查 | 换中文 Embedding 模型,或调整 Chunk 大小 |
| 检索分数普遍偏低 | Query 与语料风格差异大 | 观察分数分布 | 对 Query 做改写或加前缀优化 |
| LLM 回答出幻觉 | Prompt 边界不够强 | 检查回答中是否出现资料外信息 | 强化 Prompt 约束,加入“没有找到就直说” |
| API 调用超时 | 网络问题或模型推理太慢 | 查看接口响应时间 | 调大超时时间,或切换更小的模型 |
| 批量任务中途失败 | 单一 query 触发异常 | 打印回溯信息 | 用try-except记录失败 query,继续执行 |
| 端口被占用 | 之前有 Jupyter 实例未关闭 | `netstat -ano | findstr 8888` |
13. 最佳实践与使用建议
从 Notebook 转向正式 RAG 应用时,有几条经验值得记住。
第一,先小后大。第一次跑通链路使用 5 个以内的文档、TopK 设置为 2,不要一次性灌入整个知识库。先把链路跑通,再逐步加数据。
第二,保留一份最小可运行版本。不要让 Notebook 变成大杂烩。建议拆成多个 Notebook:01_parse.ipynb、02_chunk.ipynb、03_embed.ipynb、04_retrieve.ipynb、05_generate.ipynb、06_evaluate.ipynb。这样每一步的产物都能持久化保存,不需要每次从头跑。
第三,目录管理要规范:
docs/ raw/ # 原始文档 processed/ # 清洗后文本 outputs/ chunks.pkl # 切分结果 faiss_index.bin # 向量索引 eval_results/ # 评估结果 cache/ models/ # 本地模型缓存第四,接口服务必须限制访问范围。如果你把 FastAPI 服务暴露到局域网,要加 API Key 校验,不能让任意客户端调用你的模型接口产生费用。
第五,涉及版权和隐私必须谨慎。RAG 知识库如果处理的是内部资料、受版权保护的书籍、他人文档或包含个人信息的文件,只能用于合法授权的测试环境和个人学习。不要用未经授权的资料构建可公开访问的知识库服务;不要上传包含敏感身份信息的文件到在线 API 服务。使用声音、人脸、图像类数据时,同样要确认授权边界。
第六,发布或商用前做效果复核。RAG 系统在实验环境表现好,不代表在真实流量下准确率达标。上线前至少准备 50 条真实用户问题做压测,观察检索召回指标和人工评判的有用性指标。
14. 总结与下一步
RAG Refresher Notebook 的价值不在于替代 Dify、FastGPT 这类框架,而在于把 RAG 的每个环节从黑盒变成白盒。你可以亲手切 Chunk,亲手看向量检索分数,亲手调 Prompt 模板,最终理解 RAG 的“准”是怎么来的。
如果你打算开始做 RAG 实战,优先验证三件事:
- 你选的 PDF 能否被正常解析成文本。
- 你的 Chunk 切分质量是否通过人眼抽检。
- 你的检索结果在不用 LLM 的情况下是否已经看起来相关。
这三步通过之后,再接入生成模型。最容易踩的坑是跳过检索评估直接看最终回答,导致你无法定位是检索问题还是生成问题。
下一步可以做三件事:第一,把你的业务文档导入这个 Notebook 做一轮端到端测试;第二,建立 20 到 50 条测试集并跑一遍评估,记录基线和 badcase;第三,把验证通过的链路封装成 FastAPI 服务,再接一个前端或接入飞书、钉钉机器人,形成可用的 RAG 知识库应用。
这套 Notebook 经验沉淀下来,之后无论迁移到 Dify、RAGFlow、LangChain 还是自研服务,你都能带着清晰的判断力去调整参数,而不是靠感觉调来调去。建议收藏备用,回头跑 RAG 项目时对着这份链路一项项检查。