【AI 大模型】RAG 检索增强生成 ⑥ ( 使用 向量数据库 作为 RAG 知识库 完整实现 )

上一篇博客 【AI 大模型】RAG 检索增强生成 ⑤ ( 向量数据库 | 向量数据库 索引结构和搜索算法 | 常见 向量数据库 对比 | 安装并使用 向量数据库 chromadb 案例 ) 中 , 简单介绍了 向量数据库 概念 , 以及对常用的 向量数据库 进行了对比 , 并给出了 chromadb 向量数据库 的 实际使用案例 , 将 文本 使用 OpenAI 的 text-embedding-ada-002 文本向量模型 转为 文本向量 , 然后将这些文本向量 存储到 chromadb 向量数据库 中 , 给出一个 文本 并转为 文本向量 , 从 向量数据库中 查询 存入的 文本数据 ;

一、本地知识库 搭建

1、大模型弊端

大模型弊端 : 大模型 有强大的 语言处理和理解 能力 , 但是 大模型本身并不具备 最新的数据 以及 特定领域的专有数据 , 一旦大模型没有相关知识内容 , 就会出现 " 幻觉 " , 开始 胡编乱造 ;

• 数据迟滞 : 大模型 训练数据 往往存在一定的时滞性 , 如 : 1 月收集的数据 开始训练 , 要花半年才能训练完毕 , 使用大模型时是没有最近半年的数据的 ;
• 数据缺失 : 特定领域的专业数据 也可能并不包含在大模型的训练集中 , 如 : 公司或行业的 保密数据 , 商业机密等 ; 大模型 在回答相关领域问题时 缺乏准确性 ;

2、向量数据库 -> 本地知识库

为了克服 大模型的 数据迟滞 和 数据缺失 弊端 , 可以通过 向量数据库 来 构建一个本地化的 知识库 , 以实现 特定领域知识 的 实时更新 , 进而 开发基于 本地知识库的智能问答系统 ;

本地知识库 作为 大模型的补充 , 存储

• 最新数据
• 特定领域的专有知识 ;

本地知识库 参与 提示词构建 : 通过 向量数据库 的 向量检索技术 , 我们可以 快速、准确地从知识库 中提取出与用户查询相关的信息 , 然后将本地知识 构建到 Prompt 提示词 中 ;

本地知识库 数据维护 : 本地知识库 应该 建立一套完善的 知识库 更新机制 , 当新的 数据或领域知识 出现时 , 可以及时 将 新知识 添加到 向量数据库 中 , 确保 本地知识库始终 保持最新、最全面的状态 ;

通过向量数据库搭建本地知识库 , 并实现领域知识的实时更新 , 可以有效解决大模型缺乏最新数据和特定领域专有数据的问题 , 充分利用 本地知识库 中的最新数据和专有知识 , 提高 大模型输出的准确性 ;

二、RAG 检索增强生成 三阶段

1、RAG 三阶段简介

RAG ( Retrieval-Augmented Generation , 检索增强生成 )是一种结合 信息检索 与 生成模型 的技术 , 通过 外部知识库 提升 生成内容的 准确性 和 相关性 ;

RAG 流程可分为 三阶段 :

• 检索 Retrieval : 向量相似度 匹配本地知识
• 增强 Augmented : 本地知识库信息 注入提示词
• 生成 Generation : 大模型整合提示词 输出

RAG 积极意义 :

• 动态知识更新 : 无需 重新训练模型 即可通过 更新 本地知识库 适应 新领域数据 或 实时数据 ;
• 可解释性增强 : 生成结果可追溯至检索内容 , 便于验证来源可信度。

RAG 适用场景 : 问答系统、客服对话、长文本生成（如报告撰写）等需结合外部知识的任务 ;

2、RAG 三阶段 流程图

使用 Mermaid 语法绘制流程图 :

graph TD
    A[用户提出问题] --> B{查询 本地 向量数据库}
    B --> C[根据 语义相似度 获取本地知识]
    C --> D[Prompt 提示词构建]
    D --> E{提示词 输入 大模型}
    E --> F[获得最终回答]

graph 是指要绘制的图表类型 ,

• TD 是 指方向为 自上而下 Top Down
• LR 从左到右
• BT 从下到上
• RL 从右到左

3、检索 Retrieval - 向量相似度匹配

从 本地知识库 中快速 定位 与用户查询 最相关的信息片段 , 涉及到如下技术 :

• 文本向量转化 : 使用 嵌入模型 ( 文本向量模型 ) , 如 : BERT、Sentence-BERT , 将 本地知识 转换为高维向量 ;
• 文本向量相似度计算 : 通过 余弦距离 或 欧氏距离 衡量查询向量与文档向量间的相关性 ;
• 向量高效检索 : 借助近似最近邻算法（如FAISS、HNSW）或 搜索引擎（如Elasticsearch）加速大规模向量匹配 ; 这里推荐使用 向量数据库 ;
• 关键技术问题 : 平衡检索速度与准确性 , 处理不同语言的语义多样性 , 如 : 同义词、多义词 ;

4、增强 Augmented - 知识库信息注入

将检索到的关键信息整合到生成模型的输入中 , 也就是 Prompt 提示词 , 引导 大语言模型 生成更可靠的回答 ;

• 提示工程 : 在输入提示中拼接检索到的文本（如“基于以下内容回答：[检索结果]+用户问题”）。
• 注意力增强 : 通过模型架构设计（如Fusion-in-Decoder），让生成过程更聚焦于检索内容。
• 减少幻觉 : 减少生成模型的“幻觉”（虚构事实），提升回答的可信度。
• 关键问题 : 过滤噪声信息，避免信息过载对生成质量的影响。

5、生成 Generation - 大模型整合输出

基于检索信息和模型内部知识，生成自然、连贯且准确的最终回答。

• 模型选择 : 通常采用大语言模型（如GPT-4、Llama、PaLM），利用其强大的语言理解和生成能力。
• 生成策略 : 根据场景选择束搜索（保证连贯性）或采样（增加多样性），并可能加入后处理（如去重、引用标注）。
• 关键技术点 : 回答的准确性、与检索内容的一致性，以及逻辑流畅性。

三、完整代码示例

1、生成本地知识库代码示例

参考上一篇博客 【AI 大模型】RAG 检索增强生成 ⑤ ( 向量数据库 | 向量数据库 索引结构和搜索算法 | 常见 向量数据库 对比 | 安装并使用 向量数据库 chromadb 案例 ) 中 , 向量数据库 保存 相关代码 , 使用 ChromaDB 向量数据库 , 在 工程的根目录 创建 chroma_db 数据库 , 在该数据库 中创建 news_articles 集合 , 也就是数据库表 , 使用该数据库表 作为 本地知识库 的载体 ;

当使用 RAG 服务时 , 首先从该 chroma_db 向量数据 中的 news_articles 集合 中 , 搜索 指定个数 的 相似结果 , 然后将这些相似的 文本 拼接到 提示词 中 , 再将提示词输入到 LLM 大语言模型 中 , 得到输出结果 ;

代码示例 ( 可独立运行 ) : 下面代码中的 OpenAI 的 API Key 和 服务地址 需要替换成自己购买的 ;

# 导入所需库
import chromadb  # ChromaDB 向量数据库
from openai import OpenAI  # OpenAI 客户端

# 初始化 OpenAI 客户端 (替换成自己的 API 信息)
client = OpenAI(
    api_key="sk-i3daF6",  # 替换为你的 OpenAI API Key , 这里我把自己的 API-KEY 隐藏了
    base_url="https://api.xiaoai.plus/v1"  # 替换为你的 API 服务端点
)


def get_embeddings(texts, model="text-embedding-ada-002"):
    """将文本转换为向量表示
    Args:
        texts: 需要编码的文本列表
        model: 使用的嵌入模型（默认OpenAI官方推荐模型）
    Returns:
        包含向量数据的列表，每个元素对应输入文本的768维向量
    """
    response = client.embeddings.create(
        input=texts,
        model=model
    )
    # 从响应中提取向量数据
    return [item.embedding for item in response.data]


# 初始化 ChromaDB 客户端 (持久化到本地目录)
chroma_client = chromadb.PersistentClient(path="chroma_db")

# 创建或获取集合 (相当于数据库表)
collection = chroma_client.get_or_create_collection(
    name="news_articles",  # 集合名称
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# 原始文本数据集（示例新闻标题）
documents = [
    "李彦宏称大模型成本每年降低90%",  # 科技类
    "乌军大批直升机击落多架俄无人机",  # 国际争端
    "王力宏回应是否想找新伴侣",      # 娱乐新闻
    "饺子不知道观众怎么想出的藕饼cp",  # 影视相关
    "加沙停火协议关键时刻生变"       # 国际争端
]

# 批量生成文档向量（OpenAI API调用）
document_embeddings = get_embeddings(documents)

# 生成唯一文档 ID (需要唯一标识符)
document_ids = [str(i) for i in range(len(documents))]  # 生成 ["0", "1", ..., "4"]

# 将文档插入数据库
collection.add(
    ids=document_ids,  # 唯一ID列表
    embeddings=document_embeddings,  # 文本向量列表
    documents=documents  # 原始文本列表
)

2、生成本地知识库执行结果

上述代码执行后 , 在本地生成数据库文件 ;

3、RAG 服务实现

RAG 流程可分为 三阶段 : 这 三个阶段 , 分别有对应的代码模块 ;

• 检索 Retrieval : 向量相似度 匹配本地知识
• 增强 Augmented : 本地知识库信息 注入提示词
• 生成 Generation : 大模型整合提示词 输出

① 检索 Retrieval 模块

检索 Retrieval : 向量相似度 匹配本地知识

• 首先 , 获取 chroma_db 向量数据库 以及 数据库 中的 news_articles 集合 ( 数据库表 ) , 这是 上一步 创建的 chromadb 向量数据库表 ;

    # 初始化向量数据库连接
    chroma_client = chromadb.PersistentClient(path="chroma_db")
    # 创建或获取集合 (相当于数据库表)
    collection = chroma_client.get_or_create_collection(
        name="news_articles",  # 集合名称
        metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
    )

• 然后 , 将要查询的 文本 转为 文本向量 ;

        # 1. 向量检索, 生成查询向量, 将文本转为向量
        query_embedding = get_embeddings([user_query])[0]

• 最后 , 从 news_articles 数据库表 中 , 查询 目标文本向量 中最相近的 2 个文本 ;

        # 执行相似性查询
        search_results = collection.query(
            query_embeddings=[query_embedding],  # 查询向量
            n_results=2  # 返回前2个最相似结果
        )

② 增强 Augmented 模块

增强 Augmented : 本地知识库信息 注入提示词

• 首先 , 给定一个提示词模板 prompt_template , 其中的 __INFO__ 需要替换为 向量数据库 中查询的内容 , __QUERY__ 需要替换为 用户的提问 ;

# 提示词模板（控制模型输出格式）
prompt_template = """
你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。

已知信息 : 
__INFO__

用户提问 : 
__QUERY__

"""

• 然后 , 使用下面 的 build_prompt 函数 , 处理上述 提示词模板 , 分别传入 info=search_results['documents'][0] 和 query=user_query 两个 键值对参数 , 这样就将 __INFO__ 和 __QUERY__ 替换为了指定的文本 ;

def build_prompt(**kwargs):
    """动态构建提示词
    Args:
        prompt_template: 提示词模板字符串
        **kwargs: 需要替换的键值对（自动匹配__KEY__占位符）
    Returns:
        完成替换后的完整提示词
    """
    prompt = prompt_template
    for k, v in kwargs.items():
        # 处理不同类型的数据输入
        if isinstance(v, str):
            val = v
        elif isinstance(v, list) and all(isinstance(elem, str) for elem in v):
            val = '\n'.join(v)  # 列表转换为多行文本
        else:
            val = str(v)  # 其他类型转为字符串
        prompt = prompt.replace(f"__{k.upper()}__", val)
    print("构建提示词:" + prompt)
    return prompt

• 传参的实际调用代码为 :

        # 2. 构建增强提示
        prompt = build_prompt(
            info=search_results['documents'][0],  # 取最相关结果
            query=user_query
        )

③ 生成 Generation 模块

生成 Generation 模块 很简单 , 就是个 普通的 LLM 问答模块 , 调用下面的 函数 , 传入 构建好的 Prompt 提示词 , 完成 问答 ,得到一个 LLM 输出的内容 ;

def get_completion(prompt, model="gpt-3.5-turbo"):
    """封装OpenAI API调用
    Args:
        prompt: 完整的提示词内容
        model: 选择的大模型版本（默认gpt-3.5-turbo）
    Returns:
        模型生成的文本响应
    """
    messages = [{"role": "user", "content": prompt}]
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,  # 控制输出随机性（0为最确定性输出）
    )
    return response.choices[0].message.content  # 提取响应内容

实际的调用代码如下 :

        # 3. 调用大模型生成
        return self.llm_api(prompt)

④ 完整代码

完整代码示例 ( 可独立运行 ) : 下面的代码中 , 替换成自己的 OpenAI 的 api_key 和 base_url 即可 , 注意 Python 要使用 3.9 以上版本 , 否则无法成功安装 chromadb 向量数据库 ;

import chromadb
from openai import OpenAI

# 初始化 OpenAI 客户端 (替换成自己的 API 信息)
client = OpenAI(
    api_key="sk-i3dHqaF6",  # 替换为你的 OpenAI API Key , 这里我把自己的 API-KEY 隐藏了
    base_url="https://api.xiaoai.plus/v1"  # 替换为你的 API 服务端点
)

# 提示词模板（控制模型输出格式）
prompt_template = """
你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。

已知信息 : 
__INFO__

用户提问 : 
__QUERY__

"""

def build_prompt(**kwargs):
    """动态构建提示词
    Args:
        prompt_template: 提示词模板字符串
        **kwargs: 需要替换的键值对（自动匹配__KEY__占位符）
    Returns:
        完成替换后的完整提示词
    """
    prompt = prompt_template
    for k, v in kwargs.items():
        # 处理不同类型的数据输入
        if isinstance(v, str):
            val = v
        elif isinstance(v, list) and all(isinstance(elem, str) for elem in v):
            val = '\n'.join(v)  # 列表转换为多行文本
        else:
            val = str(v)  # 其他类型转为字符串
        prompt = prompt.replace(f"__{k.upper()}__", val)
    print("构建提示词:" + prompt)
    return prompt


def get_completion(prompt, model="gpt-3.5-turbo"):
    """封装OpenAI API调用
    Args:
        prompt: 完整的提示词内容
        model: 选择的大模型版本（默认gpt-3.5-turbo）
    Returns:
        模型生成的文本响应
    """
    messages = [{"role": "user", "content": prompt}]
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0,  # 控制输出随机性（0为最确定性输出）
    )
    return response.choices[0].message.content  # 提取响应内容

def get_embeddings(texts, model="text-embedding-ada-002"):
    """将文本转换为向量表示
    Args:
        texts: 需要编码的文本列表
        model: 使用的嵌入模型（默认OpenAI官方推荐模型）
    Returns:
        包含向量数据的列表，每个元素对应输入文本的768维向量
    """
    response = client.embeddings.create(
        input=texts,
        model=model
    )
    # 从响应中提取向量数据
    return [item.embedding for item in response.data]

class RAG_Service:
    def __init__(self, vector_db, llm_api, n_results=2):
        """初始化RAG机器人
        Args:
            vector_db: 已初始化的向量数据库连接对象
            llm_api: 大模型API调用函数
            n_results: 默认检索结果数量
        """
        self.vector_db = vector_db  # 向量数据库实例
        self.llm_api = llm_api  # LLM调用接口
        self.n_results = n_results  # 检索结果数量

    def chat(self, user_query):
        """处理用户查询的完整流程
        Args:
            user_query: 用户输入的自然语言问题
        Returns:
            结合知识库生成的回答
        """

        # 1. 向量检索, 生成查询向量, 将文本转为向量
        query_embedding = get_embeddings([user_query])[0]
        # 执行相似性查询
        search_results = collection.query(
            query_embeddings=[query_embedding],  # 查询向量
            n_results=2  # 返回前2个最相似结果
        )

        # 2. 构建增强提示
        prompt = build_prompt(
            info=search_results['documents'][0],  # 取最相关结果
            query=user_query
        )

        # 3. 调用大模型生成
        return self.llm_api(prompt)


# RAG 使用示例
if __name__ == "__main__":
    # 初始化向量数据库连接
    chroma_client = chromadb.PersistentClient(path="chroma_db")
    # 创建或获取集合 (相当于数据库表)
    collection = chroma_client.get_or_create_collection(
        name="news_articles",  # 集合名称
        metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
    )

    # 创建 RAG 实例
    bot = RAG_Service(
        vector_db=collection,
        llm_api=get_completion
    )

    # 示例查询
    user_query = "你对国际争端新闻的评价"

    # 生成查询向量
    query_embedding = get_embeddings([user_query])[0]

    # RAG 回答
    response = bot.chat(user_query)

    print("RAG 系统回答：", response)

4、RAG 服务 执行结果

代码执行后 , 用户提出的问题是 " 你对国际争端新闻的评价 " ,

从本地知识库中 , 查询 与 用户问题 相似的 知识文档 , 查询结果是 :

加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机

将上述查询结果 , 拼接到 提示词中 , 则 完整的提示词为 :

你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。

已知信息 : 
加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机

用户提问 : 
你对国际争端新闻的评价

将上述提示词 输入到 OpenAI 大模型 中 , 得到的结果为 : " 作为国际军事问题专家，我认为国际争端新闻总是令人担忧的。加沙停火协议的变化和乌克兰军队击落俄罗斯无人机的事件都表明国际局势仍然不稳定。这些事件可能会导致更多的紧张局势和冲突，需要各方保持冷静和谨慎，寻求通过对话和外交手段解决分歧。国际争端的解决需要各方共同努力，避免采取激烈的行动，以避免进一步升级局势。 "

执行结果 :

D:\001_Develop\022_Python\Python39\python.exe D:/002_Project/011_Python/OpenAI/rag_service.py
构建提示词:
你是一名国际军事问题专家。你的任务是根据现有知识库和用户问题给出最佳答案。

已知信息 : 
加沙停火协议关键时刻生变
乌军大批直升机击落多架俄无人机

用户提问 : 
你对国际争端新闻的评价


RAG 系统回答： 作为国际军事问题专家，我认为国际争端新闻总是令人担忧的。加沙停火协议的变化和乌克兰军队击落俄罗斯无人机的事件都表明国际局势仍然不稳定。这些事件可能会导致更多的紧张局势和冲突，需要各方保持冷静和谨慎，寻求通过对话和外交手段解决分歧。国际争端的解决需要各方共同努力，避免采取激烈的行动，以避免进一步升级局势。

Process finished with exit code 0