本文为Datawhale 开源项目wow-rag的学习笔记与分享，仅供参考，如有错误描述，请评论指正，感谢。
开源项目地址：
github.com/datawhalechina/wow-rag
欢迎大家点击关注我，学习更多AI相关知识。
获取免费大模型APIKEY
这里我们选择使用 智谱 的，操作很简单，访问下面链接，注册登录。
https://open.bigmodel.cn/usercenter/proj-mgmt/apikeys
然后按图示步骤操作，如果你登录后不在APIKEY生成页面，也可以先点击右上角的钥匙按钮进入，然后点击添加新的API Key。
给一个名字，方便后面区分（比如后面有其他业务或者学习使用其他的API Key ，后续可以区分调用来源）。
创建完成后，页面上点击复制即可。（看图示操作）
注意智谱的 embedding-2 模型不免费提供，需要先访问下方链接购买下，测试买这个7.5元的即可。（“图”方便就不找免费的代替了，欢迎评论区分享）
https://open.bigmodel.cn/tokenspropay?productIds=product-074
模型测试调用
安装需要的包：
pip install faiss-cpu scikit-learn scipy openai
首先创建一个 client，后面可以使用 client 调用各种接口模型。
'''
代码来源：github.com/datawhalechina/wow-rag
'''
# 智谱调用
from openai import OpenAI
api_key = "上一步骤复制的api_key"
base_url = "https://open.bigmodel.cn/api/paas/v4/"
chat_model = "glm-4-flash"
emb_model = "embedding-2"
client = OpenAI(
api_key = api_key,
base_url = base_url
)
对话模型使用的是 glm-4-flash，嵌入式模型模型使用的 embedding-2。
调用对话模型 glm-4-flash
很固定的格式，调用了 client.chat.completions.create函数，主要参数有：
model，使用的模型 messages，消息列表 stream，是否使用流式输出
'''
代码来源：github.com/datawhalechina/wow-rag
'''
# 使用 OpenAI 的 Chat Completions API 创建一个聊天完成请求
prompt = "你好，你可以做什么，列5点"
response = client.chat.completions.create(
model=chat_model, # 填写需要调用的模型名称
messages=[
{"role": "user", "content": prompt},
],
stream=True,
)
# 流式输出
if response:
# 遍历响应中的每个块
for chunk in response:
# 获取当前块的内容
content = chunk.choices[0].delta.content
# 如果内容存在
if content:
# 打印内容，并刷新输出缓冲区
print(content, end='', flush=True)
调用嵌入式模型 embedding-2
首先 clone github.com/datawhalechina/wow-rag 内容，作为源料，
git clone https://github.com/datawhalechina/wow-rag
然后读取 wow-rag/docs/第1课-手搓一个土得掉渣的RAG.md 文档作为测试 embedding 源料，读取后再使用 for 循环进行了简单的分块处理，设置每个文本块的大小为 150 个字符，结果存储在 chunks 变量。
'''
代码来源：github.com/datawhalechina/wow-rag
'''
with open("./wow-rag/docs/第1课-手搓一个土得掉渣的RAG.md") as f:
embedding_text = f.read()
# 设置每个文本块的大小为 150 个字符
chunk_size = 150
# 使用列表推导式将长文本分割成多个块，每个块的大小为 chunk_size
chunks = [embedding_text[i:i + chunk_size] for i in range(0, len(embedding_text), chunk_size)]
接下来就是主要代码了，核心逻辑流程如下：
嵌入生成： 使用 OpenAI API将文本块转换为嵌入向量，并将结果存入embeddings列表。归一化处理： 调用 normalize方法对嵌入向量进行归一化处理。向量存储： 使用 faiss.IndexFlatIP创建一个用于内积相似性计算的索引。将归一化后的向量存入索引。 索引查询： 使用 index.ntotal获取索引中的向量总数。
输出：
打印索引中的向量总数，验证是否成功存储。
Tips 为什么要归一化？
若向量未归一化，则内积值不仅受方向影响，还受模长的影响。向量经过归一化后模为1，则内积等价于余弦相似性，确保仅计算方向的相似性。
'''
代码来源：github.com/datawhalechina/wow-rag
'''
# 用于对数据进行归一化处理，使每个嵌入向量的模为1
from sklearn.preprocessing import normalize
import numpy as np
# 一个高效的相似性搜索库，用于存储和检索向量
import faiss
import time
# 初始化一个空列表来存储每个文本块的嵌入向量
embeddings = []
print("开始向量化存储～")
t1 = time.time()
# 遍历每个文本块
for chunk in chunks:
# 使用 OpenAI API 为当前文本块创建嵌入向量
response = client.embeddings.create(
model=emb_model,
input=chunk,
)
# 将嵌入向量添加到列表中
embeddings.append(response.data[0].embedding)
# 使用 sklearn 的 normalize 函数对嵌入向量进行归一化处理
normalized_embeddings = normalize(np.array(embeddings).astype('float32'))
# 获取嵌入向量的维度
d = len(embeddings[0])
# 创建一个 Faiss 索引，用于存储和检索嵌入向量
index = faiss.IndexFlatIP(d)
# 将归一化后的嵌入向量添加到索引中
index.add(normalized_embeddings)
print(f"完成向量化存储～耗时：{round(time.time()-t1, 2)}s")
t1 = time.time()
# 获取索引中的向量总数
n_vectors = index.ntotal
print(n_vectors)
综合使用
首先写一个函数match_text用于找到与输入文本最相似的前k个文本块。代码运行逻辑：
问题转化为嵌入向量： 利用 OpenAI 接口代码调用嵌入模型（embedding-2），将输入文本块转化为嵌入向量 Faiss 索引： 归一化处理后，通过 index.search 函数高效检索最相似的K个嵌入向量 返回结果： 遍历打印匹配的嵌入向量对应的内容，并返回
'''
代码来源：github.com/datawhalechina/wow-rag
'''
from sklearn.preprocessing import normalize
def match_text(input_text, index, chunks, k=2):
"""
在给定的文本块集合中，找到与输入文本最相似的前k个文本块。
参数:
input_text (str): 要匹配的输入文本。
index (faiss.Index): 用于搜索的Faiss索引。
chunks (list of str): 文本块的列表。
k (int, optional): 要返回的最相似文本块的数量。默认值为2。
返回:
str: 格式化的字符串，包含最相似的文本块及其相似度。
"""
# 确保k不超过文本块的总数
k = min(k, len(chunks))
# 使用OpenAI API为输入文本创建嵌入向量
response = client.embeddings.create(
model=emb_model,
input=input_text,
)
# 获取输入文本的嵌入向量
input_embedding = response.data[0].embedding
# 对输入嵌入向量进行归一化处理
input_embedding = normalize(np.array([input_embedding]).astype('float32'))
# 在索引中搜索与输入嵌入向量最相似的k个向量
distances, indices = index.search(input_embedding, k)
# 初始化一个字符串来存储匹配的文本
matching_texts = ""
# 遍历搜索结果
for i, idx in enumerate(indices[0]):
# 打印每个匹配文本块的相似度和文本内容
print(f"similarity: {distances[0][i]:.4f}\nmatching text: \n{chunks[idx]}\n")
# 将相似度和文本内容添加到匹配文本字符串中
matching_texts += f"similarity: {distances[0][i]:.4f}\nmatching text: \n{chunks[idx]}\n"
# 返回包含匹配文本块及其相似度的字符串
return matching_texts
调用测试：
input_text = "教程使用了哪些模型"
matching_texts = match_text(input_text, index=index, chunks=chunks, k=3)
print(matching_texts)
有了检索函数match_text，接下来我们写一个简单的调用即可完成一个简单的RAG检索demo，运行逻辑：
输入要检索的问题 调用 match_text 从向量库中查找出相似度最高的3个记录 将查找出的内容和问题一起丢给大模型，让 大模型 基于查找内容进行回复
'''
代码来源：github.com/datawhalechina/wow-rag
'''
def get_completion_stream(prompt):
"""
使用 OpenAI 的 Chat Completions API 生成流式的文本回复。
参数:
prompt (str): 要生成回复的提示文本。
返回:
None: 该函数直接打印生成的回复内容。
"""
# 使用 OpenAI 的 Chat Completions API 创建一个聊天完成请求
response = client.chat.completions.create(
model=chat_model, # 填写需要调用的模型名称
messages=[
{"role": "user", "content": prompt},
],
stream=True,
)
# 如果响应存在
if response:
# 遍历响应中的每个块
for chunk in response:
# 获取当前块的内容
content = chunk.choices[0].delta.content
# 如果内容存在
if content:
# 打印内容，并刷新输出缓冲区
print(content, end='', flush=True)
input_text = "教程使用了哪些模型"
matching_texts = match_text(input_text, index=index, chunks=chunks, k=3)
prompt = f"""
根据找到的文档
{matching_texts}
生成
{input_text}
的答案，尽可能使用文档语句的原文回答。不要复述问题，直接开始回答，注意使用中文回复。
"""
get_completion_stream(prompt)
恭喜你，跟着跑到这里，你就完成了一个最简单的 RAG 项目，通过这个项目你可以很直观的了解到 RAG检索增强生成系统的运行流程、核心思路，这看似很简单，实际还有很多需要优化的地方，比如：怎么切分导入文本才是最优的 、数据如何长期存储、导入数据向量化操作如何并发等等。
欢迎大家去wow-rag项目学习后面更多内容，有什么问题或者想法也欢迎大家评论区交流反馈，也可以直接加我微信交流。
https://github.com/datawhalechina/wow-rag
今天就分享到这，我是老表，想要成为一名终身学习者。
阅读推荐
《AI群星闪耀时》是一部由清华大学刘知远团队倾力打造的作品，讲述了28位人工智能先驱的传奇人生和他们对AI发展的巨大贡献。本书以通俗易懂的语言，将科技故事描绘得热血沸腾，是了解人工智能发展史的绝佳读物。
翻开它，感受科技之光如何照亮人类历史。
拓展学习：内积是什么？怎么计算的
1. 内积索引的概念
Faiss 是一个高效的相似性搜索库，其中 内积索引 (IndexFlatIP) 是一种基于内积的向量相似性搜索方法。
内积（Inner Product）：内积是一种向量相似性的度量方法，对于两个向量 (a) 和 (b)，内积定义为：
内积的值越大，表示两个向量的方向越接近（假设向量已归一化）。
Faiss 内积索引的逻辑：
存储多个高维向量。 在查询时，比较查询向量和索引中存储的向量的内积值，返回最相似的向量。
2. 内积索引的转换逻辑
内积索引的转换和使用逻辑分为以下几步：
2.1 数据准备和归一化
在内积索引中，向量的模会影响内积值的大小：
若向量未归一化，则内积值不仅受方向影响，还受模长的影响。 如果向量经过归一化（模为1），则内积等价于余弦相似性。
因此，案例代码中使用了 sklearn.preprocessing.normalize 将嵌入向量归一化，确保仅计算方向的相似性。
2.2 向量存储
创建一个内积索引 faiss.IndexFlatIP(d)，其中d是向量的维度。使用 index.add(normalized_embeddings)将归一化后的向量存入索引。
2.3 向量检索
在检索阶段：
提供一个查询向量 (q)。 Faiss 会计算查询向量与所有索引向量的内积：其中 (x_i) 是索引中的向量。 返回内积值最高的 (k) 个向量及其索引位置。
3. 使用场景
内积索引在许多机器学习和自然语言处理任务中非常常用，以下是几个典型场景：
3.1 语义搜索
场景：基于用户输入的查询语句，找到最相关的文档。 逻辑： 将文档和查询语句转化为嵌入向量。 在索引中搜索与查询语句嵌入向量最接近的文档嵌入向量。 应用： 搜索引擎（如知识库检索、FAQ 系统）。 嵌入式问答。
3.2 推荐系统
场景：基于用户的兴趣向量，找到与其兴趣最相似的产品或内容。 逻辑： 将用户历史行为转化为嵌入向量。 在索引中检索与用户兴趣向量最相似的内容嵌入向量。 应用： 商品推荐（如电商平台）。 内容推荐（如电影、音乐、新闻）。
3.3 聚类分析
场景：在高维向量空间中找到相似的向量组。 逻辑： 通过 Faiss 的内积索引，快速找到向量间的邻居。 应用： 社交网络分析（如好友推荐）。 数据分组（如特征相似样本的分簇）。
3.4 异常检测
场景：识别数据集中异常或不相似的向量。 逻辑： 计算每个向量的邻居。 若向量的内积分数低于某一阈值，则标记为异常。 应用： 网络安全中的异常检测。 数据清洗中的异常识别。
4. 如何使用存储的数据
4.1 插入更多向量
可以通过 index.add(new_vectors)动态添加新的嵌入向量，扩展索引的存储。
4.2 查询相似向量
使用 index.search(query, k)检索最相似的 (k) 个向量：D, I = index.search(query_vector, k)D是内积分数。I是索引中相似向量的编号。
4.3 保存和加载索引
保存索引：将索引存储到文件以供后续使用： faiss.write_index(index, "index_file_name")加载索引：从文件加载索引以快速恢复： index = faiss.read_index("index_file_name")
4.4 删除数据
原始的 IndexFlatIP 不支持删除操作。如果需要支持删除，可以使用 faiss.IndexIDMap 或 faiss.IndexIVFFlat 这样的高级结构。
5. 总结
内积索引适合归一化后的嵌入向量，用于高效的相似性检索 在语义搜索、推荐系统、异常检测等场景中非常有用 存储的数据可以动态更新，并支持快速查询和持久化保存
参考
[1 ] Datawhale wow-rag 教程开源项目：https://github.com/datawhalechina/wow-rag [2] 向量的内积（点乘）：https://www.cnblogs.com/gxcdream/p/7597865.html [3] Faiss index进阶操作：https://blog.csdn.net/u013066730/article/details/106298939 [4] 向量内积在机器学习中的重要性：https://juejin.cn/post/7320855522307489843 [5] 使用检索增强生成构建语义搜索：https://medium.com/orkes/rag-explained-using-retrieval-augmented-generation-to-build-semantic-search-97a3495df892
扫码加老表微信，进入学习交流群
