大模型RAG知识库搭建：文档向量化与检索增强生成

黑漂技术佬的第8篇实战笔记 —— 大模型很聪明，但它不知道你家售货柜昨天报了什么错

一、RAG是什么？为什么需要它？

大语言模型（LLM）很强，能写代码、能聊天、能写诗。但它有三个硬伤：

1. 知识截止：训练数据有截止日期，问它昨天的事它不知道
2. 幻觉：不知道的事也敢一本正经地胡说八道
3. 不认识你的私有数据：你公司的故障手册、运维文档，它压根没见过

RAG（Retrieval Augmented Generation，检索增强生成）就是解决这些问题的方案。思路简单粗暴：

用户提问 → 先从你的知识库里检索相关文档 → 把检索到的文档塞进Prompt → 大模型基于这些文档回答

打个比方：大模型像一个聪明但没见过你公司内部资料的实习生。RAG就是在他回答之前，先把你公司的相关文档翻出来递给他看一眼。他看完之后再回答，准确率就高多了。

为什么不直接把所有文档都塞进Prompt？因为大模型有上下文长度限制。就算窗口够大，塞太多无关内容也会让模型"走神"，回答质量下降。所以需要先检索出最相关的片段，只塞这些进去。

二、RAG完整流程

2.1 流程总览

┌─────────────────────────── 离线建库阶段 ───────────────────────────┐
│                                                                     │
│  原始文档 → 文本分块(Chunking) → 向量化(Embedding) → 存入向量数据库  │
│  (PDF/Word/TXT)     ↓                ↓                    ↓         │
│                  切成小段落      每段转成向量        FAISS/Milvus     │
│                  ~500字/块      768/1536维          支持相似度检索    │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────── 在线问答阶段 ───────────────────────────┐
│                                                                     │
│  用户问题 → 向量化(Embedding) → 向量数据库检索TopK → 拼接Prompt     │
│                                    ↓                    ↓            │
│                              返回最相关的文档块     "根据以下文档     │
│                                                 回答用户的问题"       │
│                                                      ↓               │
│                                                 大模型生成回答        │
│                                                      ↓               │
│                                                  返回给用户           │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

2.2 文档分块（Chunking）

文档不能整篇转成一个向量——太长了信息会被压缩得面目全非。需要把文档切成小块，每块几百字左右。

分块策略很关键，切得太碎语义不完整，切得太大检索精度下降。常见做法是按Token数切，带重叠（overlap），避免在句子中间硬切导致语义断裂。

from typing import List

class TextChunker:
    """文本分块器"""
    
    def __init__(self, chunk_size=500, overlap=50):
        """
        :param chunk_size: 每块的最大字符数
        :param overlap: 相邻块之间的重叠字符数（防止语义断裂）
        """
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk_text(self, text: str) -> List[str]:
        """将长文本切分成多个块"""
        chunks = []
        start = 0
        
        while start < len(text):
            # 取一块
            end = start + self.chunk_size
            chunk = text[start:end]
            
            # 尽量在句号/换行处断开，避免硬切
            if end < len(text):
                # 往后找最近的句号或换行
                for sep in ['\n\n', '\n', '。', '！', '？', '.', '!', '?']:
                    last_sep = chunk.rfind(sep)
                    if last_sep > self.chunk_size * 0.5:  # 至少保留一半长度
                        end = start + last_sep + len(sep)
                        chunk = text[start:end]
                        break
            
            chunks.append(chunk.strip())
            # 下一块从overlap处开始
            start = end - self.overlap
        
        # 过滤空块
        return [c for c in chunks if c]
    
    def chunk_with_metadata(self, text: str, source: str, page: int = 0) -> List[dict]:
        """分块并附加元数据"""
        chunks = self.chunk_text(text)
        return [
            {
                'content': chunk,
                'source': source,      # 来源文件名
                'page': page,          # 页码
                'chunk_index': i,      # 块序号
                'char_count': len(chunk)
            }
            for i, chunk in enumerate(chunks)
        ]

# 使用示例
chunker = TextChunker(chunk_size=500, overlap=50)

sample_text = """
无人售货柜常见故障及处理方法：

一、柜门无法打开
1. 检查电磁锁供电是否正常，正常电压为12V
2. 检查门控板指示灯，红灯表示通信异常
3. 重启门控板，长按复位键3秒
4. 若以上步骤无效，更换电磁锁模块

二、商品识别错误
1. 检查摄像头是否被遮挡
2. 确认特征库版本是否最新
3. 清洁摄像头镜头
4. 重新校准货架位置参数

三、支付失败
1. 检查网络连接状态
2. 确认支付模块是否在线
3. 查看支付日志中的错误码
4. 常见错误码：1001=网络超时，1002=余额不足，1003=签名错误
"""

chunks = chunker.chunk_with_metadata(sample_text, source="故障手册.pdf", page=1)
print(f"共切分为 {len(chunks)} 个块")
for i, chunk in enumerate(chunks):
    print(f"  块{i}: {chunk['char_count']}字, 内容前30字: {chunk['content'][:30]}...")

2.3 文本向量化（Embedding）

文本分块后，每一块都要转成向量。这里用的不是ResNet（那是图像的），而是文本Embedding模型。

文本Embedding的原理：把一段文本映射成一个高维向量，语义相近的文本向量距离也近。"柜门打不开"和"电磁锁故障"的向量会比较近，而"柜门打不开"和"今天天气不错"的向量会很远。

from sentence_transformers import SentenceTransformer
import numpy as np

class TextEmbedder:
    """文本向量化器"""
    
    def __init__(self, model_name='BAAI/bge-small-zh-v1.5'):
        """
        :param model_name: Embedding模型名称
        常用选择:
        - BAAI/bge-small-zh-v1.5: 中文, 512维, 速度快, 本地可跑
        - BAAI/bge-large-zh-v1.5: 中文, 1024维, 精度高, 需要更多资源
        - moka-ai/m3e-small: 中文, 384维, 轻量级
        - text-embedding-ada-002: OpenAI, 1536维, 需要API调用
        """
        self.model = SentenceTransformer(model_name)
        self.dimension = self.model.get_sentence_embedding_dimension()
        print(f"Embedding模型加载完成: {model_name}, 输出维度: {self.dimension}")
    
    def embed(self, text: str) -> np.ndarray:
        """将单段文本转为向量"""
        embedding = self.model.encode(text, normalize_embeddings=True)
        return embedding.astype('float32')
    
    def embed_batch(self, texts: List[str]) -> np.ndarray:
        """批量向量化（比循环调用快得多）"""
        embeddings = self.model.encode(texts, normalize_embeddings=True, 
                                        batch_size=32, show_progress_bar=True)
        return embeddings.astype('float32')

# 使用示例
embedder = TextEmbedder('BAAI/bge-small-zh-v1.5')

vec1 = embedder.embed("售货柜门打不开")
vec2 = embedder.embed("电磁锁供电异常")
vec3 = embedder.embed("今天天气真好")

sim_12 = np.dot(vec1, vec2)  # 余弦相似度（已归一化）
sim_13 = np.dot(vec1, vec3)

print(f"'门打不开' vs '电磁锁异常' 相似度: {sim_12:.4f}")
print(f"'门打不开' vs '天气真好' 相似度: {sim_13:.4f}")
# 输出大概: 0.72 vs 0.21 —— 语义相近的文本相似度明显更高

2.4 向量入库

import faiss
import json

class VectorStore:
    """向量存储管理"""
    
    def __init__(self, dimension, index_path="rag_index.faiss", 
                 metadata_path="rag_metadata.json"):
        self.dimension = dimension
        self.index_path = index_path
        self.metadata_path = metadata_path
        self.metadata = []
        
        # 使用IVFFlat索引（适合中小规模知识库）
        nlist = 100
        quantizer = faiss.IndexFlatIP(dimension)
        self.index = faiss.IndexIVFFlat(quantizer, dimension, nlist, 
                                         faiss.METRIC_INNER_PRODUCT)
        self.is_trained = False
    
    def add_documents(self, chunks: List[dict], embedder: TextEmbedder):
        """将文档块向量化并存入索引"""
        texts = [chunk['content'] for chunk in chunks]
        vectors = embedder.embed_batch(texts)
        
        # 首次添加需要训练索引
        if not self.is_trained:
            self.index.train(vectors)
            self.is_trained = True
        
        self.index.add(vectors)
        self.metadata.extend(chunks)
        
        print(f"已添加 {len(chunks)} 个文档块, 总计 {self.index.ntotal}")
    
    def save(self):
        """持久化索引和元数据"""
        faiss.write_index(self.index, self.index_path)
        with open(self.metadata_path, 'w', encoding='utf-8') as f:
            json.dump(self.metadata, f, ensure_ascii=False, indent=2)
        print(f"索引已保存: {self.index_path}")
    
    def load(self):
        """加载已有索引"""
        self.index = faiss.read_index(self.index_path)
        with open(self.metadata_path, 'r', encoding='utf-8') as f:
            self.metadata = json.load(f)
        self.is_trained = True
        print(f"索引已加载: {self.index.ntotal} 条记录")

2.5 检索与生成

class RAGRetriever:
    """RAG检索器"""
    
    def __init__(self, vector_store: VectorStore, embedder: TextEmbedder):
        self.store = vector_store
        self.embedder = embedder
    
    def retrieve(self, question: str, top_k: int = 5) -> List[dict]:
        """
        根据问题检索相关文档块
        :param question: 用户问题
        :param top_k: 返回前K个最相关的文档块
        """
        # 问题向量化
        query_vec = self.embedder.embed(question)
        query_vec = np.array([query_vec])
        
        # 向量检索
        scores, indices = self.store.index.search(query_vec, top_k)
        
        # 组装结果
        results = []
        for i in range(top_k):
            idx = indices[0][i]
            if idx >= 0 and idx < len(self.store.metadata):
                chunk = self.store.metadata[idx].copy()
                chunk['score'] = float(scores[0][i])
                results.append(chunk)
        
        return results
    
    def build_prompt(self, question: str, retrieved_chunks: List[dict]) -> str:
        """将检索结果拼入Prompt"""
        context = "\n\n---\n\n".join([
            f"[来源: {chunk['source']}, 块{chunk['chunk_index']}]\n{chunk['content']}"
            for chunk in retrieved_chunks
        ])
        
        prompt = f"""你是一个专业的无人售货柜运维助手。请根据以下参考文档回答用户的问题。

要求：
1. 只根据参考文档内容回答，不要编造
2. 如果文档中没有相关信息，明确告知用户
3. 回答要简洁准确，给出具体操作步骤

参考文档：
{context}

用户问题：{question}

回答："""
        return prompt

2.6 接入大模型生成回答

import openai

class RAGSystem:
    """完整的RAG系统"""
    
    def __init__(self, vector_store, embedder, llm_model="gpt-3.5-turbo"):
        self.retriever = RAGRetriever(vector_store, embedder)
        self.llm_model = llm_model
    
    def ask(self, question: str, top_k: int = 5) -> dict:
        """
        完整的RAG问答流程
        :return: {answer, sources, retrieval_results}
        """
        # 第1步：检索
        retrieved = self.retriever.retrieve(question, top_k)
        
        if not retrieved:
            return {
                'answer': "抱歉，知识库中没有找到相关内容。",
                'sources': [],
                'retrieval_results': []
            }
        
        # 第2步：构建Prompt
        prompt = self.retriever.build_prompt(question, retrieved)
        
        # 第3步：调用大模型
        response = openai.ChatCompletion.create(
            model=self.llm_model,
            messages=[
                {"role": "user", "content": prompt}
            ],
            temperature=0.1,  # 低温度=更确定性
            max_tokens=1000
        )
        
        answer = response.choices[0].message.content
        
        return {
            'answer': answer,
            'sources': [chunk['source'] for chunk in retrieved],
            'retrieval_scores': [chunk['score'] for chunk in retrieved]
        }

# 使用示例
rag = RAGSystem(vector_store, embedder)

result = rag.ask("售货柜门打不开怎么办？")
print(f"回答: {result['answer']}")
print(f"参考来源: {result['sources']}")

三、文本Embedding模型选择

模型	语言	维度	特点	部署方式
text-embedding-ada-002	多语言	1536	效果好但需API	OpenAI API
BAAI/bge-large-zh-v1.5	中文	1024	中文效果最佳	本地部署
BAAI/bge-small-zh-v1.5	中文	512	轻量快速	本地部署
moka-ai/m3e-base	中文	768	社区流行	本地部署
BAAI/bge-m3	多语言	1024	支持长文本	本地部署

选择建议：

• 预算充足、不介意API调用：ada-002，效果稳定
• 中文场景、本地部署：bge-large-zh，中文检索效果最好
• 资源受限的嵌入式端：bge-small-zh或m3e-small，体积小速度快
• 需要支持多语言：bge-m3

四、LangChain框架整合

LangChain是RAG领域最流行的框架，封装了上面那些步骤，几行代码就能搭起来：

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader

# 第1步：加载文档
loader = TextLoader("运维手册.txt", encoding='utf-8')
documents = loader.load()

# 第2步：分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)
chunks = text_splitter.split_documents(documents)

# 第3步：向量化并入库
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={'device': 'cpu'}
)
vectorstore = FAISS.from_documents(chunks, embeddings)

# 第4步：创建检索链
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.1)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # stuff=把所有检索结果塞进一个Prompt
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

# 第5步：提问
question = "支付失败错误码1002是什么意思？"
result = qa_chain({"query": question})

print(f"回答: {result['result']}")
print(f"参考文档数: {len(result['source_documents'])}")

LangChain的RecursiveCharacterTextSplitter比我们自己写的分块器更智能，它会按分隔符优先级递归切分，尽量保持语义完整。

五、无人售货柜运维知识库RAG实战

5.1 故障手册向量化

import os
from langchain.document_loaders import DirectoryLoader, PDFMinerLoader

def build_maintenance_kb(docs_dir: str, vectorstore_path: str):
    """
    构建运维知识库
    :param docs_dir: 文档目录（包含PDF/Word/TXT）
    :param vectorstore_path: 向量库保存路径
    """
    # 加载多种格式的文档
    all_chunks = []
    
    # TXT文件
    txt_loader = DirectoryLoader(docs_dir, glob="**/*.txt", 
                                  loader_cls=TextLoader)
    all_chunks.extend(txt_loader.load())
    
    # PDF文件
    pdf_loader = DirectoryLoader(docs_dir, glob="**/*.pdf",
                                  loader_cls=PDFMinerLoader)
    all_chunks.extend(pdf_loader.load())
    
    # 分块
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50
    )
    chunks = text_splitter.split_documents(all_chunks)
    
    # 向量化入库
    embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
    vectorstore = FAISS.from_documents(chunks, embeddings)
    
    # 保存
    vectorstore.save_local(vectorstore_path)
    print(f"运维知识库构建完成: {len(chunks)} 个文档块")
    
    return vectorstore

# 构建知识库
vectorstore = build_maintenance_kb("./运维文档", "./rag_vectorstore")

5.2 运维人员提问 → 检索 → AI回答

class MaintenanceAssistant:
    """运维助手"""
    
    def __init__(self, vectorstore_path: str):
        # 加载向量库
        self.embeddings = HuggingFaceEmbeddings(
            model_name="BAAI/bge-small-zh-v1.5"
        )
        self.vectorstore = FAISS.load_local(
            vectorstore_path, self.embeddings
        )
        
        # 创建问答链
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=ChatOpenAI(temperature=0.1),
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(search_kwargs={"k": 5}),
            return_source_documents=True
        )
    
    def ask(self, question: str) -> str:
        """运维人员提问"""
        result = self.qa_chain({"query": question})
        
        answer = result['result']
        sources = result['source_documents']
        
        # 格式化输出
        output = f"问题: {question}\n\n"
        output += f"回答: {answer}\n\n"
        output += f"参考来源:\n"
        for i, doc in enumerate(sources):
            source_file = doc.metadata.get('source', '未知')
            output += f"  {i+1}. {source_file}\n"
        
        return output

# 实际使用
assistant = MaintenanceAssistant("./rag_vectorstore")

questions = [
    "售货柜支付失败怎么处理？",
    "电磁锁不工作，电压应该是多少？",
    "摄像头识别不到商品，排查步骤是什么？"
]

for q in questions:
    print(assistant.ask(q))
    print("=" * 60)

六、RAG优化技巧

Chunk大小调优：没有万能参数，需要根据文档类型实验。FAQ类文档用小块（200-300字），技术手册类用大块（500-800字）。

混合检索：纯向量检索对精确匹配（如错误码"1002"）效果不好，可以结合关键词检索：

# 简单的混合检索：向量检索 + BM25关键词检索
def hybrid_search(question, vectorstore, top_k=5):
    # 向量检索
    vec_results = vectorstore.similarity_search_with_score(question, k=top_k*2)
    
    # 关键词检索（简单版：检查是否包含问题中的关键词）
    keywords = [w for w in question if len(w) > 1]
    keyword_results = []
    for doc, score in vec_results:
        kw_score = sum(1 for kw in keywords if kw in doc.page_content)
        keyword_results.append((doc, score, kw_score))
    
    # 融合排序：向量分数 * 0.7 + 关键词分数 * 0.3
    keyword_results.sort(key=lambda x: x[1] * 0.7 + x[2] * 0.3, reverse=True)
    return keyword_results[:top_k]

重排序模型：用一个专门的模型对检索结果重新排序，比简单距离排序精度更高。常用的有bge-reranker：

from sentence_transformers import CrossEncoder

# 加载重排序模型
reranker = CrossEncoder('BAAI/bge-reranker-base')

def rerank(question, documents, top_k=5):
    """使用CrossEncoder重排序"""
    pairs = [(question, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)
    
    ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:top_k]]

RAG的本质就一句话：先找对资料，再让大模型回答。找资料靠向量检索，回答靠大模型。检索质量决定了RAG的上限，模型能力决定了下限。所以别光盯着大模型调参，向量库的构建质量才是第一位的。