1.RAG的概念及作用

1.大模型的缺陷

首先要知道RAG是什么，能做什么，他是如何应用的，我们需要先了解一下大模型的缺陷，我们在用一些ai对话工具时，你有时候问一个问题，会发现

1.偶尔他回答的就是胡说八道，更过分可能风马牛不相及，俗称幻觉

2.你问一下比较新的知识，他回答不上来

3.问一个你擅长的小众的领域知识，或者黑话，他不知道

为啥呢？这中间存在很多因素，就会造成这些问题，下面简单说明一下，包括但不限于，例如

1.工具背后的模型训练参数不够，或者吸取了无效知识，然后它本身目标生成是找最符合上下文概率分布的，没有明确答案的时候，它可能随便捏造一个.

2.本身知识更新有点缓慢，小参数模型，或者私有部署的更明显.

3.训练时可能更多的追求广度去学习一些基本的公开的数据，对于极个别专业的领域深度不够.

2.RAG引入

上面简单引入了一些问题和对它产生问题的分析，不一定完全准确，只是阐述下大概的意思，重点是能理解就行，还有没有其他的原因呢？也许有！不过没必要深究，那对于产生问题的几点因素，如何解决呢？

1.提示词工程

2.RAG 检索增强生成

3.模型微调

这三个分别由易到难排列（大佬除外），然后附上各自对应解决的问题类型，提示词在上一篇分享中有可以参靠，但是我个人理解，提示词主要针对的问题点，是教会和指导你如何提问，解决的是因为你不会提问导致的回答不准，在这部分我主要介绍RAG

问题类型	最佳方案
解决胡说八道（幻觉）	提示词约束 + RAG
新知识不懂	RAG（实时更新）
小众领域不行	RAG（快速注入）或微调（深度内化）

3.什么是RAG?

RAG(检索增强生成)的初衷是为了增强大模型的事实性，时效性，减少幻觉，引入专业知识优化生成回复，举个大白话例子

你入职一个公司做点人事, 想问大模型我们公司内部裁员的流程，或者各种套路，那它肯定不知道，后面你找老板说，把秘密.PDF 发给你，然后你给到了大模型学习一下，后面再问的时候，他就知道了

Q：如何降本增效？

A: 多增加管理流程，制定各种考核试试服从性

Q: 哇哦，学到了

这里注意不要搞混了

1.模型微调的过程会改变模型中数十亿甚至数千亿个连接的权重。训练完成后，这些新知识就固化在模型里了，模型变得“不同”了。就像让学生复读一年，用新的教材、新的习题集（你的专业数据）重新学习。

2.RAG不改变模型。它只是把检索到的内容，和你的问题拼在一起，形成一个更长的、包含答案线索的提示词，再发给模型。模型只是基于这个更丰富的输入来生成答案。就类似学生不改变，还是原来的知识水平。但在考试的时候，你允许他带一本专门针对这场考试的书，他做到一个不会的问题，翻书翻到了，把问题答案一起给大模型，然后又大模型组织输出给你。

2.RAG的工作流程

RAG的发展过程也分为3个阶段，朴素RAG（Naive）、高级RAG（Advanced）、模块化RAG（Modular） 目前只学习了，普通和高级，这里用普通的作为实战参考，后续会继续高级部分

朴素RAG（Naive）

从图中可以看出，如果总结一下，它主要包含3个重要阶段

1.索引化 在准备阶段将文档切分（切割）成片段，再编码为向量，存储在向量数据库里。

2.检索 问答阶段在向量数据库中检索到与问题相似度最高的top k的片段，只有一次检索，只用向量相似度。

3.生成 将原始问题和检索得到的块所有内容都作为LLM的输入提示词，令其生成最终的回答。

但它在实际应用中会有不少坑，比如检索到的内容不相关（翻错页了），检索到的内容太多或太少（翻到的页数不对），多个资料矛盾（参考书前后说法不一），大模型忽略了检索到的资料（类似学生没看参考书），那这时就可以用Advanced RAG来进行优化

高级RAG（Advanced)

在高级RAG，优化增强了各个阶段，就是再把普通RAG的每个阶段都增强。

1.索引优化，在数据准备阶段做的更精细，切分数据更智能，可以按段落、句子、语义边界切，并保留重叠，保留元数据标题、章节、时间、作者，检索时能按照元数据过滤。

2.检索 可以查询改写让大模型先把问题优化成更适合检索的多个子问题或者关键词，不只是直接返回top k判断，而是用一个更精准的排序模型对召回结果重新打分，检索先找相关文档，再在文档内部找段落，或根据第一次结果二次检索，递归迭代。

3.生成将原始问题和检索得到的块进行上下文压缩，只保留与问题最相关的句子，控制长度。生成后让模型自查答案是否基于资料，有幻觉就重新生成进行自我纠错，多个答案，用投票或大模型总结的方式统一多个来源的答案。

1.数据准备文档分块策略

1.按照句子来切分

import re
text="拖延症，作为现代人类与截止日期之间的一场漫长博弈，致力于将“稍后再做”这一概念发挥到极致。在这个领域，大脑的奖励机制发挥着至关重要的误导作用。利用多样的借口，我们得以逃避、推迟乃至遗忘那些本该完成的任务。从深夜刷手机到整理并不乱的桌面，从计划明天的健身到发呆，拖延症的应用已遍布各个生活场景。随着焦虑情绪的飞速逼近，我们的工作效率与心态均遭遇了巨大挑战。如今，部分尖端的拖延症患者甚至能够处理极其复杂的赶工任务，如在最后一小时完成整篇论文、在通勤路上解决所有邮件和会议等。拖延症的研究推进不仅优化了我们“生死时速”的心理素质，也对提升我们在高压环境下的爆发力起到了关键作用。"
sent = re.split(r'(。|？|！|\...\...)', text)
print(f'分割后的数据是:{sent}')
chunks =[]
print(type(chunks))
for sen,pun in zip(sent[::2],sent[1::2]):
chunks.append(sen+pun)
print(f'拼接后的数据是：{chunks}')

2.按照字符数来切分

text=".........."
chunks =[]
for i in range (0,len(text),50):
chunks.append(text[i:i+50])
print(chunks)

3.按固定字符数 结合overlapping window

def overlap_split(text, n, stride):
for i in  range(0, len(text), stride):
print(text[i: i + n])
text = "......"
overlap_split(text, 100, 20)

4.递归方法 使用langchain的RecursiveCharacterTextSplitter

from langchain.text_splitter import RecursiveCharacterTextSplitter
text=".........需要被索引的文档数据"
split_data = RecursiveCharacterTextSplitter(
chunk_size = 50,
chunk_overlap=10,
length_function=len,
# text_splitter=TextSplitter,
)
print(split_data)
data =split_data.split_text(text)
print(data)

3.向量和Embedding

1.向量

在数学中，向量（也称为欧几里得向量、几何向量），指具有大小（magnitude）和方向的量。它可以形象化地表示为带箭头的线段。箭头所指：代表向量的方向；线段长度：代表向量的大小。

1.将文本转成一组浮点数：每个下标 i，对应一个维度 2.整个数组对应一个 n 维空间的一个点，即文本向量又叫 Embeddings 3.向量之间可以计算距离，距离远近对应语义相似度大小

上面的这段话摘抄自专业术语，我也不知道对不对，大白话理解，向量就是一个特征代号清单，AI眼中，电脑不懂如何定义具体对象和物品，他只认识数字，例如我需要描述一只猫，怎么描述

1. 是不是活的？（1 代表是，0 代表不是） -> 1

2. 有没有毛？ -> 1

3. 会不会抓老鼠

4. 体型大小（0-1-）->0.3

5. 凶猛程度

那他的清单（向量）就是：[1，1，0.9，0.3，0.1]，同理我们再扩展2个老虎和汽车，按照清单老虎可以是[1，1，0.8，0.9，0.8]，汽车就是 [0,0,0,0.5,0]

可以发现猫和老虎有些层面他们一致，甚至很相似，但是和汽车却相差很远，换句话说如果2个对象，或者2段话，他们的意思差不多，那么转化的那串数字长得就很像，在数学空间里他们距离就近，反之就离得远。这种利用距离表示相似度的计算方式有2种，一个叫余弦距离Cosine，一种叫欧式距离L2，他们都是值越小越相似，与余弦距离还有一个紧密相关的概念余弦相似度 (Cosine Similarity)就不是，它是值越大越相似。

那么大模型是怎么回答问题的呢?

我们所提的问题会转换成向量, 然后会拿这个向量在向量空间进行对比,找到和小猫相近的内容,比方说老鼠,有没有毛,离的比较近那么大模型会根据概率的问题把这句话完整的写出来(当然其中还需要很多其他的东西)。

2.Embedding

数据的向量其实是非常复杂的,表示学习会对向量进行优化得到的一个新的向量内容我们会叫做嵌入。

嵌入:

表示学习的一种形式，通常用于将高维数据映射到低维空间中的表示形式。嵌入可以是词嵌入、图像嵌入、图嵌入等。例如，在自然语言处理中，词嵌入将词语映射到低维向量空间，以捕捉词语之间的语义和句法关系。在图像处理中，图像嵌入将图像映射到低维向量空间，以表示图像的视觉特征。

它和向量的关系就是通过Embedding把具体描述的对象转换为向量数字。 我们在使用时不需要过多关注Embedding，都是使用现成的转换工具，可以使用自己部署的私有向量模型，也可以使用一些厂商提供的，我下面就是使用百炼的免费向量模型。

from openai import OpenAI
from dotenv import load_dotenv
import os
load_dotenv()
# 2. 获取配置信息
API_KEY = os.getenv("QW_KEY")
API_URL = os.getenv("QW_URL")
# 3. 初始化客户端
client = OpenAI(api_key=API_KEY, base_url=API_URL)
def get_embedding(text):
data = client.embeddings.create(input=text,model='text-embedding-v2')
return [i.embedding for i in data.data]
test_query = ['小猫']
vec=get_embedding(test_query)
print(vec)
print(len(vec[0]))
test_query = ['老虎']
vec=get_embedding(test_query)
print(vec)
print(len(vec[0]))

最终调用输出就是这样的数据，上面只用了5个维度举例，这里结果数组有1536个，也可以理解为用了1536个维度来表示小猫，或者老虎，维度越多代表越细，至于有哪些，我也没研究过，内部计算肯定是非常复杂的，而且和不同模型有关，从成本权衡的角度出发，维度越多，计算越慢，占用的存储空间也越大。

3.本地部署向量模型

1.可以访问魔塔社区[1] 去选择文本向量，然后选择需要的下载自己需要的模型

2.安装魔塔的下载器，然后下载到具体目录 先设置下目录的下载参数

export MODELSCOPE_CACHE="/Users/yui/3.Resources/Demo/llm/embdding/bge-large-zhv15" #把所有文件下载这里
export MODELSCOPE_DOWNLOAD_THREADS=10 #开启10个线程一起下

pip install modelscope
# 下载模型到本地
# pip install modelscope
from modelscope import snapshot_download
#这里是mac的路径，windows直接输入具体路径
model_dir = snapshot_download("BAAI/bge-large-zh-v1.5", cache_dir="/Users/yui/3.Resources/Demo/llm/embdding/bge-large-zhv15")
print(f'模型下载完成')

3.本地嵌入模型

# pip install sentence-transformers 下载用来加载模型库的包
from sentence_transformers import SentenceTransformer
# 1. 加载本地模型
# 注意：Mac 路径使用正斜杠 /，且不需要 r 前缀
model = SentenceTransformer('/Users/yui/3.Resources/Demo/llm/embdding/bge-large-zhv15/BAAI/bge-large-zh-v1___5')
# 2. 准备数据
data = ['你好']
# 3. 转换成向量
embedding = model.encode(data)
# 4. 打印结果
print(embedding.shape)

4.向量距离计算

就是将输入的问题以及对存在的多个答案或者一堆答案，进行向量化，然后将输入的向量和答案的向量，根据不同的算法，进行相似度计算，具体看下面吧~

余弦距离Cosine：基于两个向量夹角的余弦值来衡量相似度,计算公式: 两个向量的乘积 / 两个向量范数的乘积 A×B / ∥A∥ × ∥B∥ 范数: l1(绝对值), l2(勾股定理), l无穷

欧式距离L2：通过计算向量之间的欧几里得距离来衡量相似度。点积：计算两个向量的点积，适合归一化后的向量。

import os
import numpy as np
from numpy import dot
from numpy.linalg import norm
from openai import OpenAI
from dotenv import load_dotenv
# 1. 初始化配置
load_dotenv()
API_KEY = os.getenv("QW_KEY")
API_URL = os.getenv("QW_URL")
client = OpenAI(api_key=API_KEY, base_url=API_URL)
# 2. 定义相似度计算函数
def cos_sim(a, b):
"""
余弦相似度
范围: [-1, 1]，越大越相似 (1 表示完全相同)
"""
return dot(a, b) / (norm(a) * norm(b))
def l2_distance(a, b):
"""
欧氏距离
范围: [0, +∞)，越小越相似 (0 表示完全相同)
"""
return norm(np.asarray(a) - np.asarray(b))
# 3. 定义获取向量的函数
def get_embedding(texts):
"""调用 API 获取文本向量"""
# 如果传入的是字符串，转为列表以统一处理
if isinstance(texts, str):
texts = [texts]
response = client.embeddings.create(input=texts, model="text-embedding-v2")
return [item.embedding for item in response.data]
# 4. 准备数据
query = "人工智能的发展对就业市场的影响"
documents = [
"全球气候变化导致极端天气事件频发",
"某科技公司发布了新一代智能手机",
"专家预测AI将取代部分重复性工作，同时创造新的岗位",
"国际足联宣布2026年世界杯举办地",
"本地超市推出夏季促销活动"
]
# 5. 计算向量
# 获取查询向量 (取第一个元素，因为返回的是列表)
query_vec = np.array(get_embedding(query)[0])
# 获取文档向量列表
doc_vecs = [np.array(vec) for vec in get_embedding(documents)]
# 6. 打印结果
print(f"--- 查询: '{query}' ---\n")
print("【余弦相似度】(越大越相似，最大为 1):")
for i, doc in enumerate(documents):
score = cos_sim(query_vec, doc_vecs[i])
print(f"{score:.4f} - {doc}")
print("\n【欧氏距离】(越小越相似，最小为 0):")
for i, doc in enumerate(documents):
dist = l2_distance(query_vec, doc_vecs[i])
print(f"{dist:.4f} - {doc}")

可以看第一个距离其中最大的是第三个0.6，第二个距离最小的也是第三个0.8，所以就匹配成功，这一步咱们已经能把提问和答案进行向量化，后续就需要把向量存起来，那就需要使用到向量数据库，这里容易有个误区，向量数据库存储的是“答案”，而问题不需要存起来~

4.向量数据库使用

向量数据库已成为数据管理和AI模型不可或缺的一部分。向量数据库是一种专门设计用来存储和查询向量嵌入数据的数据库，他的作用就是存储被向量化的非结构化数据，并没有很神秘，就像mysql，sqlserver一样，存储的一般是结构化的数据，只不过侧重点不同而已。我们来熟悉一下，在AI中通常使用哪些向量数据库

数据库名称	类型	核心优势	适用场景
Milvus	专用开源	性能怪兽：分布式架构，支持千亿级向量，高可用，生态最完善。	超大规模企业级应用、海量数据检索。
Qdrant	专用开源	Rust 编写：内存效率高，速度极快，API 设计现代，支持复杂的过滤查询。	对性能敏感、资源受限、追求现代化的 RAG 系统。
Weaviate	专用开源	AI 原生：内置向量化模块（不用自己调 API），支持图数据库功能（知识图谱）。	快速构建 RAG、需要语义与知识图谱结合的场景。
Chroma	专用开源	极简轻量：Python 开发者最爱，可嵌入代码运行，无需独立服务器，开发调试极快。	本地开发测试、快速原型验证、中小型应用。
Pinecone	全托管 SaaS	开箱即用：完全托管，自动扩缩容，稳定性极高，无需运维。	初创团队、不想运维、追求极致稳定的生产环境。
ElasticSearch	传统扩展	混合搜索之王：全文检索（关键词）+ 向量检索结合得最好，生态成熟。	已有 ELK 技术栈、需要复杂关键词过滤+语义搜索的场景。
Redis	传统扩展	极速缓存：基于内存，读写速度微秒级，支持向量索引。	实时推荐、作为缓存层加速、对延迟要求极高的场景。
PostgreSQL	传统扩展	SQL 融合：通过 pgvector 插件支持，既能存用户数据又能存向量，用 SQL 就能查。	已有 PG 架构、中小规模数据、不想引入新数据库的团队。

建议

•想省事、不差钱 ➡️ Pinecone•数据量巨大（亿级）、追求极致性能 ➡️ Milvus•追求轻量高性能 ➡️ Qdrant•Python 开发、做个 Demo 或内部工具 ➡️ Chroma•不想引入新组件、已有 PG 数据库 ➡️ pgvector•需要“关键词+语义”混合搜索 ➡️ ElasticSearch•缓存加速 ➡️ Redis

1. chroma演示

在学习阶段，就是用chroma来作为向量数据库，比较方便

1.安装chroma

pip install chromadb==0.5.3

2.chroma基本操作使用

import chromadb
# 创建客户端存内存
client = chromadb.Client()
#创建客户端存本地文件系统
#client = chromadb.PersistentClient("/Users/yui/3.Resources/Demo/llm/db/chroma")
# 创建或者获取当前db
cli = client.get_or_create_collection(name='chroma_test')
# 添加数据
cli.add(
documents=["张三","李四","王五"] # 添加元素
,embeddings=[[1,2,3],[4,5,6],[7,8,9]] #对应3个向量[7,8,9]:王五
,ids=["1","2","3"])#对应id 1：张三...
# 查询收据
getData = cli.get(ids=["3"],include=["embeddings","documents"],where_document={"$contains":"五"})
print(f'查询数据：{cli.get()}')
#删除数据
cli.delete(ids=["3"])
print(f'删除后的数据：{cli.get()}')
#修改数据
cli.update(
documents=["赵6","李四","王五"]
,embeddings=[[1,2,3],[4,5,6],[7,8,9]]
,ids=["1","2","3"]
)
print(f'修改后的数据：{cli.get()}')

2.结合Chroma、Embedding操作结构化数据

思路 1.准备结构化json文件数据,在真正的RAG中，这些数据的来源可能从数据库，也可能从接口请求来

2.将文件数据的问题进行向量化，答案作为文档数据，然后存入向量数据库

3.把用户提问的问题进行向量化，然后搜索向量库

1.先准备一些结构化的json数据。

{"instruction": "这段时间去上厕所本来想小便的可是每次都会拉大便", "input": "", "output": "这可能是因为你的饮食习惯或者消化系统的问题导致的。建议你试着调整一下饮食，增加膳食纤维的摄入量，多喝水，避免过度依赖泻药。同时，如果问题持续存在，建议去医院检查一下，排除肠道疾病等可能性。"}
{"instruction": "医生呀！我刚被查出得了白癜风，不知道初期治疗需要多少钱？", "input": "", "output": "白癜风的治疗费用因个体差异和治疗方案的不同而有所差异。初期治疗主要以口服药物和外用药物为主，费用相对较低，一般几百元左右。但是，如果采用激光治疗、光疗等高端治疗方法，费用会更高。建议您咨询专业医生，根据自己的情况进行治疗方案的选择，同时了解相关的费用情况。"}
{"instruction": "男朋友31岁，前列腺B超数据是34*48*31，，包膜完整平滑，内部回声均匀，未见明显异常回声，结论是前列腺肥大，增生，请问情况很严重嘛，该怎么治疗呢", "input": "", "output": "根据您提供的数据，男朋友的前列腺B超结果表明前列腺肥大增生，但是并未出现明显异常回声，所以情况并不算严重。前列腺增生是男性常见的一种生理现象，是由于年龄增长和性激素水平变化引起的。具体治疗方案需要根据症状和严重程度来确定，一般包括药物治疗和手术治疗两种方式。\n\n药物治疗可以采用口服药物或者局部治疗，如口服α1-受体阻滞剂、5α-还原酶抑制剂、荷尔蒙治疗等。手术治疗主要包括经尿道电切术、经皮穿刺前列腺电子治疗、经皮穿刺前列腺热疗、经直肠前列腺切除术等。\n\n建议男朋友及时就医，并由专业医生根据具体情况制定合适的治疗方案。同时，男性朋友应该保持良好的生活习惯，如均衡饮食、适量运动、避免过度劳累等，有助于减缓前列腺肥大增生的进程。"}
{"instruction": "左大腿脾经内侧疼痛大概有6寸左右，有几天了，不知道怎么导致的，表面没什么就是一按就刺痛，这怎么回事", "input": "", "output": "左大腿内侧疼痛可能是由多种因素引起的，例如：\n\n1. 肌肉劳损或拉伤：如果您最近进行了剧烈的运动或活动，可能会导致肌肉劳损或拉伤，从而引起疼痛。\n\n2. 腰椎间盘突出：腰椎间盘突出也可能会引起大腿内侧疼痛，尤其是在坐姿或弯腰时更加明显。\n\n3. 神经痛：大腿内侧疼痛可能是由于神经受损或受压所致。神经痛通常表现为刺痛或针刺感。\n\n4. 骨折或骨裂：如果您最近遭受了外伤或摔倒，可能会导致骨折或骨裂，引起大腿内侧疼痛。\n\n建议您及时到医院进行检查，确定疼痛的原因，并针对性治疗。同时，您可以适当休息、热敷或按摩缓解疼痛。"}
{"instruction": "你好，全身没劲，没精神，吃不下饭，只想睡觉，是什么情况", "input": "", "output": "这些症状可能是身体出现了一些问题，建议您去看医生进行检查。这些症状可能和许多疾病有关，比如抑郁症、贫血、甲状腺问题等等。医生可以为您进行全面的身体检查，帮助您找到病因并给出相应的治疗建议。同时，您可以尝试保持良好的生活习惯，例如规律作息、均衡饮食、适量运动等，这些都有助于提高身体的免疫力和健康状况。"}
{"instruction": "喉咙发紧干精神有点紧绷早上起来手也没有力气。一直在担心。", "input": "", "output": "您可能需要注意以下几点：\n\n1. 饮食方面，注意多喝水并避免食用刺激性食物，如辣椒、咖啡、酒等。\n\n2. 保持室内空气湿润，可以使用加湿器或者放置一盆水。\n\n3. 适当休息，避免过度疲劳。可以尝试进行一些轻度的运动或者放松的活动，如瑜伽、冥想等。\n\n4. 如果症状持续或者加重，建议就医或者咨询医生。可能需要进一步的检查和治疗。"}
{"instruction": "女宝宝十个半月体重十六斤多，身高67cm，是不是发育太慢了，最近三个月身高没有变法，一颗牙都没有扎，也不流口水，", "input": "", "output": "根据你提供的数据，宝宝的体重和身高处于正常范围内。一般来说，女宝宝的身高增长速度会在出生后逐渐变慢，尤其是在10个月左右时会明显减缓。此外，宝宝的牙齿生长时间也有差异，有些宝宝可能在10个月左右才开始长牙。因此，你不必过于担心宝宝的发育情况，但如果你还是有疑虑，建议咨询儿科医生的意见。"}
{"instruction": "就是庠没有其他问题，然后想知道用什么药或者怎么样才能好还有是什么原因弄成这样的", "input": "", "output": "很抱歉，我不清楚您所指的具体情况是什么。如果您有身体不适或疾病，建议您咨询医生或专业医疗机构，以便获得准确的诊断和治疗建议。"}
{"instruction": "我是早段时候觉得我总是脚疼，一开始没在意，后来是疼的厉害，就去医院看了，检查下来是尿酸高，我想知道这样的话需要注意什么，还有饮食方面可以吃咸菜吧。", "input": "", "output": "如果您被诊断为尿酸高，需要注意以下几点：\n\n1. 饮食方面：尿酸高的人需要避免高嘌呤食物，如猪肝、肾脏、海鲜、肉类、鱼子酱等。此外，您应该控制蛋白质的摄入量，每天的摄入量不应超过100克。咸菜中含有较高的盐分，不建议长期食用。\n\n2. 控制体重：肥胖是尿酸高的重要因素之一，因此您需要控制体重，保持健康的体重。\n\n3. 饮水：增加饮水量，有助于尿酸的排出。\n\n4. 合理运动：适量的运动可以帮助减轻体重，降低尿酸水平。\n\n5. 避免饮酒：饮酒可以增加尿酸的产生和减少排出，所以需要避免饮酒。\n\n如果您的尿酸值很高，建议您去医院找专业医生进行诊治。同时，也要注意饮食和生活习惯的改变，以降低尿酸水平，保持身体健康。"}

2.通过调用embedding模型来进行向量化 3.根据具体问题搜索

import chromadb
import json
from openai import OpenAI
from dotenv import load_dotenv
import os
class MyVectoryDbConnector:
def __init__(self):
# 初始化 ChromaDB 持久化客户端
self.db = chromadb.PersistentClient("Users/yui/3.Resources/Demo/llm/db/chroma")
# 获取或创建集合
self.collection = self.db.get_or_create_collection("case_db")
pass  # 原始代码保留
def get_embedding(self, text, model='text-embedding-v2'):
# 初始化 OpenAI 客户端，从环境变量读取 Key 和 URL
client = OpenAI(api_key=os.getenv("QW_KEY"), base_url=os.getenv("QW_URL"))
# 调用 API 获取嵌入向量
data = client.embeddings.create(input=text, model=model).data
return [x.embedding for x in data]
def add_document(self, content, output):
# 获取内容的向量
embeddings = self.get_embedding(content)
# 将文档、向量和 ID 添加到数据库
self.collection.add(
documents=output,
embeddings=embeddings,
ids=[f"id{i}" for i in range(0, len(output))]
)
def search(self, query):
# 获取查询语句的向量
q_embeddings = self.get_embedding(query)
# 在数据库中查询最相似的 n_results 条数据
result = self.collection.query(
query_embeddings=q_embeddings,
n_results=2
)
return result
if __name__ == '__main__':
# 加载 .env 环境变量
load_dotenv()
# 读取 JSON 文件（按行解析）
with open('train_zh.json', 'r', encoding='utf-8') as f:
data = [json.loads(line) for line in f]
# 提取前 10 条数据的 instruction 和 output
instruction = [entry['instruction'] for entry in data[0:10]]
outputs = [entry['output'] for entry in data[0:10]]
# 初始化数据库连接器
db = MyVectoryDbConnector()
# 添加文档到向量库
db.add_document(instruction, outputs)
# 定义查询语句
user_query = '吃不下饭'
# 执行搜索
result = db.search(query=user_query)
ids = result['ids'][0]
documents = result['documents'][0]
print(f"--- 查询: {user_query} ---")
# 使用 zip 同时遍历 id 和 文档内容
for i, (doc_id, doc_content) in enumerate(zip(ids, documents)):
print(f"\n[结果 {i + 1}]")
print(f"ID: {doc_id}")
print(f"文档: {doc_content}")

4.结果

5.RAG平台搭建的方式

现在有很多的平台提供的功能就把RAG包括在内，例如Coze（扣子）、Dify、FastGPT，这些平台化产品都内置了完整 RAG 能力， 主打：

低代码 / 无代码，上手快 不用自己搭向量库、处理文档、做分段召回 维护简单，部署省心

但代价也很明显：

灵活性被锁死想改召回策略、模型逻辑、工作流，往往改不动，只能用平台给的功能。 成本不可控今天便宜，明年可能涨价几倍；用量一大，费用直接起飞。 数据安全本质上不可靠，你理解的安全 = 别人不知道 ，安全平台定义的安全：你把秘密交给我，我承诺不泄露 ， 安全只要文档上传到别人服务器，就不再完全属于你。

1.本地部署FastGPT

接下来使用FastGPT来试试RAG，不需要本地部署大模型，现在各大厂商都在拉用户，免费的学习体验完全够用，我们就用百炼的。至于fastgpt具体安装步骤，建议去看官方文档上的步骤，使用docker部署相对来说非常简单,写好配置文件，然后Docker-compose 一键启动就完事了。

2.使用知识库

1.部署成功后打开docker会发现，他依赖了很多中间件，在内部它把这些当做插件来读取的。

2.如果容器运行正常，那么此时你打开http://localhost:3000/[2] 就可以登录，账号root 密码1234

3.先创建知识库，然后上传文件，进行rag，中间还有很多设置模型的，一般都很傻瓜，一看就明白

4.再创建Agent，选择对应知识库和模型

5.保存发布，选择渠道，可以生成api也可以生成免登录连接6.访问链接后进行问答

6.然后访问连接进行问答就好了

6.结合大模型代码实战

1.现在将Embedding、还有大模型一起来做一个简单的问答，Embedding 和llm都是采用阿里百炼的免费模型，简单的时序图如下

伪代码思路，主要分为几部分各自职责如下

1.读取PDF或者word ,需要先下载包模块，然后定义读取pdf函数和切割文档函数

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple python-docx
pip  install -i https://pypi.tuna.tsinghua.edu.cn/simple pdfminer.six

# 定义读取pdf函数
def extract_text_from_pdf(filename,page_number=None,min_size=1):
pass
# 切割文档 使用滑动，这样就不会截断句子
def sliding_window(text, chunk_size, stride):
pass

2.定义MyVectorDBConnector类，主要用来embedding，把结果添加到向量库、根据查询条件从向量库中查询

class MyVectorDBConnector:
def __init__(self):
pass
def get_embeddings(self, texts, model='text-embedding-v3'):
pass
def add_document(self,instruction):
pass
def search_document(self,query,n_results):
pass

3.定义问答类，用来组装查询结果及调用大模型

class RAG_bot:
#同c#中的构造函数
def __init__(self,vector_db,n_res):
pass
#调用大模型
def get_complete(self,prompt,model='qwen-turbo'):
pass
#回答输出
def chat(self,query):
pass

4.然后入口函数中准备提示词及调用

if __name__ == '__main__':
load_dotenv()
prompt_template = """你是一个问答机器人。已知信息: __INFO__用户问：__QUERY__请用中文回答用户问题。"""
page_data = extract_text_from_pdf('财务管理文档.pdf',page_number=[0,1,2])
vector = MyVectorDBConnector()
vector.add_document(page_data)
rag_bot = RAG_bot(vector_db=vector,n_res=2)
rag_bot.chat('财务管理权限划分')

完整代码如下

import os
import chromadb
from dotenv import load_dotenv
from openai import OpenAI
from pdfminer.high_level import extract_pages
from pdfminer.layout import LTTextContainer
# --- 文本处理函数 ---
def sliding_window(text, chunk_size, stride):
"""
使用滑动窗口将文本分割成指定大小的块。
Args:
text (str): 待分割的原始文本。
chunk_size (int): 每个文本块的大小。
stride (int): 滑动步长。
Returns:
List[str]: 分割后的文本块列表。
"""
result = []
for i in range(0, len(text), stride):
chunk = text[i:i + chunk_size]
result.append(chunk)
return result
def extract_text_from_pdf(filename, page_number=None, min_size=1):
"""
从 PDF 文件中提取文本，并使用滑动窗口进行分块。
Args:
filename (str): PDF 文件路径。
page_number (List[int], optional): 指定要提取的页码列表（从0开始）。如果为None，则提取所有页。
min_size (int, optional): 未在当前逻辑中使用，保留参数。
Returns:
List[str]: 提取并分块后的文本列表。
"""
paragraphs = []
full_text = ''
# 读取PDF内容
content = extract_pages(filename)
# 将内容读取出来为文本，加入 full_text
for i, page_layout in enumerate(content):
# 如果指定了页码且当前页不在指定页码中，则跳过
if page_number is not None and i not in page_number:
continue
if page_layout is not None:
for element in page_layout:
if isinstance(element, LTTextContainer):
# 提取文本并移除换行符
full_text += element.get_text().replace("\n", "").replace("", "")
# 如果 full_text 有内容，再调用滑动切分
if full_text:
text_chunks = sliding_window(full_text, 250, 100)
for chunk in text_chunks:
paragraphs.append(chunk)
return paragraphs
# --- 向量数据库连接器 ---
class MyVectorDBConnector:
"""
封装 ChromaDB 操作和文本嵌入生成的类。
"""
def __init__(self):
"""
初始化 ChromaDB 客户端和集合。
"""
self.db = chromadb.Client()
self.collection = self.db.get_or_create_collection("Rag_case")
def get_embeddings(self, texts, model='text-embedding-v3'):
"""
获取文本列表的嵌入向量。使用分批处理以避免超过 API 限制。
Args:
texts (List[str]): 待嵌入的文本列表。
model (str, optional): 使用的嵌入模型名称。默认为 'text-embedding-v3'。
Returns:
List[List[float]]: 嵌入向量列表。
"""
# 初始化 OpenAI 客户端 (使用 Qwen API)
client = OpenAI(api_key=os.getenv("QW_KEY"), base_url=os.getenv("QW_URL"))
all_embeddings = []
batch_size = 10  # 【关键】根据报错，这里最大只能是 10
# 1. 循环切分列表
for i in range(0, len(texts), batch_size):
# 取出当前批次（例如 0-10, 10-20...）
batch_texts = texts[i: i + batch_size]
# 2. 发送请求
try:
response = client.embeddings.create(input=batch_texts, model=model)
# 3. 提取向量并加入总列表
# 注意这里要遍历 response.data
for item in response.data:
all_embeddings.append(item.embedding)
except Exception as e:
print(f"❌ 处理第 {i} 到 {i + batch_size} 条数据时出错: {e}")
raise e
return all_embeddings
def add_document(self, instruction):
"""
将文档文本添加到向量数据库中。
Args:
instruction (List[str]): 待添加的文档文本列表。
"""
embeddings = self.get_embeddings(instruction)
self.collection.add(
embeddings=embeddings,
documents=instruction,
ids=[f"id{i}" for i in range(len(instruction))]
)
def search_document(self, query, n_results):
"""
在向量数据库中搜索与查询最相关的文档。
Args:
query (str): 查询文本。
n_results (int): 返回的结果数量。
Returns:
QueryResult: ChromaDB 的查询结果对象，包含匹配的文档等信息。
"""
query_embeddings = self.get_embeddings([query])
result = self.collection.query(
query_embeddings=query_embeddings,
n_results=n_results
)
return result
# --- RAG 机器人 ---
class RAG_bot:
"""
实现 RAG (检索增强生成) 逻辑的聊天机器人。
"""
def __init__(self, vector_db, n_res):
"""
初始化 RAG 机器人。
Args:
vector_db (MyVectorDBConnector): 向量数据库连接器实例。
n_res (int): 每次检索返回的文档数量。
"""
self.vector_db = vector_db
self.n_res = n_res
def get_complete(self, prompt, model='qwen-turbo'):
"""
调用大语言模型生成回复。
Args:
prompt (str): 发送给模型的提示词。
model (str, optional): 使用的模型名称。默认为 'qwen-turbo'。
Returns:
str: 模型生成的文本回复。
"""
message = [{"role": "user", "content": prompt}]
client = OpenAI(api_key=os.getenv("QW_KEY"), base_url=os.getenv("QW_URL"))
response = client.chat.completions.create(
model=model,
messages=message,
temperature=0
)
return response.choices[0].message.content
def chat(self, query):
"""
处理用户聊天查询。
1. 使用查询在向量数据库中搜索相关文档。
2. 将相关文档和用户查询组合成提示词。
3. 调用大语言模型生成回复。
4. 打印回复。
Args:
query (str): 用户的查询。
"""
# 搜索相关文档
search_data = self.vector_db.search_document(
query=query,
n_results=self.n_res
)
# 打印搜索到的文档内容 (用于调试)
print(search_data['documents'][0])
# 构建提示词，替换模板中的占位符
# 注意：这里假设 prompt_template 已在全局定义
prompt = prompt_template.replace('__INFO__', '\n'.join(search_data['documents'][0])).replace('__QUERY__', query)
# 获取模型回复
response = self.get_complete(prompt)
# 打印 AI 回复
print(f'AI的回复{response}')
# --- 主程序入口 ---
if __name__ == '__main__':
# 加载 .env 文件中的环境变量
load_dotenv()
# 定义 RAG 提示词模板
prompt_template = """
你是一个问答机器人。
你的任务是根据下述给定的已知信息回答用户问题。
确保你的回复完全依据下述已知信息。不要编造答案。
如果下述已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。
已知信息:
__INFO__
用户问：
__QUERY__
请用中文回答用户问题。
"""
# 1. 从 PDF 提取文本
page_data = extract_text_from_pdf('财务管理文档.pdf', page_number=[0, 1, 2])
# 2. 初始化向量数据库连接器
vector = MyVectorDBConnector()
# 3. 将提取的文本添加到向量数据库
vector.add_document(page_data)
# 4. 初始化 RAG 机器人
rag_bot = RAG_bot(vector_db=vector, n_res=2)
# 5. 进行聊天查询
rag_bot.chat('财务管理权限划分')

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】