一切的起点是一顿臭骂

上个月，我被领导叫进办公室骂了整整二十分钟。

起因是这样的——我们部门负责维护一套内部知识库系统，里面沉淀了公司近五年的技术文档、故障处理手册、还有各种规范流程。问题是，这玩意儿除了当摆设，几乎没人用。为啥？因为搜索太烂了，关键词匹配的那种，你搜服务器宕机怎么办，它给你返回一堆包含服务器的文档，真正有用的那篇反而排在第三页。

新同事入职问问题，老员工翻文档找答案，大家宁可在群里@人问，也不愿意去知识库里查。

然后领导发话了：你不是天天研究什么大模型吗？能不能整个智能问答，让大家直接问问题就能得到答案？

我当时脑子一热，拍胸脯说没问题。结果第一版上线三天就被骂下来了——用户问我们的MySQL主从切换流程是什么，大模型回答得头头是道，但内容完全是它自己编的！跟我们公司的实际流程八竿子打不着。

这就是所谓的大模型幻觉问题，我当时对RAG的理解还停留在把文档丢进去就行的水平，太天真了。

不过，后来的故事还算圆满。我花了将近三周时间重构了整个方案，现在这套系统已经成了部门的标配工具，月活跃用户从0涨到了200多，领导在季度会上还专门表扬了一回。今天这篇文章，我就把整个踩坑过程原原本本地记录下来，包括代码、架构设计、以及那些教科书上不会告诉你的实战细节。

一、RAG到底在解决什么问题

在动手之前，我想先聊聊RAG这个概念，因为很多刚接触的朋友容易搞混。

大模型很强，但它有两个致命弱点：

第一，知识有截止日期。 GPT-4的训练数据截止到某个时间点，它不知道你们公司上周发布的新规范，也不知道你们昨天刚修复的那个bug是怎么解决的。

第二，会一本正经地胡说八道。 当大模型遇到它不知道的问题时，它不会老老实实说我不知道，而是会基于它学过的通用知识，给你编一个看起来很合理但其实是错的答案。这就是所谓的幻觉（Hallucination）。

RAG（Retrieval-Augmented Generation，检索增强生成）的核心思路其实很简单：别让大模型靠想象力答题，先帮它把参考资料找出来，让它照着资料回答。
 

具体来说分三步：

1. 把你的私有文档切成小块，转成向量存起来
2. 用户提问时，先根据问题检索出最相关的文档片段
3. 把问题和检索到的内容一起喂给大模型，让它基于这些材料生成答案

听起来不复杂对吧？我当时也是这么想的，然后就踩了一堆坑。

二、第一个大坑：文档切分没那么简单

我最初的方案特别粗暴——用LangChain的RecursiveCharacterTextSplitter，设置chunk_size=500，overlap=50，直接把所有文档切成小块。
 

代码写起来确实很简单：

from langchain.text_splitter import RecursiveCharacterTextSplitter

def naive_split(text):
    最初的简单切分方案——后来证明这是个坑
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50,
        separators=[\n\n, \n, 。, ！, ？,  , ]
    )
    chunks = splitter.split_text(text)
    return chunks

# 测试一下
sample_text = 
# MySQL主从切换操作手册

## 1. 前置检查
在执行主从切换之前，必须完成以下检查：
- 确认从库同步状态正常（Seconds_Behind_Master = 0）
- 确认没有正在执行的大事务
- 通知相关业务方，确认切换时间窗口

## 2. 切换步骤
2.1 在主库执行只读设置
SET GLOBAL read_only = 1;

2.2 等待从库完全同步
在从库执行 SHOW SLAVE STATUS，确认 Seconds_Behind_Master = 0

2.3 停止从库复制
STOP SLAVE;
RESET SLAVE ALL;

## 3. 回滚方案
如果切换失败，按以下步骤回滚...


chunks = naive_split(sample_text)
for i, chunk in enumerate(chunks):
    print(f Chunk {i+1} )
    print(chunk[:100] + ... if len(chunk) > 100 else chunk)

看起来没毛病是吧？但实际用起来问题大了。

有一次用户问：MySQL切换前需要做哪些检查？系统返回的文档片段是这样的：

确认没有正在执行的大事务
- 通知相关业务方，确认切换时间窗口

## 2. 切换步骤
2.1 在主库执行只读设置
SET GLOBAL read_only = 1;

发现问题了吗？这个片段恰好从检查步骤的中间切开了！第一条检查项确认从库同步状态正常被切到了上一个chunk里。用户问的是需要做哪些检查，结果我们给大模型的参考资料里，第一条检查项就没包含进去。

核心教训：机械地按字数切分，会打断文档的语义完整性。

后来我改成了基于语义结构的切分策略：

import re
from typing import List, Dict

class SmartDocumentSplitter:
    
    语义感知的文档切分器
    核心思路：尊重文档的原有结构，按标题、段落等语义边界切分
    
    
    def __init__(self, max_chunk_size=800, min_chunk_size=100):
        self.max_chunk_size = max_chunk_size
        self.min_chunk_size = min_chunk_size
    
    def split_markdown(self, text: str) -> List[Dict]:
        
        针对Markdown文档的切分
        保持标题层级结构，每个chunk都带上完整的上下文路径
        
        chunks = []
        current_headers = {1: , 2: , 3: }  # 记录当前的标题层级
        
        # 按行处理，识别标题和内容
        lines = text.split('\n')
        current_content = []
        
        for line in lines:
            # 检测Markdown标题
            header_match = re.match(r'^(#{1,3})\s+(.+)$', line)
            
            if header_match:
                # 遇到新标题，先保存之前的内容
                if current_content:
                    chunk_text = '\n'.join(current_content).strip()
                    if len(chunk_text) >= self.min_chunk_size:
                        chunks.append({
                            'content': chunk_text,
                            'headers': dict(current_headers),
                            'context_path': self._build_context_path(current_headers)
                        })
                    current_content = []
                
                # 更新标题层级
                level = len(header_match.group(1))
                title = header_match.group(2)
                current_headers[level] = title
                
                # 清除下级标题
                for l in range(level + 1, 4):
                    current_headers[l] = 
                
                current_content.append(line)
            else:
                current_content.append(line)
                
                # 如果当前内容超过最大长度，强制切分（但尽量在段落边界）
                content_so_far = '\n'.join(current_content)
                if len(content_so_far) > self.max_chunk_size:
                    chunk_text = content_so_far.strip()
                    chunks.append({
                        'content': chunk_text,
                        'headers': dict(current_headers),
                        'context_path': self._build_context_path(current_headers)
                    })
                    current_content = []
        
        # 别忘了最后一段
        if current_content:
            chunk_text = '\n'.join(current_content).strip()
            if len(chunk_text) >= self.min_chunk_size:
                chunks.append({
                    'content': chunk_text,
                    'headers': dict(current_headers),
                    'context_path': self._build_context_path(current_headers)
                })
        
        return chunks
    
    def _build_context_path(self, headers: Dict) -> str:
        构建层级路径，比如：MySQL主从切换 > 前置检查
        path_parts = [h for h in [headers[1], headers[2], headers[3]] if h]
        return ' > '.join(path_parts) if path_parts else '未分类'
    
    def enrich_chunk_with_context(self, chunk: Dict) -> str:
        
        关键技巧：给每个chunk加上上下文前缀
        这样即使单独看这个片段，也能知道它属于哪个章节
        
        context = f[文档路径：{chunk['context_path']}]\n\n
        return context + chunk['content']


# 实际使用示例
splitter = SmartDocumentSplitter(max_chunk_size=800)
chunks = splitter.split_markdown(sample_text)

print(f切分后共 {len(chunks)} 个片段\n)
for i, chunk in enumerate(chunks):
    print(f=== Chunk {i+1} ===)
    print(f路径：{chunk['context_path']})
    print(f内容预览：{chunk['content'][:150]}...)
    print()

这样切出来的效果就好多了。每个chunk开头都会带上它的位置信息，大模型在回答时能更准确地理解这段内容的上下文。

不过说实话，这个方案也不是万能的。对于那些格式不规范的老文档（没有清晰的标题结构），切分效果依然一般。后来我又针对不同类型的文档做了差异化处理，这个我们后面再说。

三、第二个大坑：向量检索的语义鸿沟

解决了切分问题，下一步就是向量化和检索了。我用的是开源的BGE模型做Embedding，用Milvus做向量数据库。

第一版的检索代码很直白：

from sentence_transformers import SentenceTransformer
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
import numpy as np

class VectorStore:
    向量存储和检索
    
    def __init__(self, model_name='BAAI/bge-base-zh-v1.5'):
        # 加载Embedding模型
        self.model = SentenceTransformer(model_name)
        self.dim = 768  # BGE base模型的向量维度
        
        # 连接Milvus
        connections.connect(default, host=localhost, port=19530)
        
    def create_collection(self, collection_name: str):
        创建集合
        if utility.has_collection(collection_name):
            utility.drop_collection(collection_name)
        
        fields = [
            FieldSchema(name=id, dtype=DataType.INT64, is_primary=True, auto_id=True),
            FieldSchema(name=content, dtype=DataType.VARCHAR, max_length=4096),
            FieldSchema(name=context_path, dtype=DataType.VARCHAR, max_length=512),
            FieldSchema(name=embedding, dtype=DataType.FLOAT_VECTOR, dim=self.dim)
        ]
        schema = CollectionSchema(fields, description=知识库文档)
        collection = Collection(collection_name, schema)
        
        # 创建索引
        index_params = {
            metric_type: COSINE,
            index_type: IVF_FLAT,
            params: {nlist: 128}
        }
        collection.create_index(embedding, index_params)
        return collection
    
    def insert_documents(self, collection_name: str, chunks: list):
        插入文档
        collection = Collection(collection_name)
        
        contents = [chunk['content'] for chunk in chunks]
        context_paths = [chunk['context_path'] for chunk in chunks]
        
        # 批量生成Embedding
        embeddings = self.model.encode(contents, normalize_embeddings=True)
        
        collection.insert([contents, context_paths, embeddings.tolist()])
        collection.flush()
        print(f成功插入 {len(chunks)} 条文档)
    
    def search(self, collection_name: str, query: str, top_k: int = 5):
        基础检索
        collection = Collection(collection_name)
        collection.load()
        
        # 生成查询向量
        query_embedding = self.model.encode([query], normalize_embeddings=True)
        
        results = collection.search(
            data=query_embedding.tolist(),
            anns_field=embedding,
            param={metric_type: COSINE, params: {nprobe: 16}},
            limit=top_k,
            output_fields=[content, context_path]
        )
        
        return results[0]