一. 基础概念

1  LLM的缺陷

   (1) LLM有很强的时效性，比如该模型训练到25年12月，那么26年以后的知识它是不知道。

   (2) 在一些深层次的专业领域，LLM训练不到位。

   引入RAG的初衷：① 增强LLM的事实性和时效性；② 引入专业知识优化生成回复。

2  RAG概念

   (1) 顾名思义： 检索外部数据，增强大模型的生成。

   (2) 举例:   我想询问LLM有关于我公司内部的请假流程它肯定不知道，但是我把公司的员工手册先给它看，当我在询问的时候他就会知道。

3  RAG的工作流程

  它主要包括三个步骤：

  (1) 向量化：将文档切分成片段，再编码为向量，存储在向量数据库里。

  (2) 检索：检索到与问题相似度最高的top k 的片段。

  (3) 生成：将原始问题和检索得到的“块”都作为LLM的输入，令其生成最终的回答。

补充RAG与微调对比：

二. 文档分块

1 按句子切分

代码分享

import re

text = (
    "自然语言处理（NLP），作为计算机科学、人工智能与语言学的交融之地致力于赋予计算机解析和处理人类语言的能力。在这个领域，"
    "机器学习发挥着至关重要的作用。利用多样的算法，机器得以分析、领会乃至创造我们所理解的语言。"
    "从机器翻译到情感分析，从自动摘要到实体识别，NLP的应用已遍布各个领域。随着深度学习技术的飞速进步，"
    "NLP的精确度与效能均实现了巨大飞跃。如今，部分尖端的NLP系统甚至能够处理复杂的语言理解任务，"
    "如问答系统、语音识别和对话系统等。NLP的研究推进不仅优化了人机交流，也对提升机器的自主性和智能水平起到了关键作用。"
)

# 按中文句号、问号、感叹号、省略号分割文本
sent = re.split(r"(。|？|！|\...\...)", text)
print(sent)  # 将分隔符也保留下来

# 处理分隔符的问题1
chunks = []
for sen, pun in zip(sent[::2], sent[1::2]):  # sent[::2]取偶数下标  sent[1::2]取奇数下标
    chunks.append(sen + pun)
print(chunks)

# 处理分隔符的问题2
# pun2为空则不追加
chunks2 = []
for sen2, pun2 in zip(sent[::2], sent[1::2]):
    chunks2.append(sen2 + (pun2 if pun2 else ""))
print(chunks2)
# enumerate 遍历（带索引）
for i, chunk in enumerate(chunks2):
    print(f"块{i + 1}的长度为 {len(chunk)}: {chunk}")

2 按固定字符切分

代码分享

text = (
    "自然语言处理（NLP），作为计算机科学、人工智能与语言学的交融之地致力于赋予计算机解析和处理人类语言的能力。在这个领域，"
    "机器学习发挥着至关重要的作用。利用多样的算法，机器得以分析、领会乃至创造我们所理解的语言。"
    "从机器翻译到情感分析，从自动摘要到实体识别，NLP的应用已遍布各个领域。随着深度学习技术的飞速进步，"
    "NLP的精确度与效能均实现了巨大飞跃。如今，部分尖端的NLP系统甚至能够处理复杂的语言理解任务，"
    "如问答系统、语音识别和对话系统等。NLP的研究推进不仅优化了人机交流，也对提升机器的自主性和智能水平起到了关键作用。"
)
chunks = []
for i in range(0, len(text), 50):  # 从0开始，到len(text)结束，步长为50
    print(text[i : i + 50])
    chunks.append(text[i : i + 50])  # 每 50 个字符切一段
print(chunks)

3 封装固定字符分割

代码分享

def overlap_split(text, step):
    """
    对文本进行带步长的切片
    :param text: 原始文本
    :param step: 步长
    """
    list = []
    for i in range(0, len(text), step):
        list.append(text[i : i + step])
    return list


res = overlap_split(text, 30)
print(res)

4 递归分割

使用LangChain框架实现

代码分享

"""
1. 这里使用0.3.7 LangChain，内置了 angchain-text-splitters
   pip install langchain==0.3.7  -i https://pypi.tuna.tsinghua.edu.cn/simple
2. LangChain 1.x中 单独安装langchain-text-splitters, 因为 RecursiveCharacterTextSplitter 现在不在 langchain 里了
   pip install langchain-text-splitters -i https://pypi.tuna.tsinghua.edu.cn/simple
解决版本冲突
    pip uninstall langchain langchain-core langchain-text-splitters -y
强制安装，自动解决冲突
    pip install --force-reinstall langchain-text-splitters -i https://pypi.tuna.tsinghua.edu.cn/simple
"""

from langchain_text_splitters import RecursiveCharacterTextSplitter

text = (
    "自然语言处理（NLP），作为计算机科学、人工智能与语言学的交融之地致力于赋予计算机解析和处理人类语言的能力。在这个领域，"
    "机器学习发挥着至关重要的作用。利用多样的算法，机器得以分析、领会乃至创造我们所理解的语言。"
    "从机器翻译到情感分析，从自动摘要到实体识别，NLP的应用已遍布各个领域。随着深度学习技术的飞速进步，"
    "NLP的精确度与效能均实现了巨大飞跃。如今，部分尖端的NLP系统甚至能够处理复杂的语言理解任务，"
    "如问答系统、语音识别和对话系统等。NLP的研究推进不仅优化了人机交流，也对提升机器的自主性和智能水平起到了关键作用。"
)
#  chunk_size：每一段文本最多30个字符，你设置多少，就是多少
#  chunk_overlap：两段之间重叠的字数，主要是为了防止语义被切断，这里10表示：第一段末尾 10 个字 = 第二段开头 10 个字
#  length_function=len ：表示按字符个数计算（中文、英文、数字都算 1 个）
fun = RecursiveCharacterTextSplitter(chunk_size=30, chunk_overlap=10, length_function=len)
data = fun.split_text(text)
print(data)

三. 向量与Embedding

1 向量

  向量在数学中，指的是：具有大小和方向的量，可以形象化的表示为带箭头的线段。 箭头所指：代表向量的方向；线段长度：代表向量的大小。

 (1)  将文本转成一组浮点数：每个下标 i，对应一个维度

 (2)  整个数组对应一个 n 维空间的一个点

 (3)  向量之间可以计算距离，距离的远近 对应 语义相似度的大小。

2 Embedding

 嵌入: 表示学习的一种形式，通常用于将高维数据映射到低维空间中的表示形式。嵌入可以是词嵌入、图像嵌入、图嵌入等。

 例如，在自然语言处理中，词嵌入将词语映射到低维向量空间，以捕捉词语之间的语义和句法关系。在图像处理中，图像嵌入将图像映射到低维向量空间，以表示图像的视觉特征。

补充：向量和embeddings区别：

(1) embeddings可以理解成一个模型(text-embeddings-v3)，一个工具，它可以把文本、图片、语音 → 变成固定长度的数字数组

(2) 向量就是一串数字，embeddings后的输出结果。

代码分享：

from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()
client = OpenAI(api_key=os.getenv("api_key"), base_url=os.getenv("base_url"))
def get_embedding(text):
    res = client.embeddings.create(input=text, model="text-embedding-v3")
    return [i.embedding for i in res.data]

test_query = ["我爱你"]
vec = get_embedding(test_query)
# print(vec)  # 形如[[xxx,xxx,]] 所以 vec[0]，才是向量数组
print(vec[0])  # 输出向量
print(len(vec[0]))  # 输出向量的长度

3 向量间的相似度计算

    常用形似都计算的方法：

(1)  余弦距离Cosine

   基于两个向量夹角的余弦值来衡量相似度。

代码分享：

(2) 欧式距离L2

 通过计算向量之间的欧几里得距离来衡量相似度。

代码分享：

(3) 点积

 计算两个向量的点积，适合归一化后的向量。

代码分享：

 import numpy as np
from numpy import dot
from numpy.linalg import norm
from openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()
client = OpenAI(api_key=os.getenv("api_key"), base_url=os.getenv("base_url"))

# 余弦距离
def cos_sim(a, b):
    """余弦距离   越大越相似"""
    return dot(a, b) / (norm(a) * norm(b))

# 欧式距离
def l2(a, b):
    """距离关系 越小越相似"""
    x = np.asarray(a) - np.asarray(b)
    return norm(x)

# 封装获取embedding的方法
def get_embedding(text):
    data = client.embeddings.create(input=text, model="text-embedding-v3")
    return [i.embedding for i in data.data]


# # # # # # # # # # # # # # # # # #下面开始调用 # # # # # # # # # # # # # # # # # # # # # # # # # # # #

query = "我国开展舱外辐射生物学暴露实验"
documents = [
    "联合国就苏丹达尔富尔地区大规模暴力事件发出警告",
    "土耳其、芬兰、瑞典与北约代表将继续就瑞典“入约”问题进行谈判",
    "日本岐阜市陆上自卫队射击场内发生枪击事件 3人受伤",
    "国家游泳中心（水立方）：恢复游泳、嬉水乐园等水上项目运营",
    "我国首次在空间站开展舱外辐射生物学暴露实验",
]

query_vec = get_embedding([query])[0]
doc_vec = get_embedding(documents)
# print(query_vec)

print("余弦距离")
# 余弦距离
print(f"完全一样的两个值的余弦距离：{cos_sim(query_vec, query_vec)}")  # 完全一样，最大值为1
for vec in doc_vec:
    print(cos_sim(query_vec, vec))

print("欧式距离")
# 欧式距离
print(f"完全一样的两个值的余弦距离：{l2(query_vec, query_vec)}")  # 距离越近，越相似 ，完全一样为 0
for vec in doc_vec:
    print(l2(query_vec, vec))

4 常用的向量数据库

(1) 常见向量数据库

(2) chromadb的使用

详见：https://zhuanlan.zhihu.com/p/680661442 

安装模块：

 (1)  需要先安装C++ 14的编译环境, 使用vs的安装包，安装C++开发桌面程序即可

 （2) 【pip install chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple】           【速度非常快  安装的是 1.5.5 版本】

代码分享：

"""
安装步骤
    1 需要先安装C++ 14的编译环境, 使用vs的安装包，安装C++开发桌面程序即可
    2 pip install chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple 【1.5.5 版本】  推荐使用这个哦
"""

import chromadb

# 初始化方式1：基于内存
client = chromadb.Client()
# 初始化方式2：存放到本地
# client = chromadb.PersistentClient(path=r"D:\DB")
# 初始化客户端 (这里的表叫做collection，表名为：test)
cll = client.get_or_create_collection(name="test")

# 添加数据
cll.add(
    # 文本内容列表，每个元素是一段文本（如文章、句子等）
    documents=["Article by john", "Article by Jack", "Article by Jill"],
    # 嵌入向量列表，每个元素是一个与 documents 对应的向量表示
    embeddings=[[1, 2, 3], [4, 5, 6], [7, 8, 9]],
    # 自定义 ID 列表，用于唯一标识每条记录
    ids=["1", "2", "3"],
)

# 查询数据--参数化查询，查询里面含有by的
aa = cll.get(ids=["1"], include=["embeddings"], where_document={"$contains": "by"})
print(aa)

# ========== 修正1：精准ID查询（用get方法，仅按ID查，无文本过滤） ==========
# get方法仅支持ids/where（元数据过滤），不支持where_document
result_by_id = cll.get(ids=["1"], include=["embeddings", "documents"])  # 仅按ID查询  # 指定返回向量+文本
print("【按ID精准查询结果】")
print(result_by_id)

# 删除数据
cll.delete(ids=["1"])
print(cll.get())

# 修改数据
cll.update(
    documents=["Article 12323456", "Article xxxxxxx", "Article by Jill"],  # 文本内容列表，每个元素是一段文本（如文章、句子等）
    embeddings=[[1, 2, 3], [4, 5, 6], [7, 8, 9]],  # 嵌入向量列表，每个元素是一个与 documents 对应的向量表示
    ids=["1", "2", "3"],  # 自定义 ID 列表，用于唯一标识每条记录
)
print(cll.get())

(3) redis的使用

   windows版本【仅测试】：https://github.com/tporadowski/redis/releases

安装模块：

     【pip install redis -i https://pypi.tuna.tsinghua.edu.cn/simple】

代码分享：

"""
如何查找这个库都有哪些版本 :直接去 https://pypi.org/    或者   pip index versions redis
安装    pip install redis -i https://pypi.tuna.tsinghua.edu.cn/simple
这里安装的  7.3.0
"""

import json
import redis

# 创建 Redis 连接
r = redis.Redis(host="60.xxx.xxx.80", port=6379, password="xxxx", db=2, decode_responses=True)


# 从JSON文件中读取数据
def read_data():
    with open("../04-RAG基础/train_zh.json", "r", encoding="utf-8") as f:
        res = [json.loads(line) for line in f]
    instructions1 = [entry["instruction"] for entry in res[0:1000]]
    outputs1 = [entry["output"] for entry in res[0:1000]]
    return instructions1, outputs1


# 将读取出来的数据存入Redis中
def set_redis_documents(instructions1, outputs1):
    for instruction, output in zip(instructions1, outputs1):
        r.set(instruction, output)


# 在Redis中根据关键词进行模糊搜索
def search_instructions(instruction_key, top_n=3):
    keys = r.keys(pattern="*" + instruction_key + "*")
    res = []
    for key in keys:
        res.append(r.get(key))
    return res[:top_n]


# 先从文件中读取数据
instructions, outputs = read_data()
# 在把数据存入到Redis中
set_redis_documents(instructions, outputs)
# 在Redis中进行检索
data = search_instructions("怀孕")
print(data)

四. 基于RAG知识库实现AI系统案例

大致内容：

 (1) 读取pdf 和 word中的内容，进行切割。

   (2) 将分块后的内容进行 embeddings，将输出的向量存到 chromadb 数据库中

代码分享： 

# 先下载解析PDF和word的模块
# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple python-docx
# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pdfminer.six


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_chunks(text, chunk_size, stride):
    """
    滑动窗口切割文本
    :param text: 原始文本
    :param chunk_size: 每段长度
    :param stride: 滑动步长
    :return: 切割后的文本列表
    """
    return [text[i : i + chunk_size] for i in range(0, len(text), stride)]


# 读取PDF并割文档
def extract_text_from_pdf(filename, page_numbers=None, min_line_length=1):
    """从 PDF 文件中（按指定页码）提取文字"""
    """
    :param filename: PDF文件路径
    :param page_numbers: 要提取的页码，形如[1,2,3]
    :param min_line_length: 最小行长度
    :return: 分段后的文本列表
    """
    paragraphs = []
    full_text = ""
    # 提取全部文本
    for i, page_layout in enumerate(extract_pages(filename)):
        # 如果指定了页码范围，跳过范围外的页
        if page_numbers is not None and i not in page_numbers:
            continue
        for element in page_layout:
            # 检查element是不是文本
            if isinstance(element, LTTextContainer):
                # 将换行和空格去掉
                full_text += element.get_text().replace("\n", "").replace(" ", "")
    if full_text:
        # 调用切割函数
        text_chunks = sliding_window_chunks(full_text, 250, 100)
        for text in text_chunks:
            paragraphs.append(text)
    return paragraphs


# 向量类封装(与前面04类似)
class MyVectorDBConnector:
    # 默认构造函数
    def __init__(self):
        self.db = chromadb.Client()
        self.collection = self.db.get_or_create_collection(name="demo")

    # 向量化
    def get_embeddings(self, texts, model="text-embedding-v2"):
        data = client.embeddings.create(input=texts, model=model).data
        return [x.embedding for x in data]

    # 添加
    def add_documents(self, documents):
        embeddings = self.get_embeddings(documents)
        self.collection.add(
            embeddings=embeddings,
            documents=documents,
            ids=[f"id{i}" for i in range(len(documents))],
        )

    # 搜索
    def search(self, query, top_k):
        return self.collection.query(
            query_embeddings=self.get_embeddings([query]),
            n_results=top_k,
        )


# 封装机器人类--调用大模型
class RagBot:
    def __init__(self, vector_db, top_k=2):
        self.vector_db = vector_db
        self.top_k = top_k

    def get_completion(self, prompt, model="qwen-turbo"):
        messages = [{"role": "user", "content": prompt}]
        response = client.chat.completions.create(model=model, messages=messages, temperature=0)
        return response.choices[0].message.content

    def chat(self, user_query):
        # 1. 检索
        search_result = self.vector_db.search(user_query, self.top_k)
        context = "\n".join(search_result["documents"][0])
        # 2. 构建提示词
        prompt_template = """
            你是一个问答机器人。
            你的任务是根据下述给定的已知信息回答用户问题。
            确保你的回复完全依据下述已知信息。不要编造答案。
            如果下述已知信息不足以回答用户的问题，请直接回复"我无法回答您的问题"。

            已知信息:
            __INFO__

            用户问：
            __QUERY__

            请用中文回答用户问题。
            """
        prompt = prompt_template.replace("__INFO__", context).replace("__QUERY__", user_query)
        # 3. AI 回答
        reply = self.get_completion(prompt)
        print("AI回复：", reply)


# 主函数调用
if __name__ == "__main__":
    load_dotenv()
    client = OpenAI(api_key=os.getenv("api_key"), base_url=os.getenv("base_url"))
    page_data = extract_text_from_pdf("../04-RAG基础/财务管理文档.pdf", page_numbers=[0, 1, 2])

    # 创建向量数据库对象
    vector_db = MyVectorDBConnector()
    # 添加向量
    vector_db.add_documents(page_data)
    # 创建机器人对象
    rag_bot = RagBot(vector_db, top_k=2)
    rag_bot.chat("财务管理权限划分")