1. 为什么企业都在往 RAG 知识库上砸资源
这两年做企业服务,绕不开的一个词就是 RAG。全称 Retrieval-Augmented Generation,检索增强生成。说白了就是给大模型外挂一个"资料库",让它回答问题的时候先去资料库里翻一翻,找到相关内容再组织语言回答,而不是全靠脑子里那点训练时记住的东西。
我接触过的企业客户里,十个有八个第一句话就是"我们想把公司内部的文档用起来"。这个需求太真实了。一家中型公司,几年下来积累的产品手册、技术文档、客服话术、合同模板、会议纪要,少说几千份,多则几十万份。这些文档躺在共享盘里,找起来全靠关键词搜索,搜出来的东西还未必是想要的。新员工入职培训,光熟悉这些文档就得花一两个月。老员工遇到问题,也得在群里问一圈才知道答案在哪份文件里。
RAG 知识库解决的正是这个痛点。你把文档喂进去,它自动切分、向量化、存进向量数据库。用户用自然语言提问,系统把问题也转成向量,去数据库里找最相似的文档片段,然后把这几段内容和问题一起交给大模型,让它基于这些内容生成回答。整个过程用户感知到的就是"问了一个问题,得到了一个带出处的准确答案"。
但这里有个关键问题:向量数据库怎么选、怎么用。很多人一上来就用本地文件或者内存里的简单索引,文档少的时候没问题,一旦上到几万、几十万条向量,检索速度断崖式下跌,内存也扛不住。这时候就需要专业的向量数据库。腾讯云的向量数据库(Tencent Cloud VectorDB)是我在几个项目里实际用过的方案,它兼容 Milvus 接口,支持 Python SDK,对于已经用腾讯云生态的团队来说接入成本很低。
这篇文章我打算把基于腾讯云向量数据库搭建 RAG 企业知识库的完整流程拆开讲。从环境准备、文档处理、向量化、入库、检索到最终和大模型对接,每一步都会给出可运行的 Python 代码和参数说明。同时我会把实际踩过的坑、参数怎么调、成本怎么控这些经验一并写出来。适合正在做企业知识库的开发者、技术负责人,也适合想了解 RAG 落地细节的产品经理。哪怕你 Python 刚入门,跟着步骤走也能跑通一个最小可用版本。
2. 整体方案设计与技术选型思路
2.1 RAG 系统的核心链路拆解
一个完整的 RAG 系统,拆开来看就是两条链路:写入链路和查询链路。
写入链路负责把原始文档变成可检索的向量。流程是:文档加载 → 文本清洗 → 分块(chunking)→ 向量化(embedding)→ 存入向量数据库。每一步都有讲究。文档加载要处理 PDF、Word、Markdown、HTML 等各种格式;文本清洗要去掉页眉页脚、乱码、多余空行;分块决定了检索的粒度,块太大检索不精准,块太小上下文不完整;向量化模型的选择直接影响检索质量;入库时要带上元数据,方便后续过滤。
查询链路负责把用户问题变成答案。流程是:问题向量化 → 向量检索 → 重排序(可选)→ 组装 Prompt → 大模型生成。问题向量化必须用和入库时同一个 embedding 模型,否则向量空间对不上,检索结果就是乱的。向量检索返回 Top-K 个最相似的片段,K 值一般取 3 到 10。重排序是用一个更精细的模型对初步检索结果再排一次序,提升精度。最后把检索到的片段和用户问题拼成 Prompt,交给大模型生成回答。
这两条链路里,向量数据库是承上启下的核心组件。它决定了检索的速度、规模和稳定性。我见过不少团队前期用 FAISS 或者 Chroma 做原型,效果不错,但一到生产环境,数据量上来了,单机内存撑不住,检索延迟从几十毫秒涨到几秒,用户体验直接崩掉。所以如果目标是企业级应用,一开始就选一个支持分布式、能水平扩展的向量数据库,能省掉后面迁移的麻烦。
2.2 为什么选腾讯云向量数据库
向量数据库这个赛道,选择其实不少。Milvus、Qdrant、Weaviate、Pinecone、腾讯云 VectorDB、阿里云 DashVector 等等。我选腾讯云 VectorDB 主要基于几个实际考量。
第一是运维成本。自建 Milvus 集群,你得考虑机器、存储、网络、监控、备份、扩容,这些隐性成本很高。腾讯云 VectorDB 是托管服务,这些都不用管,控制台点几下就能创建实例,SDK 直接连。对于团队里没有专职运维的情况,这一点很关键。
第二是接口兼容性。腾讯云 VectorDB 兼容 Milvus 的 API,这意味着你之前用 Milvus 写的代码,改一下连接地址和认证信息就能迁移过来。反过来,如果你以后想从腾讯云迁到自建 Milvus,代码也不用大改。这种兼容性降低了锁定风险。
第三是生态整合。如果公司已经在用腾讯云的 CVM、COS、TI 平台,那么向量数据库和这些服务的网络互通、权限管理、计费都是打通的,接入起来很顺。特别是和 COS 对象存储配合,文档存在 COS 里,向量数据库直接读,省去了很多数据搬运的工作。
第四是性能和规模。腾讯云 VectorDB 支持亿级向量规模,单次查询延迟在毫秒级。它底层做了很多优化,比如量化索引、GPU 加速、分布式检索。对于企业知识库这种场景,几万到几百万条向量是常态,这个规模它完全能扛住。
当然也有取舍。托管服务意味着你没法直接控制底层参数,某些极端定制化的需求可能满足不了。另外数据放在云上,对于数据敏感度极高的行业,需要评估合规性。但对于大多数企业知识库场景,这些都不是问题。
2.3 技术栈全景与版本选择
把整个方案的技术栈列一下,方便你对照准备环境。
| 组件 | 选型 | 说明 |
|---|---|---|
| 编程语言 | Python 3.10+ | 3.8 也能跑,但 3.10 对类型提示和异步支持更好 |
| 向量数据库 | 腾讯云 VectorDB | 兼容 Milvus 接口,托管服务 |
| Embedding 模型 | BGE-M3 或 text-embedding-3 | 中文场景 BGE-M3 表现好,英文可用 OpenAI |
| 大模型 | DeepSeek、混元、Qwen 等 | 根据预算和合规要求选 |
| 文档解析 | PyMuPDF、python-docx、markdown | 按格式选对应库 |
| 文本分块 | LangChain TextSplitter 或自研 | 推荐按语义分块 |
| 框架 | LangChain 或 LlamaIndex | 可选,也可以纯手写 |
| 部署 | CVM 或容器服务 | 生产环境建议容器化 |
Python 版本我建议 3.10 以上。3.8 虽然还能用,但很多新库已经不支持了,而且 3.10 的 match-case 语法、更完善的类型系统,写起来舒服很多。安装 Python 直接去官网下载对应系统的安装包,Windows 记得勾选"Add Python to PATH",Mac 用 Homebrew 装最省事。
Embedding 模型这块,中文企业知识库我强烈推荐 BGE-M3。它是智源研究院出的,支持多语言、多粒度,在中文检索任务上表现很稳。而且可以本地部署,数据不出内网,对于有合规要求的场景很重要。如果预算充足且追求效果,OpenAI 的 text-embedding-3-large 也是好选择,但要注意数据出境问题。
大模型的选择要看具体场景。DeepSeek 性价比高,混元在腾讯云生态里调用方便,Qwen 开源版本可以私有化部署。如果企业对数据安全要求极高,那就选可以私有化部署的开源模型,比如 Qwen2.5 系列。
3. 环境准备与腾讯云向量数据库开通
3.1 Python 环境搭建与依赖安装
先把 Python 环境弄好。Windows 用户去 python.org 下载 3.10 或 3.11 的安装包,安装时务必勾选"Add Python to PATH",否则后面命令行里敲 python 会提示找不到命令。Mac 用户用brew install python@3.11最省心。Linux 用户可以用系统包管理器,或者用 pyenv 管理多版本。
装完验证一下:
python --version pip --version然后建一个虚拟环境,这是好习惯,避免不同项目的依赖打架:
python -m venv rag_env # Windows rag_env\Scripts\activate # Mac/Linux source rag_env/bin/activate虚拟环境激活后,安装核心依赖:
pip install tcvectordb pip install pymupdf python-docx markdown pip install langchain langchain-community pip install sentence-transformers pip install openai pip install tqdm这里解释一下每个包的作用。tcvectordb是腾讯云向量数据库的官方 Python SDK,必须装。pymupdf用来解析 PDF,比 PyPDF2 快很多,而且对中文支持好。python-docx处理 Word 文档。markdown处理 Markdown 文件。langchain提供文本分块、链式调用的工具,能省不少事。sentence-transformers用来加载本地的 BGE-M3 模型。openai库虽然名字叫 openai,但很多兼容 OpenAI 接口的大模型服务都能用它调用,包括 DeepSeek。tqdm用来显示进度条,处理大批量文档时很有用。
如果安装sentence-transformers时遇到问题,通常是 PyTorch 没装好。可以先去 PyTorch 官网根据你的 CUDA 版本选择安装命令。没有 GPU 的话装 CPU 版本也能跑,就是慢一些。
3.2 腾讯云向量数据库实例创建
登录腾讯云控制台,搜索"向量数据库 VectorDB",进入产品页面。点击"新建实例"。
配置项有几个关键选择。地域选离你服务器最近的地域,比如你的应用部署在广州,就选广州,这样网络延迟最低。实例规格根据数据量选,初期测试选最小规格就行,1 核 2G 的入门款足够跑几万条向量。等数据量上来了再扩容,腾讯云支持在线扩容,不用停机。
网络配置这里要注意。如果你用 CVM 访问,选私有网络 VPC,把向量数据库和 CVM 放在同一个 VPC 里,走内网通信,速度快还免流量费。如果从本地开发机访问,需要开通公网访问,但生产环境不建议这么做,有安全风险。
创建完成后,在实例详情页能看到访问地址和端口。还需要创建一个数据库和集合(Collection)。集合相当于关系型数据库里的表,用来存向量数据。创建集合时要指定向量维度,这个维度必须和你的 embedding 模型输出维度一致。BGE-M3 的输出维度是 1024,OpenAI text-embedding-3-small 是 1536,text-embedding-3-large 是 3072。维度填错了,后面插入数据会直接报错。
还需要获取 API Key。在"访问管理"里创建密钥,会得到 SecretId 和 SecretKey。这两个东西相当于账号密码,不要硬编码在代码里,用环境变量或者配置文件管理。
3.3 连接测试与常见报错处理
环境准备好后,先写个最小测试脚本验证连接:
import os from tcvectordb import VectorDBClient client = VectorDBClient( url=os.environ["VDB_URL"], username="root", key=os.environ["VDB_KEY"], ) db = client.database("rag_knowledge_base") print("连接成功,数据库列表:", client.list_databases())把 VDB_URL 和 VDB_KEY 设成环境变量。Linux/Mac 用export VDB_URL=xxx,Windows 用set VDB_URL=xxx,或者在代码里用 python-dotenv 加载 .env 文件。
常见的报错有这么几个。Connection refused通常是网络不通,检查 VPC 配置或者公网访问是否开通。Authentication failed是密钥错了,检查 SecretId 和 SecretKey 有没有复制错,注意前后空格。Collection not found是集合名写错了或者还没创建。Dimension mismatch是向量维度和集合定义不一致,这个最常发生在换 embedding 模型的时候,换了模型一定要同步改集合维度。
提示:腾讯云向量数据库的免费额度有限,测试阶段注意控制数据量,避免产生意外费用。可以在控制台设置消费预警。
4. 文档处理与向量化实操
4.1 多格式文档加载与清洗
企业文档格式五花八门,PDF、Word、Excel、PPT、Markdown、HTML、纯文本都有。我写了一个统一的加载器,根据文件扩展名分派到不同的解析函数。
import fitz # PyMuPDF from docx import Document import markdown from bs4 import BeautifulSoup import os def load_pdf(path): doc = fitz.open(path) text = "" for page in doc: text += page.get_text() return text def load_docx(path): doc = Document(path) return "\n".join([p.text for p in doc.paragraphs if p.text.strip()]) def load_md(path): with open(path, "r", encoding="utf-8") as f: html = markdown.markdown(f.read()) return BeautifulSoup(html, "html.parser").get_text() def load_txt(path): with open(path, "r", encoding="utf-8") as f: return f.read() LOADERS = { ".pdf": load_pdf, ".docx": load_docx, ".md": load_md, ".txt": load_txt, } def load_document(path): ext = os.path.splitext(path)[1].lower() loader = LOADERS.get(ext) if not loader: raise ValueError(f"不支持的文件格式:{ext}") return loader(path)PDF 解析是最容易出问题的。扫描件 PDF 用 PyMuPDF 提取出来是空的,因为里面是图片不是文字。这种情况需要 OCR,可以用 PaddleOCR 或者腾讯云的 OCR 服务。另外 PDF 里的表格提取出来会变成一堆乱序的文字,如果表格内容很重要,需要用专门的表格提取工具,比如 Camelot 或者 pdfplumber。
文本清洗这一步不能省。实际文档里经常有页眉页脚、页码、水印文字、多余的空格和换行。我一般用正则做一轮清洗:
import re def clean_text(text): # 去掉多余空白 text = re.sub(r"\s+", " ", text) # 去掉页码(形如 "第 3 页" 或 "- 3 -") text = re.sub(r"第\s*\d+\s*页", "", text) text = re.sub(r"-\s*\d+\s*-", "", text) # 去掉连续空行 text = re.sub(r"\n{3,}", "\n\n", text) return text.strip()清洗规则要根据实际文档调整。我建议先拿几份典型文档跑一遍,看看清洗后还剩什么噪音,再针对性加规则。不要一上来就写一堆正则,容易误伤正文内容。
4.2 文本分块策略与参数调优
分块是 RAG 里最容易被低估的环节。块分得好不好,直接决定检索质量。分块的核心矛盾是:块太大,检索出来的内容包含太多无关信息,大模型容易被干扰;块太小,上下文不完整,答案可能缺胳膊少腿。
我常用的策略是递归字符分块,优先按段落分,段落太长再按句子分,句子还长就按字符硬切。LangChain 的 RecursiveCharacterTextSplitter 就是干这个的:
from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""], length_function=len, ) chunks = splitter.split_text(clean_text)参数怎么定?chunk_size=500是我在中文企业文档场景下比较常用的值。中文一个字符大约对应 1 到 2 个 token,500 字符大概是 500 到 1000 token,这个长度既能包含一个完整的知识点,又不会太长导致检索不精准。chunk_overlap=50是块之间的重叠部分,防止一个完整的句子被切断后两边都读不懂。重叠比例一般取 chunk_size 的 10% 到 20%。
但固定大小分块有个问题:它不管语义边界。有时候一个段落正好在中间被切开,前半段在块 A,后半段在块 B,检索到块 A 的时候答案不完整。解决办法是用语义分块,根据句子之间的语义相似度来决定切分点。这需要额外的计算,但效果确实好。LangChain 有 SemanticChunker 可以用,不过对中文的支持一般,我一般是自己写一个基于标点和段落结构的启发式分块。
还有一个实践中的技巧:给每个块加上它所属文档的标题和章节路径作为前缀。比如一个块来自《产品手册》第三章第二节,就在块内容前面加上"产品手册 > 第三章 > 第二节:安装步骤"。这样检索的时候,即使块内容本身没提到"安装",但因为前缀里有,也能被检索到。这个技巧对结构化文档特别有效。
4.3 向量化模型选择与批量编码
向量化就是把文本变成一串数字。这串数字捕捉了文本的语义信息,语义相近的文本,向量距离也近。
BGE-M3 是我在中文场景的首选。它支持 8192 token 的输入长度,输出 1024 维向量,在多语言检索基准上表现很好。用 sentence-transformers 加载:
from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-m3") def embed_texts(texts, batch_size=32): embeddings = model.encode( texts, batch_size=batch_size, normalize_embeddings=True, show_progress_bar=True, ) return embeddings.tolist()normalize_embeddings=True很重要,它把向量归一化到单位长度,这样余弦相似度计算就等价于内积,检索时更快。batch_size根据你的显存调整,GPU 显存 8G 的话 32 比较稳,显存小就降到 8 或 16。
如果文档量很大,比如几十万条,本地跑 embedding 会很慢。这时候可以考虑用腾讯云的 TI 平台或者批量调用 API。也可以多进程并行,把文档分成几份,每个进程处理一份。但要注意模型加载会占显存,多进程的话每个进程都要加载一份模型,显存不够就崩了。折中方案是用一个进程加载模型,用多线程做推理,因为推理本身是 GPU 密集型的,多线程能提高 GPU 利用率。
注意:BGE-M3 首次运行会从 HuggingFace 下载模型文件,大概 2G 多。国内网络可能下载慢或者失败,可以设置镜像源,或者手动下载后指定本地路径。
5. 向量入库与检索实现
5.1 集合创建与索引配置
在腾讯云 VectorDB 里,数据存在集合(Collection)中。创建集合时要定义字段和索引。
from tcvectordb.model.collection import Collection from tcvectordb.model.index import Index, VectorIndex, FilterIndex from tcvectordb.model.enum import FieldType, IndexType, MetricType # 定义索引 index = Index() index.add(VectorIndex( "vector", dimension=1024, metric_type=MetricType.COSINE, index_type=IndexType.HNSW, params={"M": 16, "efConstruction": 200}, )) index.add(FilterIndex("doc_id", FieldType.String, IndexType.FILTER)) index.add(FilterIndex("source", FieldType.String, IndexType.FILTER)) # 创建集合 db.create_collection( name="knowledge_chunks", dimension=1024, index=index, description="企业知识库文档块", )这里几个参数解释一下。metric_type选 COSINE 余弦相似度,适合文本向量。index_type选 HNSW,这是目前最常用的近似最近邻索引,查询快、召回率高。M是每个节点的连接数,越大索引越精确但占内存越多,16 是常用值。efConstruction是建索引时的搜索范围,越大索引质量越高但建索引越慢,200 是平衡值。
FilterIndex 是标量字段索引,用来做元数据过滤。比如你只想在某个部门的文档里检索,就可以用 filter 条件过滤。这个功能在企业场景很实用,能大幅缩小检索范围,提升精度和速度。
5.2 批量插入与元数据设计
插入数据的时候,除了向量本身,还要带上原始文本和元数据。元数据的设计很关键,它决定了你后面能做哪些过滤和溯源。
def insert_chunks(collection, chunks, embeddings, metadata_list): data = [] for i, (chunk, emb, meta) in enumerate(zip(chunks, embeddings, metadata_list)): data.append({ "id": f"{meta['doc_id']}_{i}", "vector": emb, "text": chunk, "doc_id": meta["doc_id"], "source": meta["source"], "title": meta.get("title", ""), "chunk_index": i, }) # 批量插入,每批 100 条 batch_size = 100 for i in range(0, len(data), batch_size): batch = data[i:i+batch_size] collection.insert(batch) print(f"已插入 {min(i+batch_size, len(data))}/{len(data)}")元数据我一般会存这几个字段:doc_id是文档唯一标识,source是文档来源路径或 URL,title是文档标题,chunk_index是块在文档中的位置。有了这些,检索结果就能显示"来自《XX 文档》第 N 段",用户能点进去看原文,可信度大大提升。
批量插入的 batch_size 不要太大,100 到 500 比较合适。太大容易超时,太小效率低。插入过程中如果报错,通常是某条数据的向量维度不对或者字段类型不匹配,把出错的那条单独拿出来检查。
5.3 相似度检索与结果重排
检索是 RAG 的核心。用户提问,系统把问题向量化,去数据库里找最相似的 Top-K 个块。
def search(collection, query, top_k=5, filter_expr=None): query_vector = embed_texts([query])[0] results = collection.search( data=[query_vector], limit=top_k, output_fields=["text", "doc_id", "source", "title"], filter=filter_expr, ) return results[0]top_k取多少?我一般先取 10 到 20 个候选,然后用重排序模型精排,最后取前 3 到 5 个给大模型。为什么要重排序?因为向量检索是基于语义相似度的,它找的是"意思相近"的块,但不一定是最能回答问题的块。重排序模型(比如 BGE-Reranker)会同时看问题和候选块,给出一个更精准的相关性分数。
from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-v2-m3") def rerank(query, candidates, top_n=5): pairs = [(query, c["text"]) for c in candidates] scores = reranker.predict(pairs) ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) return [c for c, s in ranked[:top_n]]重排序会增加一点延迟,大概几十到几百毫秒,但精度提升很明显。如果对延迟极其敏感,可以跳过重排序,直接取向量检索的 Top-K。实测下来,加了重排序之后,答案的准确率能提升 15% 到 30%,这个投入很值。
实操心得:检索时给结果加一个相似度阈值,低于阈值的直接丢弃。比如余弦相似度低于 0.5 的块,基本和问题无关,留着反而干扰大模型。阈值多少合适,拿一批测试问题跑一下,看分数分布来定。
6. 对接大模型与完整问答链路
6.1 Prompt 模板设计与上下文组装
检索到相关块之后,要把它们和用户问题组装成 Prompt 交给大模型。Prompt 的设计直接影响回答质量。
PROMPT_TEMPLATE = """你是一个企业知识库助手。请根据下面提供的参考资料回答用户问题。 要求: 1. 只根据参考资料回答,不要编造资料中没有的信息。 2. 如果参考资料不足以回答问题,直接说"根据现有资料无法回答该问题"。 3. 回答时在句末标注引用来源,格式为 [来源: 文档名]。 参考资料: {context} 用户问题:{question} 回答:""" def build_prompt(query, chunks): context_parts = [] for i, chunk in enumerate(chunks): context_parts.append( f"[资料{i+1}] 来源:{chunk['title']}\n{chunk['text']}" ) context = "\n\n".join(context_parts) return PROMPT_TEMPLATE.format(context=context, question=query)这个模板有几个关键点。"只根据参考资料回答"是防止大模型幻觉的核心指令。大模型有个毛病,就是不知道的时候也爱编,加上这句能大幅减少胡编乱造。"标注引用来源"让回答可追溯,用户能验证。如果资料不足就直说,这比给一个错误答案强得多。
上下文长度也要控制。大模型的上下文窗口有限,虽然现在很多模型支持 128K 甚至更长,但塞太多内容进去,一是贵,二是大模型对中间部分的内容注意力会下降(所谓的"lost in the middle"现象)。我一般控制在 3000 到 5000 字符,大概 3 到 5 个块。
6.2 大模型调用与流式输出
调用大模型我用 OpenAI 兼容的接口,这样换模型只需要改 base_url 和 model 名字。
from openai import OpenAI client = OpenAI( api_key=os.environ["LLM_API_KEY"], base_url=os.environ["LLM_BASE_URL"], ) def generate_answer(prompt, stream=True): response = client.chat.completions.create( model=os.environ["LLM_MODEL"], messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=1024, stream=stream, ) if stream: for chunk in response: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content else: return response.choices[0].message.contenttemperature=0.1是让输出更确定、更少发挥。知识库问答场景不需要创意,需要准确。max_tokens=1024限制回答长度,避免大模型啰嗦。流式输出让用户能实时看到回答一个字一个字蹦出来,体验好很多,尤其是回答较长的时候。
6.3 完整问答函数与端到端测试
把上面的环节串起来:
def ask(question, top_k=10, top_n=5): # 1. 检索 candidates = search(collection, question, top_k=top_k) # 2. 重排序 chunks = rerank(question, candidates, top_n=top_n) # 3. 组装 Prompt prompt = build_prompt(question, chunks) # 4. 生成回答 answer = "" for token in generate_answer(prompt): answer += token print(token, end="", flush=True) print() # 5. 返回答案和来源 sources = list({c["title"] for c in chunks}) return answer, sources测试一下:
answer, sources = ask("公司的报销流程是什么?") print("\n来源:", sources)如果检索到的块和问题相关,大模型的回答应该准确且带引用。如果检索结果不相关,检查一下 embedding 模型是否一致、分块是否合理、相似度阈值是否合适。
7. 常见问题排查与性能优化实录
7.1 检索不准的排查思路
检索不准是 RAG 最常见的问题。表现是:用户问 A,检索出来的却是 B 相关的内容。排查按这个顺序来。
先看 embedding 模型是否一致。入库用的模型和查询用的模型必须是同一个,否则向量空间对不上。我见过有人入库用 BGE-M3,查询用 OpenAI,结果检索全是乱的。
再看分块是否合理。如果块太大,一个块里混了好几个主题,检索时匹配到的是其中一个主题,但返回的块包含大量无关内容。如果块太小,一个完整答案被切成好几块,检索只返回其中一块,答案不完整。调整 chunk_size 试试,一般 300 到 800 之间。
然后看相似度阈值。如果阈值设得太高,相关块被过滤掉了;太低,无关块混进来。拿一批测试问题,打印出检索结果的相似度分数,看看分布,再定阈值。
最后考虑加查询改写。用户的问题往往口语化、有指代,直接拿去检索效果不好。可以用大模型先把问题改写成更适合检索的形式。比如用户问"那个报销的事怎么办",改写成"公司报销流程和所需材料"。这一步能明显提升召回率。
7.2 性能瓶颈与优化手段
性能问题主要出在三个地方:embedding 计算、向量检索、大模型生成。
Embedding 计算是写入链路的主要耗时。几万条文档,本地 CPU 跑可能要几个小时。优化手段是用 GPU,或者用批量 API。如果文档更新频繁,可以做增量处理,只处理新增和修改的文档,不用每次全量重跑。
向量检索的延迟通常在几十毫秒,但如果 top_k 设得很大,或者数据量到了千万级,延迟会上升。优化手段是加标量过滤缩小范围,或者用更快的索引类型。腾讯云 VectorDB 支持调整索引参数,可以按需优化。
大模型生成是查询链路的主要耗时,尤其是生成长回答的时候。优化手段是流式输出,让用户先看到部分内容。另外可以限制 max_tokens,避免生成过长。如果并发量大,考虑用多个模型实例做负载均衡。
7.3 成本控制与增量更新
成本主要来自三块:向量数据库实例费用、embedding 计算费用、大模型调用费用。
向量数据库按规格和存储量计费。控制成本的手段是选合适的规格,不要一上来就买大的。数据量小的时候用入门款,等真的上量了再扩容。另外定期清理无用数据,比如过期的文档、重复的块。
Embedding 如果用 API,按 token 计费。本地部署的话主要是机器成本。如果文档量大且更新不频繁,本地部署更划算。如果文档更新频繁,用 API 更灵活。
大模型调用按 token 计费。控制成本的手段是优化 Prompt,减少不必要的上下文。另外可以缓存常见问题的答案,相同问题直接返回缓存结果,不用每次都调大模型。
增量更新是个实用功能。文档库不是一成不变的,每天都有新文档、修改的文档。全量重建索引成本高、耗时长。我一般给每个文档算一个哈希值,存在元数据里。更新时先算新文档的哈希,和数据库里的对比,只处理变化的文档。删除文档时,根据 doc_id 把对应的所有块删掉。
def incremental_update(collection, doc_path): doc_id = get_doc_id(doc_path) new_hash = compute_hash(doc_path) # 查已有哈希 existing = collection.query( filter=f'doc_id == "{doc_id}"', output_fields=["doc_hash"], limit=1, ) if existing and existing[0]["doc_hash"] == new_hash: print(f"{doc_path} 未变化,跳过") return # 删除旧数据 collection.delete(filter=f'doc_id == "{doc_id}"') # 处理并插入新数据 text = clean_text(load_document(doc_path)) chunks = splitter.split_text(text) embeddings = embed_texts(chunks) insert_chunks(collection, chunks, embeddings, [{"doc_id": doc_id, ...}])这套机制跑起来之后,每天定时扫描文档目录,自动同步变化,维护成本很低。
7.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决方案 |
|---|---|---|---|
| 连接超时 | 网络不通 | 检查 VPC、安全组、公网配置 | 调整网络配置或换内网访问 |
| 认证失败 | 密钥错误 | 核对 SecretId/SecretKey | 重新生成密钥 |
| 维度不匹配 | 模型换了没改集合 | 检查 embedding 维度和集合定义 | 重建集合或换回原模型 |
| 检索结果不相关 | 模型不一致/分块不合理 | 检查 embedding 模型、chunk_size | 统一模型、调整分块参数 |
| 答案编造 | Prompt 没约束 | 检查 Prompt 模板 | 加强"只根据资料回答"指令 |
| 插入慢 | batch 太小/网络差 | 看插入日志 | 增大 batch、走内网 |
| 检索慢 | top_k 太大/数据量大 | 看查询耗时 | 减小 top_k、加过滤条件 |
| 内存溢出 | 模型太大/数据太多 | 看资源监控 | 换小模型、分批处理 |
避坑技巧:每次换 embedding 模型,一定要重建集合。我见过有人换了模型没重建,检索结果全是乱的,排查了半天才发现是维度对不上但没报错(因为新模型维度恰好也是 1024),但向量空间完全不同,检索自然不准。
8. 几个实际项目中的经验补充
8.1 知识库和结构化数据的边界
RAG 知识库擅长处理非结构化文本,但企业里还有很多结构化数据,比如数据库里的订单、Excel 里的报表。这些数据用 RAG 处理效果不好,因为向量检索不擅长精确匹配和聚合计算。用户问"上个月销售额是多少",向量检索找不到答案,因为答案不在任何一段文本里,而是需要查数据库算出来。
我的做法是把结构化查询交给专门的工具。用户问题先经过一个路由判断,如果是结构化查询,就走 SQL 或者 API;如果是文档问答,才走 RAG。这个路由可以用大模型来做,给它几个示例,让它判断问题类型。这样两套系统各司其职,效果比硬塞进 RAG 好得多。
8.2 多轮对话与上下文管理
企业知识库往往需要多轮对话。用户问"报销流程是什么",得到回答后接着问"那需要哪些材料",这时候"那"指代的是报销流程,如果直接把"那需要哪些材料"拿去检索,肯定找不到相关内容。
解决办法是在检索前做查询改写,把多轮对话的上下文考虑进去。把最近几轮对话和当前问题一起交给大模型,让它改写出一个独立的、完整的查询。比如把"那需要哪些材料"改写成"公司报销流程需要哪些材料"。然后再用改写后的查询去检索。
def rewrite_query(history, current_question): prompt = f"""根据对话历史,将用户的最新问题改写成一个独立的、完整的查询语句。 对话历史: {history} 最新问题:{current_question} 改写后的查询:""" return call_llm(prompt)这个改写步骤会增加一次大模型调用,但能显著提升多轮对话的检索质量。如果对延迟敏感,可以只在检测到指代词("那"、"它"、"这个")的时候才触发改写。
8.3 效果评估与持续迭代
RAG 系统上线不是终点,而是起点。你需要一套评估机制来持续监控和优化效果。
我一般会建一个测试集,包含几十到几百个问题和对应的标准答案。每次调整参数或者换模型,都跑一遍测试集,看准确率、召回率、回答质量的变化。评估指标包括:检索命中率(相关块是否在 Top-K 里)、答案准确率(回答是否正确)、引用准确率(引用的来源是否真的支持答案)。
人工评估也很重要。定期抽样看一些真实用户的问答记录,看看哪些问题回答得好,哪些不好。不好的案例分析原因:是检索没找到、还是找到了但大模型没用好、还是问题本身超出知识库范围。针对性地优化。
这套流程跑起来之后,知识库的效果会随着时间越来越好。我做过的一个项目,上线三个月,经过几轮迭代,答案准确率从最初的 60% 多提升到了 85% 以上。
8.4 安全与权限控制
企业知识库往往涉及敏感信息,不同部门、不同职级的员工能访问的文档范围不同。这个权限控制要在检索层面做,不能只在前端做。
我的做法是在元数据里加一个access_level字段,检索时根据当前用户的权限加过滤条件。比如普通员工只能检索access_level <= 2的文档,经理能检索access_level <= 3的。这样即使前端被绕过,后端检索也不会返回越权内容。
def search_with_permission(collection, query, user_level): filter_expr = f"access_level <= {user_level}" return search(collection, query, filter_expr=filter_expr)另外,用户的问题和检索结果里可能包含敏感信息,日志记录时要注意脱敏。大模型调用如果走外部 API,要评估数据出境合规性。对于数据敏感度高的场景,建议用私有化部署的模型,数据不出内网。
这套基于腾讯云向量数据库的 RAG 知识库方案,我在几个项目里实际跑过,从几千条到几十万条向量都验证过。核心链路不复杂,难的是细节调优和持续运营。把文档处理、分块、检索、Prompt 这几块的参数调好,效果就能达到可用水平。后面再根据实际使用中的反馈慢慢迭代,逐步提升。
最后分享一个我踩过的坑:不要一上来就追求大而全。我见过团队花几个月搭了一套复杂的系统,结果因为文档质量差、分块不合理,效果还不如人工搜索。正确的做法是先拿一个部门、一类文档做小范围试点,跑通链路,验证效果,再逐步扩大范围。小步快跑,比一步到位靠谱得多。