RAG系统构建技能

构建完整的本地RAG系统，支持离线文档导入、语义搜索和AI驱动的问答功能。

🎯 技能功能

本技能指导您构建一个完整的RAG系统，具备以下功能：

• - 导入文档：支持多种格式（TXT、PDF、DOCX、MD、HTML、JSON、XML）
• 生成嵌入向量：使用sentence-transformers（离线，无需API）
• 本地存储向量：使用FAISS进行快速相似度搜索
• 提供问答界面：通过命令行和网页界面
• 完全离线运行：无需外部API调用

📦 前置条件

bash

需要Python 3.8+

安装依赖

🚀 快速开始

1. 创建项目结构

bash

创建项目目录

创建主要文件

2. 下载嵌入模型

bash

本地下载sentence-transformers模型

3. 配置系统

创建config.py：

python
import os
from dataclasses import dataclass

@dataclass
class Config:
embedding_model: str = sentence-transformers/all-MiniLM-L6-v2
localmodelpath: str = ./models/all-MiniLM-L6-v2
chunk_size: int = 512
chunk_overlap: int = 128
vectorstorepath: str = vector_store
defaulttopk: int = 5
supported_formats: tuple = (.txt, .pdf, .docx, .md, .html, .json, .xml)

4. 构建核心组件

嵌入模块 (embeddings.py)

python
import os
import numpy as np
from typing import List
from sentence_transformers import SentenceTransformer
from config import config

class EmbeddingModel:
def init(self, model_name: str = None):
self.modelname = modelname or config.embedding_model
self.model = None
self.loadmodel()

def loadmodel(self):
加载嵌入模型，支持本地回退
print(f正在加载嵌入模型：{self.model_name})

# 优先尝试本地模型
localpath = config.localmodel_path
if os.path.exists(local_path):
print(f使用本地模型：{local_path})
try:
self.model = SentenceTransformer(local_path)
print(本地模型加载成功)
return
except Exception as e:
print(f加载本地模型出错：{e})

# 回退到HuggingFace
try:
self.model = SentenceTransformer(self.model_name)
print(从HuggingFace加载模型)
except Exception as e:
print(f错误：{e})
raise

def encode(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
将文本编码为嵌入向量
if not texts:
return np.array([])

embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
batchembeddings = self.model.encode(batch, convertto_numpy=True)
embeddings.append(batch_embeddings)

return np.vstack(embeddings)

向量存储模块 (vector_store.py)

python
import os
import json
import faiss
import numpy as np
from config import config

class VectorStore:
def init(self, base_path: str = .):
self.basepath = basepath
self.vectorstorepath = config.getvectorstorepath(basepath)
self.index = None
self.metadata = []

# 创建目录（如果不存在）
os.makedirs(self.vectorstorepath, exist_ok=True)

def build_index(self, embeddings: np.ndarray, metadata: list):
从嵌入向量构建FAISS索引
print(f正在构建索引，包含{len(embeddings)}个向量)

# 创建FAISS索引
dimension = embeddings.shape[1]
self.index = faiss.IndexFlatIP(dimension) # 内积 = 余弦相似度

# 归一化嵌入向量以计算余弦相似度
faiss.normalize_L2(embeddings)
self.index.add(embeddings)

self.metadata = metadata
print(f索引构建完成，包含{len(embeddings)}个向量)

def save(self):
将索引和元数据保存到磁盘
indexpath = os.path.join(self.vectorstorepath, config.indexfile)
metadatapath = os.path.join(self.vectorstorepath, config.metadatafile)

# 保存FAISS索引
faiss.writeindex(self.index, indexpath)

# 保存元数据
with open(metadata_path, w, encoding=utf-8) as f:
json.dump(self.metadata, f, ensure_ascii=False, indent=2)

print(f索引已保存到{index_path})
print(f元数据已保存到{metadata_path})

def load(self):
从磁盘加载索引和元数据
indexpath = os.path.join(self.vectorstorepath, config.indexfile)
metadatapath = os.path.join(self.vectorstorepath, config.metadatafile)

if os.path.exists(indexpath) and os.path.exists(metadatapath):
self.index = faiss.readindex(indexpath)
with open(metadata_path, r, encoding=utf-8) as f:
self.metadata = json.load(f)
print(f已加载索引，包含{self.index.ntotal}个向量)
return True
return False

检索模块 (retriever.py)

python
import numpy as np
from config import config

class Retriever:
def init(self, vector_store):
self.vectorstore = vectorstore

def search(self, query: str, top_k: int = None) -> list:
搜索相关文档
if top_k is None:
topk = config.defaulttop_k

if self.vector_store.index is None:
print(未加载索引，请先导入文档。)
return []

# 编码查询
from embeddings import EmbeddingModel
embedding_model = EmbeddingModel()
queryembedding = embeddingmodel.encode_single(query)

# 归一化以计算余弦相似度
queryembedding = np.expanddims(query_embedding, axis=0)
faiss.normalizeL2(queryembedding)

# 搜索
scores, indices = self.vectorstore.index.search(queryembedding, top_k)

# 返回带元数据的结果
results = []
for i, idx in enumerate(indices[0]):
if idx < len(self.vector_store.metadata):
result = self.vector_store.metadata[idx].copy()
result[score] = float(scores[0][i])
results.append(result)

return results

5. 创建命令行界面 (rag.py)

python
import os
import sys
import click
from ingestion import IngestionPipeline
from embeddings import EmbeddingModel
from vector_store import VectorStore
from retriever import Retriever
from config import config

@click.group()
def cli():
OpenClaw RAG系统 - 本地文档检索
pass

@cli.command()
@click.option(--docs-path, required=True, help=包含文档的文件夹路径)
@click.option(--chunk-size, default=512, help=文本分割的块大小)
@click.option(--chunk-overlap, default=128, help=块重叠大小)
def ingest(docspath, chunksize, chunk_overlap):
从文件夹导入文档到向量存储
click.echo(f开始从以下路径导入：{docs_path})

# 初始化组件
ingestion = IngestionPipeline(chunksize=chunksize, chunkoverlap=chunkoverlap)
embedding_model = EmbeddingModel()
vector_store = VectorStore()

# 导入文档
try:
chunks = ingestion.ingestfolder(docspath)

if not chunks:
click.echo(未找到或处理任何文档。)
return

# 提取文本和元数据
texts = [chunk[text] for chunk in chunks]
metadata = [{
text: chunk[text],
source: chunk[source],
doctype: chunk[doctype],