《构建和评估高级RAG》001: 实现高级RAG管道

分享  DeepLearning.AI 的课程：《构建和评估高级RAG》，第一部分：高级RAG管道。

主要内容：

• • 基础 RAG 管道和实现

• • 如何评估 RAG 程序

• • 高级RAG：语句窗口检索的实现

• • 高级RAG：自动合并检索的实现

• • 工具类源码

关于教程 教程是基于 LlamaIndex 和 trurea 联合推出的小课程，有兴趣的可以看英文视频教程：《Building and Evaluating Advanced RAG Applications ^[1]》。文章里的所有程序都是来自该教程，内容也是大差不大。

基础 RAG

第一篇讲高级的RAG管道， 既然有高级，那么相对的肯定有初级和基础， 我们先回顾一下基础 RAG 的整个流程：

这个经典的RAG流程图，我之前的文章里面提到过很多次，对于初学者能很清楚看明白 RAG 基础的三个步骤提取、召回、合成。

• • 提取：把文档向量化，存储到向量库

• • 召回：针对提问，从向量库里面获取最相关的内容

• • 合成：利用大语言模型的能力，把刚才从向量库查询到的内容结合提问形成最终的回答

使用程序实现也很简单。

引入相关包

import utils

import os
import openai
openai.api_key = utils.get_openai_api_key()

读取目录

from llama_index import SimpleDirectoryReader

documents = SimpleDirectoryReader(
    input_files=["./eBook-How-to-Build-a-Career-in-AI.pdf"]
).load_data()

文档对象

打印一下文档看看读取是否成功：

print(type(documents), "\n") # <class 'list'> 
print(len(documents), "\n")
print(type(documents[0])) # <class 'llama_index.schema.Document'>
print(documents[0]) # 文件对象的属性

可能是如下输出：

<class 'list'> 
41 
<class 'llama_index.schema.Document'>
Doc ID: caba8dde-0604-4305-91a3-6c40b7495dfd
Text: PAGE 1Founder, DeepLearning.AICollected Insights from Andrew Ng
How to  Build Your Career in AIA Simple Guide

Document 对象常用的几个属性：

• • doc_id: 唯一标识，例如：caba8dde-0604-4305-91a3-6c40b7495dfd

• • text: 文件文本内容

基础 RAG 实现

下面我们实现一下基础的 RAG 管道。

合并文件

把目录下的所有文件合并成一个文件：

from llama_index import Document

document = Document(text="\n\n".join([doc.text for doc in documents]))

构建 Index

使用 OpenAI 作为 Embedding 在构建向量库索引：

from llama_index import VectorStoreIndex
from llama_index import ServiceContext
from llama_index.llms import OpenAI

llm = OpenAI(model="gpt-3.5-turbo", temperature=0.1)
service_context = ServiceContext.from_defaults(
    llm=llm, embed_model="local:BAAI/bge-small-en-v1.5"
)
index = VectorStoreIndex.from_documents([document],
                                        service_context=service_context)

index 构建成功之后，就可以执行我们的各种类型的 RAG 操作了。

查询引擎

使用 index 生成查询引擎：

query_engine = index.as_query_engine()

查询

使用 query 方法查询提问：

response = query_engine.query(
    "寻找项目来积累经验时要采取哪些步骤？"
)
print(str(response))

输出可能是：

发展副业，确保该项目能够帮助您在技术上成长，与优秀的队友合作，并考虑该项目是否可以成为更大项目的垫脚石。

使用 TruLens 评估

TruLens 是一个评估和追踪 LLM 的开源程序，你可以去 Tru Lens 开源站点^[2] 了解更多内容。

评估问题集

我们现在从文件 eval_questions.txt 中读取问题构建一下评估的问题集：

eval_questions = []
with open('eval_questions.txt', 'r') as file:
    for line in file:
        # Remove newline character and convert to integer
        item = line.strip()
        print(item)
        eval_questions.append(item)

也可以手动添加一些问题：

new_question = "什么样的人工智能工作适合我？"
eval_questions.append(new_question)

打印一下所有问题：

new_question = "What is the right AI job for me?"
eval_questions.append(new_question)

['在人工智能领域建立职业生涯的关键是什么？'，
'团队合作如何有助于人工智能领域的成功？'，
'人工智能领域人际网络的重要性是什么？'，
'为了获得成功的职业生涯，需要养成哪些好习惯？'，
'利他主义如何有益于职业生涯的建立？'，
'什么是冒名顶替综合症，它与人工智能有什么关系？'，
'哪些有成就的人经历过冒名顶替综合症？'，
'擅长人工智能的第一步是什么？'，
'人工智能中常见的挑战是什么？'，
'发现人工智能的某些部分具有挑战性是正常的吗？'，
'什么样的人工智能工作适合我？']

开始评估

使用 reset_database 来重置 trulens 数据库，开始新的评估。

from trulens_eval import Tru
tru = Tru()

tru.reset_database()

TruLens 默认使用 Sqlite 数据库，DB url:  sqlite:///default.sqlite.

utils 文件里有一些工具类，来方便我们进行实验，方便大家阅读，我把文件源码放到了文章末尾。

进行评估：（评估比较耗时，由于数据量比较小大概半分钟）

from utils import get_prebuilt_trulens_recorder

tru_recorder = get_prebuilt_trulens_recorder(
    query_engine,
    app_id="Direct Query Engine")
  
with tru_recorder as recording:
    for question in eval_questions:
        response = query_engine.query(question)

records, feedback = tru.get_records_and_feedback(app_ids=[])

查看评估结果

看一下第一条记录：

records.head()

使用 Dashboard UI

和其他评估程序一样， TruLens 也提供了 Dashboard UI，启动 Dashboard:

# launches on http://127.0.0.1:8501/
tru.run_dashboard()

打开浏览器，可以看到如下界面：

我们还可以到评估界面查看所有的评估：

选择程序来查看详细信息：

包含：

• • 你的输入提问、系统的输出回答

• • 反馈详情

• • 时间线

• • App 详情等信息。

从时间线上我们可以清楚看到整个流程每一步的耗时：

高级 RAG

下面介绍两种高级 RAG 方法的使用和评估：

• • 语句窗口检索 / Sentence Window Retrieval

• • 自动合并检索 / Auto-merging retrieval

这两种方法都在召回阶段进行了优化，以达到更好的 RAG 效果。

语句窗口检索 / Sentence Window Retrieval

概念

下面展示使用语句检索窗口来进行RAG。

SWR 可以参考我RAG的文章系列，里面有更详细的介绍，这里主要讲下调用和评估。

语句窗口检索 语句窗口检索的核心思想是根据查询从自定义知识库中 选择性地获取上下文，然后利用该 上下文的更广泛版本生成更强大的文本。

实现

这次使用本地的嵌入模型BAAI/bge-small-en-v1.5 以及工具类构建索引：

from llama_index.llms import OpenAI

llm = OpenAI(model="gpt-3.5-turbo", temperature=0.1)

from utils import build_sentence_window_index

sentence_index = build_sentence_window_index(
    document,
    llm,
    embed_model="local:BAAI/bge-small-en-v1.5",
    save_dir="sentence_index"
)

生成查询引擎：

from utils import get_sentence_window_query_engine

sentence_window_engine = get_sentence_window_query_engine(sentence_index)

提问我如何开始人工智能个人项目？：

window_response = sentence_window_engine.query(
    "我如何开始人工智能个人项目？"
)
print(str(window_response))

可以看到这次的回答变得更加丰满，不再是之前的一句话回答。

要开始 AI 个人项目，您可以先确定一个与您的职业目标和兴趣相符的项目。
选择一个负责任、合乎道德且对人们有益的项目非常重要。选择项目后，
您可以按照所提供章节中概述的步骤进行操作，例如确定项目范围、
着眼于职业发展执行项目以及构建展示技能进步的作品集。
这种方法将帮助您着手一个有意义的 AI 项目，为您的职业发展做出积极贡献。

评估

重新进行评估：

tru.reset_database()

tru_recorder_sentence_window = get_prebuilt_trulens_recorder(
    sentence_window_engine,
    app_id = "Sentence Window Query Engine"
)

for question in eval_questions:
    with tru_recorder_sentence_window as recording:
        response = sentence_window_engine.query(question)
        print(question)
        print(str(response))

查看第一条数据：

tru.get_leaderboard(app_ids=[])

查看 Dashboard：

# launches on http://127.0.0.1:8501/
tru.run_dashboard()

自动合并检索 / Auto-merging retrieval

自动合并检索 通过自动合并多个检索到的段落来改进检索过程。这种方法不依赖于单个检索到的段落，而是整合来自各种来源的信息， 以产生更全面、更有凝聚力的响应。

实现自动合并检索

使用工具类函数来自动合并检索 RAG：

from utils import build_automerging_index

automerging_index = build_automerging_index(
    documents,
    llm,
    embed_model="local:BAAI/bge-small-en-v1.5",
    save_dir="merging_index"
)

构建查询引擎：

from utils import get_automerging_query_engine

automerging_query_engine = get_automerging_query_engine(
    automerging_index,
)

还是用之前的问题提问：

auto_merging_response = automerging_query_engine.query(
    "我如何开始人工智能个人项目？"
)
print(str(auto_merging_response))

这是合并两个节点回答之后的回答：

构建 AI 项目组合需要从简单的项目开始，然后随着时间的推移逐渐发展到更复杂的项目。
展示各种项目以展示您在该领域的成长和技能非常重要。
此外，有效的沟通对于向他人解释您的思维过程和工作价值至关重要，这有助于赢得信任和更大项目的资源。

评估

使用 TruLens 评估：

tru.reset_database()

tru_recorder_automerging = get_prebuilt_trulens_recorder(automerging_query_engine,
                                                         app_id="Automerging Query Engine")
for question in eval_questions:
    with tru_recorder_automerging as recording:
        response = automerging_query_engine.query(question)
        print(question)
        print(response)

同样的方式查看评估结果：

tru.get_leaderboard(app_ids=[])

# launches on http://127.0.0.1:8501/
tru.run_dashboard()

附录

工具类源码：

#!pip install python-dotenv


import os
from dotenv import load_dotenv, find_dotenv

import numpy as np
from trulens_eval import (
    Feedback,
    TruLlama,
    OpenAI
)

from trulens_eval.feedback import Groundedness
import nest_asyncio

nest_asyncio.apply()


def get_openai_api_key():
    _ = load_dotenv(find_dotenv())

    return os.getenv("OPENAI_API_KEY")


def get_hf_api_key():
    _ = load_dotenv(find_dotenv())

    return os.getenv("HUGGINGFACE_API_KEY")

openai = OpenAI()

qa_relevance = (
    Feedback(openai.relevance_with_cot_reasons, name="Answer Relevance")
    .on_input_output()
)

qs_relevance = (
    Feedback(openai.relevance_with_cot_reasons, name = "Context Relevance")
    .on_input()
    .on(TruLlama.select_source_nodes().node.text)
    .aggregate(np.mean)
)

#grounded = Groundedness(groundedness_provider=openai, summarize_provider=openai)
grounded = Groundedness(groundedness_provider=openai)

groundedness = (
    Feedback(grounded.groundedness_measure_with_cot_reasons, name="Groundedness")
        .on(TruLlama.select_source_nodes().node.text)
        .on_output()
        .aggregate(grounded.grounded_statements_aggregator)
)

feedbacks = [qa_relevance, qs_relevance, groundedness]

def get_trulens_recorder(query_engine, feedbacks, app_id):
    tru_recorder = TruLlama(
        query_engine,
        app_id=app_id,
        feedbacks=feedbacks
    )
    return tru_recorder

def get_prebuilt_trulens_recorder(query_engine, app_id):
    tru_recorder = TruLlama(
        query_engine,
        app_id=app_id,
        feedbacks=feedbacks
        )
    return tru_recorder

from llama_index import ServiceContext, VectorStoreIndex, StorageContext
from llama_index.node_parser import SentenceWindowNodeParser
from llama_index.indices.postprocessor import MetadataReplacementPostProcessor
from llama_index.indices.postprocessor import SentenceTransformerRerank
from llama_index import load_index_from_storage
import os


def build_sentence_window_index(
    document, llm, embed_model="local:BAAI/bge-small-en-v1.5", save_dir="sentence_index"
):
    # create the sentence window node parser w/ default settings
    node_parser = SentenceWindowNodeParser.from_defaults(
        window_size=3,
        window_metadata_key="window",
        original_text_metadata_key="original_text",
    )
    sentence_context = ServiceContext.from_defaults(
        llm=llm,
        embed_model=embed_model,
        node_parser=node_parser,
    )
    if not os.path.exists(save_dir):
        sentence_index = VectorStoreIndex.from_documents(
            [document], service_context=sentence_context
        )
        sentence_index.storage_context.persist(persist_dir=save_dir)
    else:
        sentence_index = load_index_from_storage(
            StorageContext.from_defaults(persist_dir=save_dir),
            service_context=sentence_context,
        )

    return sentence_index


def get_sentence_window_query_engine(
    sentence_index,
    similarity_top_k=6,
    rerank_top_n=2,
):
    # define postprocessors
    postproc = MetadataReplacementPostProcessor(target_metadata_key="window")
    rerank = SentenceTransformerRerank(
        top_n=rerank_top_n, model="BAAI/bge-reranker-base"
    )

    sentence_window_engine = sentence_index.as_query_engine(
        similarity_top_k=similarity_top_k, node_postprocessors=[postproc, rerank]
    )
    return sentence_window_engine


from llama_index.node_parser import HierarchicalNodeParser

from llama_index.node_parser import get_leaf_nodes
from llama_index import StorageContext
from llama_index.retrievers import AutoMergingRetriever
from llama_index.indices.postprocessor import SentenceTransformerRerank
from llama_index.query_engine import RetrieverQueryEngine


def build_automerging_index(
    documents,
    llm,
    embed_model="local:BAAI/bge-small-en-v1.5",
    save_dir="merging_index",
    chunk_sizes=None,
):
    chunk_sizes = chunk_sizes or [2048, 512, 128]
    node_parser = HierarchicalNodeParser.from_defaults(chunk_sizes=chunk_sizes)
    nodes = node_parser.get_nodes_from_documents(documents)
    leaf_nodes = get_leaf_nodes(nodes)
    merging_context = ServiceContext.from_defaults(
        llm=llm,
        embed_model=embed_model,
    )
    storage_context = StorageContext.from_defaults()
    storage_context.docstore.add_documents(nodes)

    if not os.path.exists(save_dir):
        automerging_index = VectorStoreIndex(
            leaf_nodes, storage_context=storage_context, service_context=merging_context
        )
        automerging_index.storage_context.persist(persist_dir=save_dir)
    else:
        automerging_index = load_index_from_storage(
            StorageContext.from_defaults(persist_dir=save_dir),
            service_context=merging_context,
        )
    return automerging_index


def get_automerging_query_engine(
    automerging_index,
    similarity_top_k=12,
    rerank_top_n=2,
):
    base_retriever = automerging_index.as_retriever(similarity_top_k=similarity_top_k)
    retriever = AutoMergingRetriever(
        base_retriever, automerging_index.storage_context, verbose=True
    )
    rerank = SentenceTransformerRerank(
        top_n=rerank_top_n, model="BAAI/bge-reranker-base"
    )
    auto_merging_engine = RetrieverQueryEngine.from_args(
        retriever, node_postprocessors=[rerank]
    )
    return auto_merging_engine

引用链接

[1] Building and Evaluating Advanced RAG Applications: https://www.deeplearning.ai/short-courses/building-evaluating-advanced-rag/
[2] Tru Lens 开源站点: https://github.com/truera/trulens

---

--- END ---