☰
1.RAG、文档切分、向量化、向量数据库
2026/11/3 10:47:38 网站建设 项目流程

前言

如果你已经用过大模型,却总遇到“知识过时、答不出私有文档、一本正经胡说”这几类问题,那么 RAG(检索增强生成)就是最值得先掌握的一条工程路径。本文面向有一定 Python 基础、希望把大模型真正落到业务里的开发者与算法入门者。

1.什么是RAG

RAG(检索增强生成)技术是一种结合信息检索与生成模型的新型架构,其核心思想是利用外部知识库或文档集合为大模型提供实时、准确的背景信息,从而弥补大模型的局限性。

RAG由两部分组成:

  • 检索模块:在知识库中检索与当前输入问题相关的文档或片段
  • 生成模块:基于检索结果和原始输入,通过大模型生成准确、丰富的回答

RAG 是“先去书架翻书,再把书页和问题一起递给考生”,不是“让考生自己跑图书馆”。

问题 ─→ [Embedding 模型] ─→ 向量 ─→ [向量库搜索] ─→ 片段
↓
[LLM] ←── 组装好的 prompt(问题+片段)
↓
回答

三个关键理解点:

  1. RAG 的本质是“开卷考试”——模型本身不变,只是每次答题前先把相关参考材料递到它面前。
  2. 向量数据库是两条链路的交汇点:左边进(建库)、下面出(检索),按“语义相似度”而非关键词匹配,所以换个说法也能查到。
  3. 解决了大模型的三个老毛病:知识过时(不用重新训练)、私有知识(您自己的文档能查)、一本正经胡说(答案能附引用来源可追溯)。

2.文档切分

方式原理优点缺点适用
1. 固定长度按固定字符/token 数硬切,加 10–20% 重叠快、简单、可预测会把句子/表格拦腰切断格式乱的纯文本兜底
2. 递归字符(最常用)按\n\n → \n → 句号 → 空格逐级回退切尊重自然边界,块大小可控对结构化文档仍会切散默认首选,RecursiveCharacterTextSplitter
3. 按结构切按 Markdown 标题 / HTML 标签 / JSON 层级切块自带上下文(继承父级标题)依赖文档格式规范手册、wiki、API 文档;MarkdownHeaderTextSplitter
4. 代码专用按函数/类边界切(按语言定义分隔符)代码语义完整仅适用于代码代码库问答;Language.PYTHON等
5. 语义切分相邻句子算 embedding 相似度,相似度骤降处下刀主题不跑偏,检索最准要多跑一次 embedding,贵、慢高质量问答、付费知识库
6. LLM 切分 / Agentic让模型自己判断主题边界和块大小效果最好,能带摘要元数据成本最高、最慢高价值文档、量不大

三个实操要点:

  1. 重叠(overlap)必须有:相邻块重叠 50–100 token,防止关键句正好被切在缝里。
  2. 块大小经验值:256–512 token 起步;问答类偏小(精确命中),摘要类偏大(上下文足)。
  3. 切块决定 RAG 上限:检索和生成再好,块切烂了(一句话断成两截)也白搭——这也是为什么吴恩达课程把 splitter 放在向量库前面讲。
#按句子切分 import re text = """在这里,我们有一段超过200字的中文文本作为输入例子。这段文本是关于 自然语言处理的简介?自然语言处理(NLP)是人工智能的重要分支!它研究计算机 如何理解、生成人类语言……在RAG系统中,文档需要先被切分成小块,再做成向量存进数据库。""" def split_by_sentence(text): # 正则捕获组 ( ) 包住标点:split 后标点会保留在结果里 # 结果形如 [句子1, '。', 句子2, '?', 句子3, '!', ..., 最后一截] sentences = re.split(r'([。!?!?…])', text) # sentences 下标偶数位是句子、奇数位是标点,zip 两两配对拼回去 # 尾部补 '' 对齐,防止最后一句没标点时被 zip 丢弃 chunks = [ sentence + (punctuation or '') for sentence, punctuation in zip(sentences[::2], sentences[1::2] + ['']) ] return [c for c in chunks if c.strip()] # 去掉空块 result = split_by_sentence(text) for i, c in enumerate(result, 1): print(f"句 {i}: {c}") #刚切分完 #sentences = ['自然语言处理很好玩', '。', '你学过NLP', '?', '真的'] ↑偶数位 ↑奇数位 ↑偶数位 ↑奇数位 ↑偶数位(最后一句没标点) #sentences[::2] 取所有句子 #['自然语言处理很好玩', '你学过NLP', '真的'] # 下标 0, 2, 4 #sentences[1::2] 取所有标点 #['。', '?'] #zip 是 Python 内置函数:把几个列表“并排拉链式”配对,打包成一组组的元组。 #names = ['张三', '李四', '王五'] #numbers = ['13800000001', '13800000002', '13800000003'] #zipped = zip(names, numbers) #[('张三', '13800000001'), ('李四', '13800000002'), ('王五', '13800000003')]
# 调用 RecursiveCharacterTextSplitter,实现递归切分 from langchain_text_splitters import RecursiveCharacterTextSplitter text = """在这里,我们,有一段超过200字的中文文本,作为,输入例子。 这段文本是关于自然语言处理的简介?自然语言处理(NLP)是计算机科学、 人工智能和语言学的交叉领域,它旨在让计算机能够理解和处理人类语言。 在这一领域中,机器学习技术扮演着核心角色,通过使用各种算法、模型, 计算机可以解析、理解、甚至生成人类语言。 这一技术已广泛应用于机器翻译、情感分析等多个方面,随着深度学习技术的发展""" def split_by_recursive_char(text): splitter = RecursiveCharacterTextSplitter( chunk_size=20, # 每块最长 20 字符 chunk_overlap=5, # 相邻块重叠 5 字符 separators=["\n\n", "\n", "。", ",", "、", "?", "!", ""], # 逐级回退 ) return splitter.split_text(text) if __name__ == "__​main__": chunks = split_by_recursive_char(text) for i, c in enumerate(chunks, 1): print(f"块 {i}({len(c)} 字符): {c}") #块 1(17 字符): 在这里,我们,有一段超过200字的 #块 2(9 字符): 中文文本,作为,输入例子 #块 3(18 字符): 这段文本是关于自然语言处理的简介 #块 4(16 字符): 自然语言处理(NLP)是计算机科学 #块 5(11 字符): 人工智能和语言学的交叉领域 #块 6(17 字符): 它旨在让计算机能够理解和处理人类语言 #块 7(7 字符): 在这一领域中 #块 8(14 字符): 机器学习技术扮演着核心角色 #块 9(10 字符): 通过使用各种算法、模型 #块 10(14 字符): 计算机可以解析、理解、甚至生成人类语言 #块 11(18 字符): 这一技术已广泛应用于机器翻译、情感分析 #块 12(11 字符): 等多个方面,随着深度学习技术的发展

3.向量化(Embedding)

把一段文字变成一串有意义的数字——这段数字就是向量。关键在于“意思相近的文字,向量在空间里就挨得近”:

文本向量(示意,实际 64~1536 维)
"Teplizumab 能治什么病"[0.12, -0.87, 0.45, ...]
"这种药治疗 1 型糖尿病"[0.11, -0.85, 0.47, ...]←很近
"今天中午吃什么"[-0.92, 0.33, -0.61, ...]←很远

距离用余弦相似度算:把两个向量当成两个箭头,余弦相似度 = 两个箭头夹角的余弦值——只看方向,不看长度。

# -*- coding: utf-8 -*- # 文本向量化.py —— 阿里云百炼(DashScope)OpenAI 兼容模式 import os from openai import OpenAI client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), # 如果没有配置环境变量,请在此处用您的API Key进行替换 base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 百炼服务的base_url ) completion = client.embeddings.create( model="text-embedding-v4", input=[ "免费资源,全部免费无套路", "今天天气真好,适合出去玩", "RAG 先把文档切块,再向量化存库", ], dimensions=64, # 指定向量维度(仅 text-embedding-v3 / v4 支持该参数) encoding_format="float", ) 取出向量 for item in completion.data: vec = item.embedding print(f"下标 {item.index}:维度={len(vec)},前 6 个分量={[round(x, 4) for x in vec[:6]]}") #下标 0:维度=64,前 6 个分量=[0.0213, -0.1847, 0.0952, -0.0311, 0.2208, -0.1476] #下标 1:维度=64,前 6 个分量=[-0.0894, 0.1203, -0.0427, 0.1906, 0.0518, 0.2331] #下标 2:维度=64,前 6 个分量=[0.0308, -0.1651, 0.1074, -0.0127, 0.1983, -0.1204]

4.向量数据库

向量化数据库(向量库)总结——专门存“向量”并支持“相似度搜索”的数据库。

它解决什么问题:

普通数据库(MySQL)只能精确匹配:WHERE name = '张三'。向量库回答的是另一种问题:“和这个向量最像的 K 个是谁”——语义搜索,没有精确等号可言。

核心工作流:

入库:文本 → 切块 → Embedding 模型转向量 → 存入向量库(带原文和元数据)
查询:问题 → 同一 Embedding 模型转向量 → ANN 检索 Top-K → 返回原文片段

内部三大件:

组件干什么关键词
向量存储存高维浮点数组 + 原文 + 元数据(章节、来源、时间)元数据可再过滤
ANN 索引不逐个比对,走图索引快速逼近HNSW(主流)、IVF、PQ 压缩
相似度计算余弦 / 点积 / 欧氏通常建库时指定,不能中途换

必须知道的四个坑:

  • 建库和查询必须同一 Embedding 模型——换了模型全库作废,重算;
  • 维度建库时定死——dimensions=64存的库,塞不进 1536 维的新向量;
  • 向量库不存储“关系”——复杂过滤(“价格<500 且相似度高”)要靠元数据过滤 + 向量检索混合查询,选型时确认产品支持;
  • 删改文档要同步删向量——忘了就是“幽灵知识”,模型还在引用已删除的旧文档。
# -*- coding: utf-8 -*- # Chroma 向量数据库使用 —— 连接器封装 import os import uuid from openai import OpenAI import chromadb # ---------- Embedding:用百炼把文本转向量 ---------- client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) def get_embeddings(texts): """批量转向量,返回 List[List[float]]""" resp = client.embeddings.create( model="text-embedding-v4", input=texts, dimensions=64, encoding_format="float", ) # 按下标排序,保证和输入顺序一致 return [d.embedding for d in sorted(resp.data, key=lambda x: x.index)] class MyVectorDBConnector: def __init__(self, collection_name="demo"): # 内存模式的 Chroma 客户端(数据存内存,进程结束即消失) self.client = chromadb.Client() # 若要持久化到磁盘,改用:chromadb.PersistentClient(path="./chroma_data") self.collection = self.client.get_or_create_collection(name=collection_name) # 添加文档与向量 def add_documents(self, documents): '''向 collection 中添加文档与向量''' embeddings = self.get_embeddings_batch(documents) self.collection.add( embeddings=embeddings, # 每个文档的向量 documents=documents, # 文档的原文 ids=[f"id{i}" for i in range(len(documents))], # 每个文档的 id # metadata=[...] # 可选:给每个文档附加元数据,如 {"来源": "第一章"} ) # ids 也可以用 uuid.uuid4() 自动生成 print("self.collection.count():", self.collection.count()) def get_embeddings_batch(self, texts, batch_size=10): """分批转向量(百炼单次最多 10 条文本,超了要拆批)""" result = [] for i in range(0, len(texts), batch_size): result += get_embeddings(texts[i:i + batch_size]) return result # 检索向量数据库,返回 top_k 个相似高的记录 def search(self, query, top_k): '''检索向量数据库 query 是用户的查询 top_k 指查出 top_k 个相似高的记录 ''' results = self.collection.query( query_embeddings=get_embeddings([query]), # 查询文本也要转向量 n_results=top_k, ) #results:{ids:[],distances:[],documents:[]} # 整理输出:原文 + 相似度距离(distance 越小越相似) return [ {"text": doc, "distance": dist, "id": id_} for doc, dist, id_ in zip( results["documents"][0], results["distances"][0], results["ids"][0], ) ] # ---------- 演示 ---------- if __name__ == "__&#8203;main__": db = MyVectorDBConnector("teplizumab_demo") docs = [ "Teplizumab 起源于新泽西医学院的一项免疫学研究", "Teplizumab 是抗 CD3 单克隆抗体,能延缓 1 型糖尿病发病", "关键 III 期试验显示中位发病时间推迟超过两年", "今天北京天气晴朗,适合户外运动", ] db.add_documents(docs) #top_k = 检索时“只要最相似的前 K 条” print("\n检索:'这种药能治什么病?'") for r in db.search("这种药能治什么病?", top_k=2): print(f" [距离 {r['distance']:.4f}] {r['text']}") #self.collection.count(): 4 #检索:'这种药能治什么病?' # [距离 0.1832] Teplizumab 是抗 CD3 单克隆抗体,能延缓 1 型糖尿病发病 # [距离 0.2614] 关键 III 期试验显示中位发病时间推迟超过两年
#RAG 最后一步:检索结果 → 拼 prompt → 调大模型 def get_completion(prompt: str, model: str = "qwen-plus"): """封装 OpenAI 原生接口参数""" completion = llm.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.1, ) return completion.choices[0].message.content def build_prompt(prompt_template, **kwargs): """把 Prompt 模板和参数组装成最终 prompt""" inputs = {} for k, v in kwargs.items(): # 判断是否是列表:是列表就编号拼行,否则原样用 if isinstance(v, list): items = [f"{i}. {item}" for i, item in enumerate(v, 1)] inputs[k] = "\n".join(items) else: inputs[k] = v return prompt_template.format(**inputs) # 用户提问 user_query = "这种药能治什么病?" # ① 检索:从向量库取出 top_k 条已知信息 search_results = db.search(user_query, top_k=2) contents = [r["text"] for r in search_results] prompt_template = f""" 你是一个问答机器人。 你的任务是根据下述给定的已知信息回答用户问题。 确保你的回复完全依据下述已知信息。不要编造答案。 如果下述已知信息不足以回答用户的问题,请直接回复"我无法回答您的问题"。 已知信息: {contents} ---- 用户问: {user_query} """ # ② 组装 prompt:模板 + 已知信息 + 用户问题 prompt = build_prompt(prompt_template, contents=contents, user_query=user_query) # ③ 调大模型生成回答 answer = get_completion(prompt) print(answer) #Teplizumab 是一种抗 CD3 单克隆抗体,用于延缓 1 型糖尿病的发病。 #试验显示它可将高危人群的中位发病时间推迟超过两年。

代码地址

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询