在实际工作里,经常遇到大量技术文档、API手册需要快速查询。传统Ctrl+F只能搜索精确字符串,无法理解语义。RAG(检索增强生成)模式先对文档做向量化存入索引,检索到相关片段后再交给大模型生成答案,可以有效解决这个问题。本文用纯Python + FAISS搭建一个完全离线、不依赖任何云API的本地RAG知识库,并提供完整可运行代码。
- # 环境准备:需安装 sentence-transformers 和 faiss-cpu
- # pip install sentence-transformers faiss-cpu
复制代码
首先准备文档数据。实际使用时可对PDF、Markdown进行切分,这里用一段示例列表:
- documents = [
- "Docker 容器是一种轻量级的虚拟化技术,它将应用及其依赖打包在一个隔离环境中运行。",
- "Kubernetes 是一个容器编排平台,用于自动化部署、扩展和管理容器化应用。",
- "Python 的 requests 库是一个简单易用的 HTTP 客户端库,支持 GET、POST 等常用方法。",
- "Git 是一个分布式版本控制系统,支持分支管理、合并和提交历史记录。",
- "Docker Compose 允许通过 YAML 文件定义和运行多容器 Docker 应用。",
- ]
复制代码
接着用 sentence-transformers 预训练模型将文本转为 768 维向量。模型 all-MiniLM-L6-v2 首次运行时会自动下载(约 420 MB),支持中文和英文。
- from sentence_transformers import SentenceTransformer
- import faiss
- import numpy as np
- model = SentenceTransformer('all-MiniLM-L6-v2')
- embeddings = model.encode(documents, show_progress_bar=True)
- print(f"向量维度: {embeddings.shape}") # 输出 (5, 768)
复制代码
接下来构建 FAISS 索引。使用 L2 距离(欧几里得距离)的 IndexFlatL2,将向量转为 float32 格式后写入索引。
- dimension = embeddings.shape[1] # 768
- index = faiss.IndexFlatL2(dimension)
- embeddings_32 = embeddings.astype('float32')
- index.add(embeddings_32)
- print(f"已存储 {index.ntotal} 条向量")
复制代码
实现检索函数。输入查询文本,先编码为向量,再用 index.search 获取距离最近的前 top_k 个文档。搜索结果包含距离值和对应文档内容。
- def search(query: str, top_k: int = 3) -> list:
- query_embedding = model.encode([query])
- query_32 = query_embedding.astype('float32')
- distances, indices = index.search(query_32, top_k)
- results = []
- for i, (dist, idx) in enumerate(zip(distances[0], indices[0])):
- results.append({
- "rank": i + 1,
- "distance": round(float(dist), 4),
- "content": documents[idx]
- })
- return results
- # 测试检索
- query = "如何管理容器化应用?"
- results = search(query, top_k=2)
- for r in results:
- print(f"[{r['rank']}] 相似度分数: {r['distance']}")
- print(f" 内容: {r['content']}\n")
复制代码
最后连接大模型生成答案。这里以本地 Ollama 为例,先启动 ollama serve,然后 POST 请求 /api/generate,传入模型名(如 qwen2.5:7b)和提示词。提示词中拼接检索到的文档作为上下文。
- import requests
- def generate_answer(query: str, retrieved_docs: list) -> str:
- context = "\n".join([d['content'] for d in retrieved_docs])
- prompt = f"""你是一个专业的技术助手。请基于以下参考资料回答问题。
- 如果资料中没有答案,请明确说"参考资料中未提及"。
- 参考资料:
- {context}
- 问题:{query}
- """
- response = requests.post(
- "http://localhost:11434/api/generate",
- json={"model": "qwen2.5:7b", "prompt": prompt, "stream": False}
- )
- return response.json().get("response", "")
- # 完整 RAG 流程
- query = "Docker 和 Kubernetes 有什么关系?"
- docs = search(query, top_k=3)
- answer = generate_answer(query, docs)
- print(answer)
复制代码
实际使用中,面对大量文档需要先做切分。建议用 LangChain 的 RecursiveCharacterTextSplitter 按段落或字符数切分,每段控制在 200-500 字,避免长文本导致向量信息稀释。进一步优化可以引入混合检索(BM25 + 向量检索)解决语义模糊,或使用 Cross-Encoder 对 top-20 结果重排序取 top-5 提升精度。数据量超过 500 万条时,推荐升级到 faiss-gpu 或采用 Milvus、Qdrant 等专用向量数据库;若需要高精度中文语义检索,可将 embedding 模型换为 BAAI/bge-base-zh-v1.5。
FAISS 单机可处理千万级向量,配合 IVF + PQ 量化可在精度损失 5% 内压缩 8 倍,让索引常驻内存。整个知识库搭建仅需不到 50 行 Python 代码,文档更新时重新向量化即可,无需重新训练模型,敏感数据也可完全本地控制。 |