背景介绍
你是不是遇到过这种情况：团队积累了几百份文档，每次找东西都要 Ctrl+F 搜半天？或者面对一份 200 页的技术文档，想查某个具体功能，还得一点点翻目录？
传统的搜索方案主要有两种：一是数据库的 LIKE 查询，二是 ElasticSearch。这两种都有明显的缺点——只能匹配关键词，无法理解语义。比如搜「如何重置密码」，它可能找不到包含「忘记密码」或「密码找回」的文档。
这两年大语言模型火起来之后，RAG（检索增强生成）架构成了解决这个问题的热门方案。核心思路是把文档转成向量存到向量数据库里，搜索时计算语义相似度，而不是简单的关键词匹配。
本文手把手教你用 Python 搭建一个本地知识库问答系统。不需要云服务，所有数据都存在本地。
问题描述
我需要构建一个本地知识库问答系统，有这几个需求：
- 文档导入。支持读取本地的 txt、Markdown 和 PDF 文件，把内容转成向量存起来。
- 语义搜索。用户用自然语言提问，系统能找到最相关的文档片段。
- LLM 生成答案。不只返回相关文档，还能让 AI 结合文档内容生成回答。
- 一键部署。不依赖复杂服务，本地能直接跑。
技术选型：用 OpenAI 的 embedding 接口转文本为向量，用 FAISS 做本地向量数据库，用 GPT API 生成答案。
详细步骤
第一步：准备环境
先装好 Python 3.8+，然后：
mkdir local-rag && cd local-rag
pip install openai faiss-cpu python-dotenv pypdf markdown在项目根目录新建 .env 文件：
OPENAI_API_KEY=your-api-key-here第二步：文档加载器
需要一个统一的加载器处理不同格式。创建 document_loader.py：
import os
from pathlib import Path
from typing import List
import markdown
from pypdf import PdfReader
class Document:
def __init__(self, content: str, source: str):
self.content = content
self.source = source
def load_document(file_path: str) -> Document:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".txt":
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
elif suffix == ".md":
with open(file_path, "r", encoding="utf-8") as f:
md_content = f.read()
content = markdown.markdown(md_content)
elif suffix == ".pdf":
reader = PdfReader(file_path)
content = "\n".join([page.extract_text() for page in reader.pages])
else:
raise ValueError(f"不支持的文件格式: {suffix}")
return Document(content=content, source=file_path)
def load_directory(directory: str) -> List[Document]:
docs = []
for root, _, files in os.walk(directory):
for file in files:
if file.endswith((".txt", ".md", ".pdf")):
try:
doc = load_document(os.path.join(root, file))
docs.append(doc)
except Exception as e:
print(f"加载 {file} 失败: {e}")
return docs支持 txt、md、pdf 三种格式。Markdown 转 HTML 再提纯文本，PDF 用 pypdf 逐页提取。
第三步：构建向量索引
把文档内容切成小块，转成向量存 FAISS。创建 index_builder.py：
import os
import numpy as np
import faiss
from dotenv import load_dotenv
from openai import OpenAI
from document_loader import Document, load_directory
load_dotenv()
client = OpenAI()
def get_embedding(text: str) -> np.ndarray:
response = client.embeddings.create(model="text-embedding-3-small", input=text)
return np.array(response.data[0].embedding, dtype="float32")
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
chunks = []
start = 0
text_len = len(text)
while start < text_len:
end = min(start + chunk_size, text_len)
chunk = text[start:end]
chunks.append(chunk)
start += chunk_size - overlap
return chunks
def build_index(documents: list[Document], index_path: str = "index.faiss"):
all_chunks = []
all_metadata = []
for doc in documents:
chunks = chunk_text(doc.content)
for chunk in chunks:
all_chunks.append(chunk)
all_metadata.append({"source": doc.source})
print(f"共 {len(all_chunks)} 个文本块")
embeddings = []
batch_size = 100
for i in range(0, len(all_chunks), batch_size):
batch = all_chunks[i:i+batch_size]
response = client.embeddings.create(model="text-embedding-3-small", input=batch)
embeddings.extend([np.array(d.embedding, dtype="float32") for d in response.data])
print(f"已处理 {min(i+batch_size, len(all_chunks))}/{len(all_chunks)} 个块")
embeddings_array = np.array(embeddings).astype("float32")
dimension = embeddings_array.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(embeddings_array)
faiss.write_index(index, index_path)
import json
with open("metadata.json", "w", encoding="utf-8") as f:
json.dump(all_metadata, f, ensure_ascii=False)
print(f"索引已保存到 {index_path}")
return index, all_metadata
if __name__ == "__main__":
docs = load_directory("./knowledge")
build_index(docs)几个关键设计：文本分块用了重叠，块之间有 50 字符重叠，保证上下文不断裂；批量调用 API 提效；用 L2 距离的 FAISS 索引。
第四步：问答搜索
搜索和问答核心逻辑。创建 retriever.py：
import json
import numpy as np
import faiss
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
def load_index(index_path: str = "index.faiss"):
index = faiss.read_index(index_path)
with open("metadata.json", "r", encoding="utf-8") as f:
metadata = json.load(f)
return index, metadata
def get_embedding(text: str) -> np.ndarray:
response = client.embeddings.create(model="text-embedding-3-small", input=text)
return np.array(response.data[0].embedding, dtype="float32")
def search(query: str, index, metadata, top_k: int = 3):
query_vector = get_embedding(query).reshape(1, -1)
distances, indices = index.search(query_vector, top_k)
results = []
for idx, distance in zip(indices[0], distances[0]):
if idx < len(metadata):
results.append({"index": int(idx), "distance": float(distance), "source": metadata[idx]["source"]})
return results
def generate_answer(query: str, context_docs: list[str]) -> str:
context = "\n\n".join([f"文档 {i+1}:\n{doc}" for i, doc in enumerate(context_docs)])
prompt = f"基于以下参考文档回答用户问题。如果文档中没有相关信息，请如实说明。\n\n参考文档：\n{context}\n\n用户问题：{query}\n\n请给出回答："
response = client.chat.completions.create(model="gpt-3.5-turbo", messages=[{{"role": "user", "content": prompt}}], temperature=0.7)
return response.choices[0].message.content
def ask(query: str, index, metadata, chunks: list[str], top_k: int = 3):
results = search(query, index, metadata, top_k)
context_docs = [chunks[r["index"]] for r in results]
answer = generate_answer(query, context_docs)
return {{"answer": answer, "sources": [r["source"] for r in results]}}第五步：主程序
把所有功能串起来。创建 main.py：
from document_loader import load_directory, load_document
from index_builder import build_index
from retriever import load_index, ask
import pickle
def init_knowledge_base(directory: str = "./knowledge"):
print("加载文档...")
docs = load_directory(directory)
print(f"已加载 {len(docs)} 个文档")
print("构建索引...")
index, metadata = build_index(docs)
with open("index_data.pkl", "wb") as f:
pickle.dump({{"metadata": metadata, "index": index}})
return index, metadata
def main():
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--build":
init_knowledge_base()
print("知识库构建完成！")
else:
with open("index_data.pkl", "rb") as f:
data = pickle.load(f)
index = data["index"]
metadata = data["metadata"]
print("本地知识库问答系统已启动")
print("输入问题进行搜索，输入 quit 退出\n")
while True:
query = input("请输入问题: ")
if query.lower() == "quit":
break
result = ask(query, index, metadata)
print(f"\n答案: {{result[\"answer\"]}}")
print(f"参考文档: {{result[\"sources\"]}}\n")
if __name__ == "__main__":
main()运行结果
先准备测试文档。在 knowledge 文件夹下放几个文件，比如 user_guide.txt、faq.md、技术文档.pdf。
然后运行构建：
python main.py --build输出：
加载文档...
已加载 3 个文档
构建索引...
共 45 个文本块
已处理 100/45 个块
索引已保存到 index.faiss
知识库构建完成！启动问答：
python main.py搜索「如何重置登录密码」返回：
答案: 根据用户手册中的"密码重置"章节，您可以通过以下步骤重置密码：
1. 在登录页面点击"忘记密码"
2. 输入您注册时使用的邮箱
3. 点击"发送验证邮件"
4. 点击邮件中的链接设置新密码
参考文档: user_guide.txt语义搜索起作用了——搜「重置密码」能找到「忘记密码」的内容。
搜索「项目怎么安装」返回：
答案: 根据技术文档，安装步骤如下：
1. 确保已安装 Node.js 18+
2. 运行 npm install
3. 配置 .env 文件
4. 运行 npm start
参考文档: 技术文档.pdf, user_guide.txt跨文档整合也没问题。
总结
一个完整的本地知识库问答系统就搭建好了。
这个系统的优点：语义理解能力强，不只是关键词匹配；部署简单，一个命令就跑起来；扩展方便，丢文档进去重新 build 就行；成本可控，只在创建索引和提问时调用 API，按量收费。
后续可以优化的方向：加更多文档格式支持、增量更新索引、本地部署 embedding 模型、加入多轮对话功能。这套方案已经有生产级的基础能力了。