本地免费知识库搭建：Ollama+FAISS+MoreLogic RAG实战

1. 为什么我要自己搭一个知识库

1.1 从“收藏夹吃灰”说起

我电脑里有个文件夹叫“待读”，里面塞了大概四百多个网页存档、PDF 和 Markdown 笔记。每次想找某个技术细节，比如“FAISS 的 IndexIVFFlat 怎么调 nprobe”，我得先回忆这东西是去年几月看的，然后翻聊天记录、翻浏览器书签、翻笔记软件，最后大概率还是重新搜一遍。这种体验就像你家里有个巨大的仓库，但所有东西都堆在地上，没有货架也没有标签，找一件东西得把整个仓库翻一遍。

后来我试过几个在线笔记和知识管理工具，问题也很明显：要么数据在别人服务器上，要么免费版有各种限制，要么搜索能力太弱，只能匹配关键词，没法理解语义。比如我搜“怎么让模型记住我上传的文档”，它只会去找包含“模型”“文档”这些词的文章，而不会理解我其实想问的是 RAG 的检索增强生成流程。

所以我就想，能不能在自己电脑上搭一个完全本地的、免费的、支持语义搜索的知识库。核心需求就三条：数据不出本机、不花钱、能理解自然语言提问。折腾了一圈，最后落到了 MoreLogic RAG 个人免费版这个方案上，配合 Ollama 做本地模型推理，FAISS 做向量检索，Python 做胶水层。整套东西跑起来之后，我那个“待读”文件夹终于变成了一个能对话的知识库。

1.2 这套方案到底适合谁

先说清楚，这不是那种企业级、高并发、多租户的知识库系统。它更适合下面这几类人：

• 个人开发者：手里有一堆技术文档、API 说明、项目笔记，想快速检索。
• 学生和研究者：需要管理论文、实验记录、参考文献，并且想用自然语言提问。
• 小团队内部使用：三五个人共享一批资料，不想把数据传到第三方平台。
• 对数据隐私敏感的人：比如律师、医生、财务人员，文档不能离开本地环境。

如果你需要的是支持上千人同时在线、有复杂权限管理、能对接企业微信和钉钉的那种系统，那这套方案不适合你，你应该去看 Dify 或者 FastGPT 这类更重的平台。但如果你只是想要一个“自己用的、能搜能问的文档库”，那这套组合拳的性价比非常高。

1.3 整体技术选型背后的逻辑

这套方案的核心链路其实不复杂：文档进来，切块，转向量，存进 FAISS，提问时把问题也转向量，去 FAISS 里找最相似的块，再把问题和这些块一起塞给 Ollama 里跑的大模型，让它生成回答。

为什么选 Ollama 而不是直接调云端 API？因为免费版的核心诉求就是“不花钱”和“数据不出门”。Ollama 可以在本地跑 Qwen、Llama、Gemma 这些开源模型，虽然效果比不上 GPT-4，但对于“从我的文档里找答案”这种任务，7B 到 14B 的模型已经够用了。而且 Ollama 的安装和管理非常简单，一条ollama run qwen2.5:7b就能跑起来。

为什么选 FAISS 而不是 Chroma 或 Milvus？因为 FAISS 是 Facebook 开源的向量检索库，纯 Python 调用，不需要额外起服务，对于个人知识库这种数据量（几千到几万条 chunk）来说，FAISS 的本地索引完全够用，而且速度极快。Chroma 虽然更方便，但它的持久化机制有时候会出一些奇怪的问题；Milvus 则是为大规模设计的，个人用属于杀鸡用牛刀。

MoreLogic RAG 个人免费版在这里的角色，其实是把上面这些组件串起来的“脚手架”。它提供了文档解析、切块、向量化、检索、生成这一整套流程的封装，让你不用从零写代码。但它的免费版功能有限，所以我在实际使用中做了一些替换和补充，下面会详细说。

2. 环境准备与核心组件安装

2.1 Python 环境的干净搭建

我强烈建议不要用系统自带的 Python，也不要在一个已经装了几百个包的环境里折腾。用 conda 或者 venv 建一个独立环境，这是血泪教训。我之前图省事直接在 base 环境里装，结果 FAISS 和某个科学计算库的版本冲突，排查了一下午。

# 用 conda 创建环境，指定 Python 3.10 conda create -n myrag python=3.10 conda activate myrag # 或者用 venv python -m venv myrag_env source myrag_env/bin/activate # Linux/Mac # myrag_env\Scripts\activate # Windows

为什么选 Python 3.10？因为 3.11 和 3.12 在某些深度学习库的兼容性上还有问题，3.10 是目前最稳的版本。装完之后，先升级 pip，然后装核心依赖：

pip install --upgrade pip pip install faiss-cpu # 如果你有 NVIDIA 显卡并且想用 GPU 加速，可以装 faiss-gpu pip install sentence-transformers pip install ollama pip install pypdf markdown beautifulsoup4 pip install numpy pandas tqdm

这里解释一下每个包的作用。faiss-cpu是向量检索的核心，它负责存储和快速查找相似向量。sentence-transformers用来把文本转成向量，我后面会讲为什么不用 OpenAI 的 embedding API。ollama是 Python 客户端，用来和本地 Ollama 服务通信。pypdf和markdown用来解析不同格式的文档。tqdm用来显示进度条，处理大量文档时没有进度条你会很焦虑。

注意：如果你在 Windows 上装 faiss-cpu 报错，大概率是缺少 Visual C++ 运行库，去微软官网下载最新的 VC++ redistributable 装上就行。另外，faiss-cpu 在 Python 3.10 上的 wheel 包很全，基本不会遇到编译问题。

2.2 Ollama 的安装与模型选择

Ollama 的安装本身很简单，官网下载对应系统的安装包，一路下一步就行。但国内用户会遇到两个问题：下载慢和模型拉取慢。

下载慢的问题，可以找国内的镜像源。Ollama 的模型仓库是放在 registry.ollama.ai 上的，国内访问确实不稳定。我的做法是先用迅雷或者 IDM 把模型文件下载下来，然后手动导入。具体操作是：在能正常访问的机器上执行ollama pull qwen2.5:7b，然后去~/.ollama/models目录下把对应的 blob 文件拷贝出来，放到目标机器的同样目录下。Ollama 的模型存储结构是models/blobs/sha256-xxx和models/manifests/registry.ollama.ai/library/qwen2.5/7b，把这两部分都拷过去就行。

模型选择方面，我试过好几个。Qwen2.5:7b 在中文理解和指令遵循上表现最好，而且对硬件要求不高，16GB 内存的机器就能跑。Llama3.1:8b 英文能力强，但中文问答有时候会答非所问。Gemma2:9b 也不错，但推理速度比 Qwen 慢一些。如果你机器配置高，可以上 Qwen2.5:14b，效果明显更好。如果只有 8GB 内存，那就用 Qwen2.5:3b 或者 Phi3:mini，虽然能力有限，但做简单的文档检索问答也够用。

# 拉取模型 ollama pull qwen2.5:7b # 测试模型是否正常 ollama run qwen2.5:7b "你好，请用一句话介绍你自己"

如果遇到ollama run qwen3.5:2b error: 500 internal server error: llama-server process这种报错，通常是模型文件损坏或者内存不足。先检查内存占用，然后尝试重新拉取模型。还有一个常见原因是 Ollama 的版本太老，去官网下载最新版覆盖安装即可。

2.3 FAISS 的索引类型选择

FAISS 提供了多种索引类型，选错了会导致检索速度慢或者精度下降。对于个人知识库这种规模，我推荐用IndexFlatL2或者IndexIVFFlat。

IndexFlatL2是最简单的暴力检索，它会把查询向量和库里所有向量逐一计算 L2 距离，然后返回最近的 k 个。优点是精度 100%，缺点是数据量大了之后速度线性下降。对于一万条以下的 chunk，用这个完全没问题，检索时间在毫秒级。

IndexIVFFlat是倒排索引，它先把向量聚类成 nlist 个簇，检索时只查最近的 nprobe 个簇。这样速度更快，但会损失一点精度。当你的 chunk 数量超过五万时，建议用这个。nlist 一般设为sqrt(N)，nprobe 设为nlist/10左右。

import faiss import numpy as np # 假设向量维度是 768 dimension = 768 # 暴力检索索引 index_flat = faiss.IndexFlatL2(dimension) # 倒排索引 nlist = 100 # 聚类中心数量 quantizer = faiss.IndexFlatL2(dimension) index_ivf = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_L2) # 需要先训练 index_ivf.train(vectors) index_ivf.add(vectors) index_ivf.nprobe = 10 # 检索时查 10 个簇

我实测下来，对于个人知识库，直接用IndexFlatL2最省心。等你真的攒了十万条以上的 chunk，再考虑换 IVF。而且 FAISS 支持保存和加载索引，faiss.write_index(index, "my_index.faiss")和faiss.read_index("my_index.faiss")，这样每次启动不用重新计算所有向量。

3. 文档处理与向量化实操

3.1 文档解析的坑与解决方案

知识库的第一步是把各种格式的文档变成纯文本。听起来简单，但实际操作中坑很多。PDF 里的表格、扫描件里的图片、Markdown 里的代码块，处理不好都会丢失信息。

对于 PDF，我用pypdf做基础提取，但它对复杂排版的 PDF 支持很差。如果 PDF 是扫描件，那pypdf提取出来全是空白。这时候需要用 OCR，我试过pytesseract配合pdf2image，效果还行，但速度慢。对于个人知识库，我建议先把扫描件用其他工具转成文字版 PDF，再入库。

对于 Markdown 和纯文本，直接读取就行。对于 Word 文档，用python-docx。对于网页存档，用BeautifulSoup提取正文，去掉导航栏和广告。

from pypdf import PdfReader import markdown from bs4 import BeautifulSoup def parse_pdf(file_path): reader = PdfReader(file_path) text = "" for page in reader.pages: text += page.extract_text() + "\n" return text def parse_markdown(file_path): with open(file_path, 'r', encoding='utf-8') as f: md_text = f.read() html = markdown.markdown(md_text) soup = BeautifulSoup(html, 'html.parser') return soup.get_text() def parse_html(file_path): with open(file_path, 'r', encoding='utf-8') as f: html = f.read() soup = BeautifulSoup(html, 'html.parser') # 去掉 script 和 style for tag in soup(['script', 'style', 'nav', 'footer', 'header']): tag.decompose() return soup.get_text()

注意：解析 PDF 时一定要加异常处理。有些 PDF 是加密的，有些是损坏的，直接抛异常会让整个批处理中断。用 try-except 包起来，记录失败的文件名，后面单独处理。

3.2 文本切块策略：不是越细越好

切块是 RAG 里最容易被忽视但影响最大的环节。切得太粗，检索出来的块包含太多无关信息，模型容易被干扰；切得太细，一个完整的语义单元被拆散，检索出来的是碎片，模型看不懂。

我的经验是：按语义边界切，而不是按固定字数切。具体来说，优先按段落切，如果段落太长（超过 800 字），再按句子切。句子边界用中文的句号、问号、感叹号，以及英文的句号、问号、感叹号来识别。

import re def split_text(text, max_chunk_size=500, overlap=50): # 先按段落分 paragraphs = re.split(r'\n\s*\n', text) chunks = [] current_chunk = "" for para in paragraphs: para = para.strip() if not para: continue # 如果单个段落就超过 max_chunk_size，按句子切 if len(para) > max_chunk_size: sentences = re.split(r'(?<=[。！？.!?])\s*', para) for sent in sentences: if len(current_chunk) + len(sent) <= max_chunk_size: current_chunk += sent else: if current_chunk: chunks.append(current_chunk) current_chunk = sent else: if len(current_chunk) + len(para) <= max_chunk_size: current_chunk += para + "\n" else: if current_chunk: chunks.append(current_chunk) current_chunk = para + "\n" if current_chunk: chunks.append(current_chunk) # 加 overlap final_chunks = [] for i, chunk in enumerate(chunks): if i > 0: prev_tail = chunks[i-1][-overlap:] chunk = prev_tail + chunk final_chunks.append(chunk) return final_chunks

overlap 的作用是防止关键信息刚好落在切分边界上。比如一句话被切成两半，前半句在 chunk A，后半句在 chunk B，检索时可能只召回 A，模型就看不到完整信息。加 50 到 100 字的 overlap 能缓解这个问题。

3.3 向量化模型的选择与本地部署

向量化就是把文本变成一串数字（向量），让语义相似的文本在向量空间里距离更近。OpenAI 的 text-embedding-ada-002 效果很好，但要花钱，而且数据要传到云端。所以我用sentence-transformers里的开源模型。

中文场景我推荐BAAI/bge-large-zh-v1.5或者BAAI/bge-base-zh-v1.5。BGE 系列在中文语义相似度任务上表现很好，而且模型大小适中。large 版是 1024 维，base 版是 768 维。如果机器内存够，用 large；否则用 base，效果差距不大。

from sentence_transformers import SentenceTransformer # 加载模型，第一次会自动下载 model = SentenceTransformer('BAAI/bge-base-zh-v1.5') # 编码 texts = ["这是第一段文本", "这是第二段文本"] embeddings = model.encode(texts, normalize_embeddings=True) print(embeddings.shape) # (2, 768)

normalize_embeddings=True很重要，它把向量归一化到单位长度，这样用内积计算相似度就等价于余弦相似度。FAISS 的IndexFlatIP就是内积索引，配合归一化向量使用效果最好。

注意：第一次运行SentenceTransformer会从 HuggingFace 下载模型，国内可能很慢。可以设置环境变量HF_ENDPOINT=https://hf-mirror.com来加速。或者提前用huggingface-cli download把模型下载到本地，然后加载时指定本地路径。

4. 检索与生成链路的完整实现

4.1 构建 FAISS 索引并持久化

有了向量之后，就可以建索引了。我把所有 chunk 的向量存成一个 numpy 数组，然后加到 FAISS 索引里。同时用一个列表保存每个 chunk 的原始文本和元数据（来源文件、页码等），这样检索到向量后能找回对应的文本。

import faiss import numpy as np import pickle import os class VectorStore: def __init__(self, dimension=768): self.dimension = dimension self.index = faiss.IndexFlatIP(dimension) # 内积索引 self.chunks = [] # 存储原始文本 self.metadata = [] # 存储元数据 def add(self, embeddings, chunks, metadata_list): # embeddings 是 numpy 数组，shape (N, dimension) self.index.add(embeddings.astype('float32')) self.chunks.extend(chunks) self.metadata.extend(metadata_list) def search(self, query_embedding, top_k=5): query_embedding = query_embedding.astype('float32').reshape(1, -1) distances, indices = self.index.search(query_embedding, top_k) results = [] for dist, idx in zip(distances[0], indices[0]): if idx == -1: continue results.append({ 'score': float(dist), 'text': self.chunks[idx], 'metadata': self.metadata[idx] }) return results def save(self, path): faiss.write_index(self.index, os.path.join(path, "index.faiss")) with open(os.path.join(path, "chunks.pkl"), 'wb') as f: pickle.dump({'chunks': self.chunks, 'metadata': self.metadata}, f) def load(self, path): self.index = faiss.read_index(os.path.join(path, "index.faiss")) with open(os.path.join(path, "chunks.pkl"), 'rb') as f: data = pickle.load(f) self.chunks = data['chunks'] self.metadata = data['metadata']

用IndexFlatIP而不是IndexFlatL2，是因为归一化后的向量用内积计算相似度更直接，分数越高越相似。如果用 L2 距离，分数越低越相似，容易搞混。

4.2 检索策略：不只是向量相似度

单纯的向量检索有时候会漏掉关键词精确匹配的情况。比如你搜“FAISS IndexIVFFlat”，向量检索可能返回一堆讲“向量索引”的通用文章，但真正包含这个具体术语的文档反而排在后面。这时候需要混合检索：向量检索 + 关键词检索。

我的做法是用rank_bm25做一个简单的 BM25 关键词检索，然后把两路结果融合。融合算法用 Reciprocal Rank Fusion（RRF），它对不同检索系统的分数尺度不敏感，效果很稳。

from rank_bm25 import BM25Okapi import jieba class HybridRetriever: def __init__(self, vector_store): self.vector_store = vector_store # 构建 BM25 索引 tokenized_chunks = [list(jieba.cut(chunk)) for chunk in vector_store.chunks] self.bm25 = BM25Okapi(tokenized_chunks) def search(self, query, query_embedding, top_k=5): # 向量检索 vector_results = self.vector_store.search(query_embedding, top_k * 2) # BM25 检索 tokenized_query = list(jieba.cut(query)) bm25_scores = self.bm25.get_scores(tokenized_query) bm25_top_indices = np.argsort(bm25_scores)[::-1][:top_k * 2] # RRF 融合 rrf_scores = {} for rank, result in enumerate(vector_results): idx = self.vector_store.chunks.index(result['text']) rrf_scores[idx] = rrf_scores.get(idx, 0) + 1 / (60 + rank + 1) for rank, idx in enumerate(bm25_top_indices): rrf_scores[idx] = rrf_scores.get(idx, 0) + 1 / (60 + rank + 1) # 排序取 top_k sorted_indices = sorted(rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True)[:top_k] results = [] for idx in sorted_indices: results.append({ 'text': self.vector_store.chunks[idx], 'metadata': self.vector_store.metadata[idx], 'rrf_score': rrf_scores[idx] }) return results

RRF 里的常数 60 是经验值，来自原论文。它的作用是平滑排名差异，让两路检索的贡献更均衡。我实测下来，混合检索比纯向量检索的召回率高 15% 到 20%，尤其是对于包含专有名词和技术术语的查询。

4.3 提示词工程：让模型好好说话

检索到相关文档块之后，要把它们和用户问题一起塞给 Ollama 里的模型。提示词的设计直接决定了回答质量。我试过很多版本，最后稳定下来的模板是这样的：

PROMPT_TEMPLATE = """你是一个知识库助手。请根据下面提供的参考资料回答用户的问题。 规则： 1. 只使用参考资料中的信息回答，不要编造。 2. 如果参考资料中没有相关信息，直接说“根据现有资料无法回答这个问题”。 3. 回答要简洁、准确，引用具体内容时注明来源。 4. 如果用户的问题涉及多个方面，分点回答。 参考资料： {context} 用户问题：{question} 回答："""

这个模板的关键在于“只使用参考资料”和“无法回答时直接说”。如果不加这两条，模型会倾向于用自己的知识编造答案，这在知识库场景里是致命的。我踩过这个坑：问一个文档里没有的问题，模型编了一段看起来很像真的回答，差点把我误导。

context 的拼接也有讲究。不要把检索到的所有块一股脑塞进去，那样会超出模型的上下文窗口，而且无关信息会干扰模型。我一般取 top 3 到 top 5，每个块截断到 500 字左右，总 context 控制在 2000 字以内。

def build_context(search_results, max_length=2000): context_parts = [] current_length = 0 for result in search_results: text = result['text'][:500] source = result['metadata'].get('source', '未知来源') part = f"[来源：{source}]\n{text}\n" if current_length + len(part) > max_length: break context_parts.append(part) current_length += len(part) return "\n---\n".join(context_parts)

4.4 调用 Ollama 生成回答

Ollama 的 Python 客户端用起来很简单，但有几个参数需要调。temperature设低一点，0.1 到 0.3 之间，这样回答更稳定、更贴近原文。num_ctx要设大一些，默认是 2048，如果你的 context 比较长，设成 4096 或 8192。top_p设 0.9 就行。

import ollama def generate_answer(question, context, model="qwen2.5:7b"): prompt = PROMPT_TEMPLATE.format(context=context, question=question) response = ollama.chat( model=model, messages=[{'role': 'user', 'content': prompt}], options={ 'temperature': 0.2, 'num_ctx': 4096, 'top_p': 0.9, } ) return response['message']['content']

注意：如果你的文档很多，检索出来的 context 经常超过 2000 字，建议把num_ctx调到 8192。但要注意，num_ctx越大，显存和内存占用越高。7B 模型在 8192 上下文下大概需要 8GB 到 10GB 显存。如果显存不够，Ollama 会自动回退到 CPU 推理，速度会慢很多。

5. 常见问题排查与性能优化

5.1 检索结果不相关怎么办

这是最常见的问题。你问“怎么配置 FAISS 的 nprobe”，检索出来的却是“FAISS 安装教程”。原因通常有三个：切块太粗、embedding 模型不适合、检索策略单一。

首先检查切块。如果每个 chunk 有 1000 字以上，那检索出来的块肯定包含大量无关信息。把 max_chunk_size 降到 300 到 500 字，重新建索引。其次检查 embedding 模型。如果你用的是英文模型处理中文文档，效果肯定差。换成 BGE 中文模型。最后，加上 BM25 混合检索，前面已经讲过怎么做了。

还有一个容易被忽视的点：查询改写。用户的问题往往很短，比如“nprobe 怎么调”，直接拿这个去检索，向量可能不够明确。可以用 Ollama 先把问题改写成更完整的查询，比如“FAISS IndexIVFFlat 的 nprobe 参数应该如何设置”，再去检索。这一步能明显提升召回质量。

def rewrite_query(question, model="qwen2.5:7b"): prompt = f"""请把下面的问题改写成更适合检索的查询语句，保持原意，但补充必要的上下文关键词。只输出改写后的查询，不要解释。 原问题：{question} 改写后：""" response = ollama.chat( model=model, messages=[{'role': 'user', 'content': prompt}], options={'temperature': 0.1} ) return response['message']['content'].strip()

5.2 Ollama 响应慢或超时

Ollama 第一次加载模型会慢，因为要把模型权重从磁盘读到内存。之后每次推理，如果模型已经在内存里，速度就正常了。如果你发现每次都很慢，可能是内存不足，模型被反复换入换出。

检查方法：跑ollama ps看模型是否在运行。如果不在，说明被卸载了。解决办法是增加系统内存，或者换更小的模型。另外，num_ctx设太大也会拖慢速度，因为注意力计算量随上下文长度平方增长。如果不需要长上下文，设成 2048 就行。

还有一个坑：Ollama 默认只保留一个模型在内存里。如果你同时用了 embedding 模型和生成模型，而且 embedding 模型也是通过 Ollama 跑的，那两者会互相挤占内存。我的做法是 embedding 用 sentence-transformers 单独跑，不走 Ollama，这样两者互不干扰。

5.3 常见问题速查表

5.4 性能优化的几个实用技巧

第一个技巧是批量编码。SentenceTransformer.encode支持传入列表，一次性编码多个文本比循环单条编码快很多。我处理 1000 个 chunk，批量编码只要十几秒，单条循环要几分钟。

第二个技巧是索引分片。如果你的文档按主题分得很清楚，可以建多个 FAISS 索引，检索时先判断问题属于哪个主题，再去对应的索引里搜。这样每个索引更小，检索更快，而且精度更高。

第三个技巧是缓存。相同的查询没必要每次都重新编码和检索。用一个简单的字典缓存 query 到结果的映射，命中缓存直接返回。对于个人知识库，查询重复率其实挺高的。

from functools import lru_cache @lru_cache(maxsize=100) def cached_search(query): # 注意：lru_cache 不能直接用于带 numpy 数组参数的函数 # 这里只是示意，实际需要把 query 转成字符串再缓存 pass

第四个技巧是异步处理。文档解析和向量化可以并行做，用concurrent.futures.ThreadPoolExecutor开几个线程同时处理多个文件。但要注意，FAISS 的 add 操作不是线程安全的，需要加锁。

6. 从零到一的完整操作流程

6.1 目录结构规划

在开始写代码之前，先把目录结构定好。我习惯这样组织：

my_knowledge_base/ ├── documents/ # 原始文档 │ ├── tech/ │ ├── papers/ │ └── notes/ ├── data/ # 处理后的数据 │ ├── index.faiss # FAISS 索引 │ ├── chunks.pkl # chunk 文本和元数据 │ └── bm25.pkl # BM25 索引 ├── scripts/ │ ├── ingest.py # 文档入库脚本 │ ├── query.py # 查询脚本 │ └── config.py # 配置文件 └── requirements.txt

把原始文档和处理后的数据分开，这样重新建索引时不会误删原始文件。config.py 里放模型名称、chunk 大小、路径这些配置，改起来方便。

6.2 入库脚本的完整实现

入库脚本做四件事：遍历文档、解析、切块、向量化、存索引。我把它写成一个可重复执行的脚本，每次有新文档就重新跑一遍。为了增量更新，可以记录已处理的文件列表，跳过没变化的文件。

import os import hashlib from pathlib import Path from tqdm import tqdm def get_file_hash(file_path): with open(file_path, 'rb') as f: return hashlib.md5(f.read()).hexdigest() def ingest_documents(doc_dir, vector_store, embed_model, processed_files=None): if processed_files is None: processed_files = {} all_chunks = [] all_metadata = [] for root, dirs, files in os.walk(doc_dir): for file in tqdm(files, desc="处理文档"): file_path = os.path.join(root, file) file_hash = get_file_hash(file_path) # 跳过未变化的文件 if file_path in processed_files and processed_files[file_path] == file_hash: continue # 根据扩展名选择解析器 ext = Path(file_path).suffix.lower() try: if ext == '.pdf': text = parse_pdf(file_path) elif ext == '.md': text = parse_markdown(file_path) elif ext in ['.txt', '.py', '.js']: with open(file_path, 'r', encoding='utf-8') as f: text = f.read() elif ext in ['.html', '.htm']: text = parse_html(file_path) else: continue except Exception as e: print(f"解析失败 {file_path}: {e}") continue # 切块 chunks = split_text(text) for i, chunk in enumerate(chunks): all_chunks.append(chunk) all_metadata.append({ 'source': file_path, 'chunk_index': i, 'file_hash': file_hash }) processed_files[file_path] = file_hash if not all_chunks: print("没有新文档需要处理") return processed_files # 批量向量化 print(f"正在向量化 {len(all_chunks)} 个 chunk...") embeddings = embed_model.encode( all_chunks, batch_size=32, show_progress_bar=True, normalize_embeddings=True ) # 加入索引 vector_store.add(embeddings, all_chunks, all_metadata) print(f"入库完成，当前索引共 {vector_store.index.ntotal} 条向量") return processed_files

这个脚本的关键点是增量更新。通过文件哈希判断文件是否变化，只处理新文件或修改过的文件。对于个人知识库，每次全量重建索引太浪费时间，增量更新能省很多事。

6.3 查询脚本的完整实现

查询脚本把前面所有模块串起来：改写查询、编码、混合检索、构建 context、调用 Ollama 生成回答。

def query_knowledge_base(question, vector_store, retriever, embed_model, model="qwen2.5:7b"): # 1. 查询改写 rewritten = rewrite_query(question, model) print(f"改写后的查询：{rewritten}") # 2. 编码 query_embedding = embed_model.encode([rewritten], normalize_embeddings=True)[0] # 3. 混合检索 results = retriever.search(rewritten, query_embedding, top_k=5) # 4. 构建 context context = build_context(results) # 5. 生成回答 answer = generate_answer(question, context, model) return answer, results # 使用示例 if __name__ == "__main__": # 加载索引 store = VectorStore() store.load("data/") # 加载 embedding 模型 embed_model = SentenceTransformer('BAAI/bge-base-zh-v1.5') # 构建混合检索器 retriever = HybridRetriever(store) # 查询 question = "FAISS 的 nprobe 参数怎么调？" answer, sources = query_knowledge_base(question, store, retriever, embed_model) print("\n回答：") print(answer) print("\n参考来源：") for s in sources: print(f"- {s['metadata']['source']} (相关度: {s['rrf_score']:.4f})")

6.4 日常使用与维护建议

这套东西搭好之后，日常使用就是往 documents 目录里丢新文件，然后跑一次 ingest.py。查询用 query.py，或者你可以用 Streamlit 做个简单的 Web 界面，这样不用每次开终端。

维护方面，我建议每个月检查一次索引大小和检索质量。如果发现检索变慢，可能是 chunk 太多了，考虑换 IVF 索引。如果发现回答质量下降，可能是模型需要更新，或者提示词需要调整。

还有一个实用技巧：给文档打标签。在 metadata 里加一个category字段，检索时可以按类别过滤。比如你问“Python 的装饰器怎么用”，可以只在category='python'的 chunk 里检索，这样精度更高。

# 带过滤的检索 def search_with_filter(self, query_embedding, category, top_k=5): # 先检索更多结果，再过滤 results = self.search(query_embedding, top_k * 5) filtered = [r for r in results if r['metadata'].get('category') == category] return filtered[:top_k]

这套方案我从去年用到现在，管理了大概两千多份文档，检索响应时间在 1 秒以内，生成回答大概 3 到 5 秒。硬件是一台 32GB 内存、RTX 3060 12GB 的台式机。如果你用笔记本，8GB 内存跑 3B 模型也能用，就是慢一点。关键是整套东西完全免费，数据完全本地，想怎么改就怎么改。