做RAG(检索增强生成)项目,绕不开的就是“查得出来”这步。我接过不少类似需求,模型接了、Prompt也写了,最后效果还是飘,问题多半出在检索链路没做扎实:文档没切好、向量化不准、向量库存取混乱、相似度算出来也不知道怎么用。这篇内容我打算用一套可以直接跑的代码,把向量、余弦相似度、ChromaDB、阿里云Embedding这几个RAG核心组件串起来,跑通“文档进得去、查得出、排得对”的最小闭环。适合刚接触RAG的开发者,也适合想快速验证检索方案的业务项目参考。
1. 先拆散再组装:RAG检索链路里的四个角色
1.1 大模型的短板和RAG的思路
大模型本身对私有文档几乎一无所知。你辛辛苦苦整理的内部FAQ、产品资料、竞品分析,如果直接丢给模型,它只能凭训练时见过的泛化知识回答,答错、编造、答非所问都太常见了。这就是所谓的知识截止和幻觉问题,也是RAG存在的根本原因。
RAG的思路不是去重新训练模型,而是在生成之前先把“答案素材”找出来。用户问一个问题,系统先去知识库里检索和这个问题最相关的几段内容,再把这几段拼进Prompt一起发给大模型。模型看到的是“资料+问题”的组合,而不是光秃秃的题目,输出自然稳得多。这个思路的关键就落在检索质量上——查不到相关内容,后面生成再强也没用。
所以一个最小可用的RAG系统,核心就是把四个东西串起来:Embedding负责把文字变成向量,向量库负责存和查,相似度算法负责判断哪些内容最相关,最后交给大模型生成答案。四个环节各有各的坑,下面逐个说清楚。
1.2 四个组件到底各干什么
先说Embedding。一个Embedding服务可以把一段文本转换成一串固定长度的浮点数,比如1024维的向量。这串数字不是随机生成的,它把文本的语义信息压缩到了向量空间里:语义相近的文本,向量距离就近;语义无关的,向量距离就远。这个“语义距离”是RAG的基石。
再说余弦相似度。两个向量之间怎么算“近”?最常用的指标是余弦相似度,公式是:
cosine_similarity = A·B / (|A| × |B|)也就是两个向量的点积除以各自的模长。结果范围在-1到1之间,越接近1表示方向越一致。用在语义检索里,就是越接近1越相关。ChromaDB内部默认用的就是余弦距离,它返回的distance实际等于1 - cosine_similarity,所以返回的distance越小,相似度反而越高,这个很多人初次接触会搞反。
然后说ChromaDB。这是一个轻量级的向量数据库,支持本地持久化,不需要单独起服务,写几行代码就能把向量和原始文本一起存进去。它解决的是“索引和管理”问题:你不可能每次查询都把所有向量拉出来挨个算相似度,向量库通过内置的ANN索引(近似最近邻搜索)大幅度缩短检索时间。
最后是阿里云Embedding这个具体服务。我这次选它,主要原因是中文效果好、接口简单、开通就能用。尤其在中文业务场景里,本地跑一个小模型往往效果打折,用云服务可以直接拿到质量比较稳定的向量化结果,减少一个变量。
这四个角色合起来,就是一条完整的检索链路:文档切分→逐段Embedding→存入ChromaDB→查询时把问题也Embedding→从ChromaDB里用余弦距离检索Top-K→拼接上下文交给LLM。接下来我先说准备阶段的事,再一步步把代码写明白。
2. 实操前的准备:环境、密钥和向量库模式
2.1 依赖安装与Python版本
这套链路其实没有特别重的依赖,我用的是Python 3.10,后面运行一直很稳。核心只需要两个包:chromadb和dashscope。前者是向量库,后者是阿里云Embedding的SDK。
命令很简单:
pip install chromadb dashscope如果是国内网络环境,可以加清华镜像源加速:
pip install chromadb dashscope -i https://pypi.tuna.tsinghua.edu.cn/simple如果你后面还想顺手做文本切分,可以再装一个langchain-text-splitters,只用它里面的RecursiveCharacterTextSplitter。不过我先说明,这个不是必须的,后面我会用一段不依赖LangChain的简单切分逻辑,先把链路跑起来再说。
注意:
chromadb在安装时会带一些依赖,包括numpy、onnxruntime等,如果机器比较旧,建议装完先执行一下chromadb --version确认安装成功。
2.2 开通阿里云Embedding并拿到密钥
阿里云的Embedding模型目前挂在百炼平台上(DashScope是它的服务模型名),你需要先注册并开通百炼服务,然后在控制台里创建一个API Key。
拿到Key之后,我建议直接写成环境变量,不要在代码里硬编码,避免仓库提交时泄漏:
export DASHSCOPE_API_KEY="你的API_KEY"Windows下对应的写法是set DASHSCOPE_API_KEY=你的API_KEY。
模型名我用的text-embedding-v3,默认输出维度是1024维,也可按需改成512或2048,向量维度越高通常语义表达越细腻,但存储和计算开销也会上去。对大多数知识库检索场景,1024维已经够用。
这里有个细节:同一个Collection里所有向量的维度必须一致。也就是说你如果中途改了dimension参数,之前存进去的向量就不能继续用了,需要重新生成或新开一个Collection。
2.3 ChromaDB两种模式怎么选
ChromaDB有两种使用方式:一个是以PersistentClient方式嵌入到你的应用进程里,数据落盘到本地目录;另一个是以HttpClient连接一个独立的Chroma服务端。
我推荐本地优先。原因很简单:省事、快、不容易出网络问题。对一个几十万条数据级别的内部知识库,本地PersistentClient完全扛得住。等到数据量大到需要多机共享或跨服务访问时,再换成服务端模式也不迟。
import chromadb client = chromadb.PersistentClient(path="./chroma_db") collection = client.get_or_create_collection( name="knowledge_base", metadata={"hnsw:space": "cosine"} )path是数据持久化目录,首次运行会自动创建。metadata={"hnsw:space": "cosine"}这行很多人会漏掉,它显式指定索引的距离算法为余弦距离。如果你不写,Chroma默认是l2(欧氏距离),虽然也能用,但和余弦相似度表达的意义有偏差,我还是建议一开始就明确指定。
另一个容易踩的坑:如果你在创建Collection时不传入embedding_function,Chroma会尝试用自己内置的默认Embedding模型,那个模型首次运行会联网下载,网络受限时直接报错。我们既然用阿里云的Embedding,就应该在后续add和query时显式传入向量,绕开默认模型。这一点后面代码里会体现。
3. 从文本到向量的完整入库流程
3.1 文本切分:先解决“粒度”问题
向量化不能把整本几十页的手册直接算成一个向量,那样信息都糊在一起了。检索之前,必须把长文档切成一段段适合检索的文本块,专业说法是chunk。
为什么切分?两个原因。第一是Embedding模型对输入长度有限制,text-embedding-v3单次输入长度虽然能到几千个token,但太长时尾部信息很容易被稀释掉。第二是检索粒度问题:用户问一个具体问题,你希望召回的是“某一段相关描述”,而不是一整章泛泛的介绍。
我的切分策略分两种情况。如果文本本身有清晰的结构,比如Markdown标题、列表、PC的段落标记,那就按标题和段落切开,每个标题下的一段内容作为一条。如果没有结构,就用固定窗口长度切,比如每500个字符一块,块与块之间重叠50个字符,避免一句话被拦腰斩断在中间导致语义残缺。
简单实现一个切分函数:
def split_text_simple(text, chunk_size=500, overlap=50): step = chunk_size - overlap chunks = [] for i in range(0, len(text), step): chunks.append(text[i:i + chunk_size]) return chunks这段代码适合快速验证。真实项目里建议升级成按结构切分,或者用RecursiveCharacterTextSplitter,它能根据分隔符优先级递归去切,尽量保持段落完整。我实际项目里踩过纯字符切分的坑:一个“根据《合同法》第54条”的句子被切成两半,检索时后半截没有前文,送到模型那里根本看不懂。所以有结构的文本一定要先按结构切,不要图省事。
3.2 调用Embedding接口生成向量
切好文本块之后,每块都要调一次Embedding接口。我是用dashscopeSDK调用的,代码很直接:
import os from dashscope import TextEmbedding def get_embeddings(texts): resp = TextEmbedding.call( model='text-embedding-v3', input=texts, dimension=1024, api_key=os.getenv("DASHSCOPE_API_KEY") ) if resp.status_code != 200: raise RuntimeError(f"Embedding调用失败: {resp.code} {resp.message}") return [item['embedding'] for item in resp.output['embeddings']]这里注意几个细节。input参数接收的是一个字符串列表,SDK会一次把多条文本一起发出去。text-embedding-v3对单次请求的文本条数有限制,我习惯一次传5条以内,控制请求体的体积,降低超时概率。如果文档量大,就分批循环处理。
批量处理时我还会加一点延迟,避免瞬间请求太多被限流。虽然DashScope的限流阈值对普通项目来说不低,但处理几千个chunk时还是稳一点好:
import time def embed_batch(texts, batch_size=5): all_vectors = [] for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] vectors = get_embeddings(batch) all_vectors.extend(vectors) time.sleep(0.1) print(f"已处理 {min(i + batch_size, len(texts))}/{len(texts)}") return all_vectors这里有个性能和可靠性的权衡点:每次调用有网络开销,批量太小会慢,批量太大会超时。5条一批是我多次试验下来比较均衡的值,你可以按你的网络情况微调。如果某一次调用失败,最简单的做法是抛异常后从断点重试,因为我们会在落库前把向量和文本对应关系先准备好,重试不会造成脏数据。
3.3 写入ChromaDB并持久化
向量生成完之后,就是入库。这里必须是“文本、向量、元数据”三样一起写,不然后面查出来只有向量没有原文,也没法拼给大模型。
我的入库逻辑是这样的:
def build_knowledge_base(chunks, metadatas, collection): ids = [f"chunk_{i}" for i in range(len(chunks))] embeddings = embed_batch(chunks) collection.add( ids=ids, documents=chunks, embeddings=embeddings, metadatas=metadatas )ids必须是非空字符串,而且不能重复。metadatas建议至少包含两个字段:来源文件名和块序号。这样一来,后面如果发现某条检索结果质量有问题,能直接定位到是哪个文件的第几段,排查效率会高很多。
metadatas = [ {"source": "产品手册.md", "chunk_index": i} for i in range(len(chunks)) ]写库的时候有个隐藏问题:如果你在一个长期运行的项目里重复执行build_knowledge_base,同一个chunk_id会被重复添加,Chroma会直接报ID already exists。所以增量更新时优先用collection.upsert而不是add:
collection.upsert( ids=ids, documents=chunks, embeddings=embeddings, metadatas=metadatas )upsert的存在就是用来处理这个场景的:id不存在就插入,存在就覆盖。我后来更新的知识库小工具里,默认都用upsert。
还有一点:Collection一旦创建,它内部保存的向量维度就被固定了。你如果第一批文本用1024维向量写入,第二批突然换成512维,Chroma会报维度不一致。这个错误信息挺明显,但如果没排查到维度这一步,可能还会怀疑是数据问题。入库前统一好维度,能省很多时间。
4. 查询:相似度计算与Top-K检索
4.1 一条query如何变成可检索的向量
入库之后就是查询。查询的流程和入库其实是镜像关系:用户输入一个问题,先调用同一个Embedding模型把问题转成向量,然后用这个向量去ChromaDB做相似度检索。
def search(query_text, collection, top_k=5): query_vector = get_embeddings([query_text])[0] results = collection.query( query_embeddings=[query_vector], n_results=top_k, include=["documents", "distances", "metadatas"] ) return results注意这里查询也必须用同一个模型。如果你入库用text-embedding-v3,查询却换成别的模型,向量空间的语义对齐就乱了,检索结果会完全不可信。这点看似基础,但我真见过因为换Embedding模型导致检索质量暴跌的案例。
collection.query返回的results是个字典,其中documents是命中的文本列表,distances是距离分数列表,metadatas是对应的元数据列表。distances就是我们之前说的余弦距离,值越小越相似。
4.2 余弦相似度的结果怎么看
ChromaDB在显式指定"hnsw:space": "cosine"后,返回的distance就等于1 - cosine_similarity。所以:
- 两个向量完全同方向,相似度为1,distance为0。
- 两个向量完全无关,相似度接近0,distance接近1。
- 两个向量方向相反,相似度为-1,distance为2。
实际跑出来的分数不会那么极端,一般相同的领域文本相似度在0.75以上,无关内容可能只有0.3左右。我常用的换算方式:
similarities = [1 - d for d in results["distances"][0]]然后就可以直接看相似度去设置阈值。比如我习惯把阈值设在0.65左右,低于这个分数就不拼进Prompt,宁可告诉用户“知识库里没有相关内容”,也不要硬凑一段不相关的内容让模型瞎编。
如果你不放心Chroma返回的distance,可以自己拿向量手动算一遍余弦相似度做验证。拿numpy来实现:
import numpy as np def cosine_similarity(vec_a, vec_b): a = np.array(vec_a, dtype=float) b = np.array(vec_b, dtype=float) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))我调试阶段经常用这个方法比对结果,能确认Chroma的索引和距离计算没有偏差。这个步骤在项目初期很有价值,因为如果连“相似度计算”这个底层环节都没验证过,后面叠加LangChain、Prompt模板、生成逻辑之后,出问题你都不知道从哪里查起。
4.3 把检索结果拼给大模型
检索只是前半场,后半场是把结果拼成上下文交给大模型。我自己习惯用一个标准模板把多条文档拼接起来:
def build_prompt(query, docs): context = "\n\n".join( f"[{meta.get('source', '未知来源')}] {doc}" for doc, meta in zip(docs["documents"][0], docs["metadatas"][0]) ) prompt = f"""请根据以下资料回答问题。 资料: {context} 问题:{query} 要求: 1. 优先使用资料中的事实回答。 2. 如果资料中没有答案,直接回答“知识库中没有找到相关内容”。 3. 不要编造资料中不存在的信息。""" return promptPrompt模板里我刻意写了“如果资料中没有答案,直接回答不知道”。如果不加这个约束,即使检索结果不相关,模型也会强行根据模糊印象作答,幻觉立刻回来。加了这个约束,虽然不能百分百杜绝,但至少给了模型一个明确的兜底出口。
拼好Prompt之后,就可以接任意大模型了。我用得最多的是阿里云百炼上的qwen-plus,当然你也可以用任何OpenAI兼容接口:
from openai import OpenAI client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) resp = client.chat.completions.create( model="qwen-plus", messages=[{"role": "user", "content": build_prompt(query, results)}] ) print(resp.choices[0].message.content)到这里,一条完整的RAG链路就跑通了:文档切分、向量化、入库、检索、拼Prompt、生成答案。很多初学者把注意力全放在生成那一步,但我的经验是,前面检索环节的调试价值远高于后面。只要检索质量过关,Prompt稍微写写效果都不会差。
5. 高频踩坑与排查实录
5.1 常见问题速查表
以下是我在实际使用中遇到过的典型问题,按现象整理成了速查表。
| 现象 | 可能原因 | 解决办法 |
|---|---|---|
ID already exists异常 | 重复add相同id的chunk | 改用collection.upsert |
| 查询结果明显不相关 | 入库和查询用的Embedding模型不一致 | 统一为同一个模型和dimension |
| 首次导入时报下载失败 | 没传embeddings,Chroma想下载默认Embedding模型 | 调用add/query时显式传入向量 |
dimension mismatch错误 | 新旧数据向量维度不一致 | 统一dimension,或新建Collection重新入库 |
| 中文长文本被截断 | 固定字符切分把句子切开了 | 按结构切分,或使用RecursiveCharacterTextSplitter |
| API调用频繁超时 | 单次请求文本条数太多 | 一次传5条以内,增加sleep间隔 |
| 检索分数普遍偏低 | chunk_size太小,单条信息量不足 | 适当放大chunk_size,加入重叠窗口 |
| 答案生成仍胡说 | 没有在Prompt里要求“资料中没有就不要答” | 增加兜底约束 |
这个表格看着简单,但每一条背后都是实际跑偏过才总结出来的。拿“Chroma想下载内置模型”那条来说,我第一次跑的时候完全没注意,结果没有网络的环境里直接卡死,后来才发现需要显式传向量。这些坑文档里都有写,但不踩过就是记不住。
5.2 检索质量调优的几个方向
如果你的RAG系统检索质量不够好,先别急着换大模型,按下面顺序排查。
第一个是chunk大小。很多项目把文档切得很碎,每块只有一两句话,结果单个向量语义过于单薄,检索时分数普遍上不去。我一般做中文场景初始值设在400到800字之间,然后看具体业务再微调。如果问题偏向“答案都能找到但不够精确”,就调小chunk并增加重叠;如果偏向“经常找不到”,就调大chunk让语义更完整。
第二个是top_k选择。top_k=3和top_k=5的差异不只是数量,还会影响大模型的注意力分配。我多数项目用5,如果答案总是不准,降到3试试,让模型聚焦在最相关的段落上。别把top_k设到10以上,Retrieval结果太多会把真正相关的内容淹没掉。
第三个是阈值过滤。很多项目把Top-K的结果无条件拼进Prompt,完全不看相似度分数。我的建议是设置阈值,比如similarity >= 0.65才使用,低于阈值直接不检索。这个做法看似简单,却能显著减少无关上下文对模型的干扰。
第四个是metadata过滤。如果你的知识库包含多类文档,比如服务条款、操作指南、FAQ,可以在query时用where参数限定来源:
collection.query( query_embeddings=[query_vector], n_results=5, where={"source": "操作指南.md"} )这样做的好处是精确命中,不用靠相似度去区分不同主题的文档,检索准确率能明显提升。
5.3 最后的调试心得
调试这套RAG链路,我的顺序永远是:先验证Embedding、再验证检索、最后再优化生成。具体做法是,先准备几条测试文本,把它们的向量打出来确认不是全零向量,再手动计算余弦相似度和Chroma返回的distance比对,确保底层逻辑一致。然后构造一个“正确答案明显在知识库里”的问题,看能不能查出来。如果这一步过了,系统的地基就算稳了,再往后接Prompt和生成才有意义。
我还会把每次查询的query、命中的chunk、相似度分数、最终答案完整打日志。一旦线上回答有问题,回溯日志就能立刻定位是检索没查到、排序不对、还是模型生成飘了。这个习惯帮我在一个文档检索项目里快速定位到问题根源,那次就是用户输入的提问方式太口语化,documents里优势词不多,Embedding结果偏离了预期。日志里看到相似度分数全在0.3以下,一下就判断出是检索问题,而不是生成问题。
另外一个小技巧:知识库更新时不要把所有chunk重新向量化一遍,那样成本太高。只需要对新增和修改的文档做切分、Embedding,然后调用upsert覆盖对应的id。ChromaDB会保留旧的索引结构,你只需要保证id策略唯一且稳定。
我自己做完这套实操下来最大的体会是,RAG一点都不玄乎,本质上就是一个“先查对资料,再让模型回答”的工程问题。只要把切分粒度、向量质量、相似度阈值这几个关键参数调试到位,即使不接复杂的RAG框架,效果也足够稳。下一步如果你想继续扩展,可以往混合检索、结果重排、多路召回方向走,但那都是在这个地基上做加法了。