Chainlit集成LlamaIndex实现知识库高级检索（简单融合寻回器）

检索原理

** 简单融合寻回器 **
简单融合寻回原理，是利用多个检索器，融合查询最终的结果返回给LLM。此检索器还将通过生成与原始问题相关的问题，用相关问题再次检索多个检索器的数据，把原始问题和相关问题经过多个检索器检索结果整理后交给LLM最最终回复。
本次代码示例中，使用简单融合寻回器生成加上原问题在内的3个问题利用两个分割大小不一样的检索器进行检索。一个是分割成512大小块的检索器，一个是分割成128块大小的向量检索器。大块的检索器利于对针对整体（问题的很长内容）的检索，小块的检索器利于对细节（答复内容很短的检索）的检索。例如：我要检索去某个地方的路线，使用大块分割检索才能吧完整的路线。又比如，我想查询，你企业的名称是什么，需要回复的内容很对，就适合使用小块的向量检索器。使用两种检索器，集合了两者的优点，让检索结果变得更准确。

该检索技术的优缺点

在讨论 LLamaIndex 中的简单融合寻回器（Simple Fusion Retriever）的优缺点之前，我们需要明确这是一个假设性的概念，因为在 LLamaIndex 的官方文档或现有资料中，并没有直接提到“简单融合寻回器”这样一个具体术语。不过，我们可以根据融合检索的一般概念来推测其潜在的优点和缺点。

优点：

增强准确性：通过融合多种检索方式的结果，可以提高检索的准确性。例如，结合基于关键词的检索与基于语义的理解，可以在不同维度上提供更全面的信息覆盖。
改进覆盖率：不同的检索方法可能会捕捉到文档中的不同方面。融合这些方法可以确保更多的信息点被考虑到，从而提高结果的覆盖面。
灵活性：融合寻回器可以根据应用场景灵活调整不同检索方法的权重，使得系统能够适应不同的需求和环境。
减少偏差：单一的检索方法可能会有特定的局限性和偏见。融合多种方法可以在一定程度上减轻这些偏见的影响。

缺点：

复杂性增加：融合多种检索方法可能会使系统变得更加复杂，需要更多的资源来进行开发、维护和优化。
性能开销：融合检索可能需要更多的计算资源，尤其是在实时环境中，这可能会导致响应时间延长。
调参难度：为了得到最佳的检索效果，可能需要对各种参数进行微调，这需要专业知识，并且可能是一个耗时的过程。
依赖性问题：某些检索方法可能依赖于特定的数据预处理或特征提取技术，如果这些技术不合适或过时，可能会影响整个系统的性能。

请注意，上述分析是基于一般融合检索的概念而得出的，并不一定完全适用于 LLamaIndex 或任何特定实现。

LlamaIndex官方地址 https://docs.llamaindex.ai/en/stable/

快速上手

创建一个文件，例如“chainlit_chat”

mkdir chainlit_chat

进入 chainlit_chat文件夹下，执行命令创建python 虚拟环境空间(需要提前安装好python sdk。 Chainlit 需要python>=3.8。,具体操作，由于文章长度问题就不在叙述，自行百度)，命令如下：

python -m venv .venv

这一步是避免python第三方库冲突，省事版可以跳过
.venv是创建的虚拟空间文件夹可以自定义

接下来激活你创建虚拟空间，命令如下：

#linux or mac
source .venv/bin/activate
#windows
.venv\Scripts\activate

在项目根目录下创建`requirements.txt`，内容如下：

chainlit
llama-index-core
llama-index-llms-dashscope
llama-index-embeddings-dashscope

执行以下命令安装依赖：

pip install -r .\requirements.txt

安装后，项目根目录下会多出.chainlit 和.files文件夹和chainlit.md文件

代码创建

只使用通义千问的DashScope模型服务灵积的接口

在项目根目录下创建`.env`环境变量，配置如下：

DASHSCOPE_API_KEY="sk-api_key"

DASHSCOPE_API_KEY 是阿里dashscope的服务的APIkey，代码中使用DashScope的sdk实现，所以不需要配置base_url。默认就是阿里的base_url。
阿里模型接口地址 https://dashscope.console.aliyun.com/model

在项目根目录下创建app.py文件，代码如下：

import os
import time

import chainlit as cl
from llama_index.core import (
    Settings,
    VectorStoreIndex,
    SimpleDirectoryReader, StorageContext, load_index_from_storage, )
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.retrievers.fusion_retriever import FUSION_MODES
from llama_index.embeddings.dashscope import DashScopeEmbedding, DashScopeTextEmbeddingModels, \
    DashScopeTextEmbeddingType
from llama_index.llms.dashscope import DashScope, DashScopeGenerationModels

Settings.llm = DashScope(
    model_name=DashScopeGenerationModels.QWEN_MAX,max_tokens=512, api_key=os.environ["DASHSCOPE_API_KEY"]
)
Settings.embed_model = DashScopeEmbedding(
    model_name=DashScopeTextEmbeddingModels.TEXT_EMBEDDING_V2,
    text_type=DashScopeTextEmbeddingType.TEXT_TYPE_DOCUMENT,
)


@cl.cache
def get_vector_store_index():
    big_storage_dir = "./storage_big"
    small_storage_dir = "./storage_small"
    if os.path.exists(big_storage_dir) and os.path.exists(small_storage_dir):
        # rebuild storage context
        big_storage_context = StorageContext.from_defaults(persist_dir=big_storage_dir)
        # load index
        big_index = load_index_from_storage(big_storage_context)
        # rebuild storage context
        small_storage_context = StorageContext.from_defaults(persist_dir=small_storage_dir)
        # load index
        small_index = load_index_from_storage(small_storage_context)
    else:
        documents = SimpleDirectoryReader("./data_file").load_data(show_progress=True)
        print(f"documents: {len(documents)}")
        big_node_parser = SentenceSplitter.from_defaults(chunk_size=512, chunk_overlap=20)
        big_nodes = big_node_parser.get_nodes_from_documents(documents)
        print(f"big_nodes: {len(big_nodes)}")
        big_index = VectorStoreIndex(nodes=big_nodes)
        small_node_parser = SentenceSplitter.from_defaults(chunk_size=128, chunk_overlap=10)
        small_nodes = small_node_parser.get_nodes_from_documents(documents)
        print(f"small_nodes: {len(small_nodes)}")
        small_index = VectorStoreIndex(nodes=small_nodes)

        big_index.storage_context.persist(persist_dir=big_storage_dir)
        small_index.storage_context.persist(persist_dir=small_storage_dir)
    return big_index, small_index


vector_big_index, vector_small_index = get_vector_store_index()


@cl.on_chat_start
async def start():
    await cl.Message(
        author="Assistant", content="你好! 我是泰山AI智能助手. 有什么可以帮助你的吗?"
    ).send()


@cl.on_message
async def main(message: cl.Message):
    start_time = time.time()
    fusion_retriever = QueryFusionRetriever(
        mode=FUSION_MODES.RELATIVE_SCORE,
        similarity_top_k=5,
        num_queries=3,
        retrievers=[vector_big_index.as_retriever(),
                    vector_small_index.as_retriever()])

    query_engine = RetrieverQueryEngine.from_args(
        fusion_retriever, streaming=True
    )
    msg = cl.Message(content="", author="Assistant")
    res = await query_engine.aquery(message.content)
    async for token in res.response_gen:
        await msg.stream_token(token)
    print(f"代码执行时间: {time.time() - start_time} 秒")
    source_names = []
    for idx, node_with_score in enumerate(res.source_nodes):
        node = node_with_score.node
        source_name = f"source_{idx}"
        source_names.append(source_name)
        msg.elements.append(
            cl.Text(content=node.get_text(), name=source_name, display="side")
        )
    await msg.stream_token(f"\n\n **数据来源**: {', '.join(source_names)}")
    await msg.send()

代码中的persist_dir=storage_dir 不设置的默认是 ./storage.
代码中的 big_node_parser = SentenceSplitter.from_defaults(chunk_size=512, chunk_overlap=20) ,chunk_size是将长文档分割的文本块的大小，chunk_overlap 是和上下文本块的重合文本的大小。
similarity_top_k=5 返回5条最相关的数据
num_queries=3 根据原问题生成两个问题加上原问题的3个问题进行检索
retrievers=[vector_big_index.as_retriever(), vector_small_index.as_retriever()]) 使用的检索器结合
代码中3个问题2个检索器，将进行3*2=6次检索。

代码解读

这段代码使用了chainlit和llama_index两个Python库来创建一个基于文档的问答系统。下面是对代码段的解释：

导入必要的模块：
- os 和 time 是Python标准库的一部分，分别用于操作系统相关的功能和计时。
- chainlit 是一个用于快速构建交互式AI应用的库。
- llama_index 是一个框架，用于构建索引并进行文档检索。
配置llama_index的核心设置：
- 设置了使用的LLM（大语言模型）为DashScope的Qwen Turbo版本，并通过环境变量获取API密钥。
- 设置了嵌入模型（Embedding Model）为DashScope的文本嵌入模型，并指定了模型类型。
- 使用SentenceSplitter来分割文本节点，定义了块大小和重叠。
- 定义了输出长度和上下文窗口大小。
缓存函数get_vector_store_index()：
- 这个函数负责加载或创建一个向量存储索引。如果存储目录存在，则从该目录加载已有的索引；否则，从指定的数据文件夹读取文档并创建新的索引。
使用chainlit装饰器定义事件处理函数：
- @cl.on_chat_start 在聊天开始时发送欢迎消息。
- @cl.on_message 在接收到用户消息时触发，使用向量索引来查询相关性最高的文档，并将结果流式传输给用户。同时，显示每个答案片段的来源。
主逻辑部分：
- 创建一个流式查询引擎，设置相似度搜索的前k个结果。
- 当接收到消息时，使用查询引擎异步查询并流式传输响应到用户。
- 计算执行时间，并记录下每个源文档的名字以便后续引用。
- 将每个源文档的内容作为元素附加到消息中，并在最后告知用户数据来源。

这个程序提供了一个基于向量存储索引的问答系统的基本框架，可以用于从大量的文档中提取信息以回答用户的问题。

postprocessor组件

要实现最终的检索我们还需要创建query engine组件，但是在query engine组件中需要设置一个postprocessor组件作为其参数，而postprocessor组件可以由若干个子组件组合在一起，下面我们首先来简单介绍一下postprocessor子组件：Replacement组件，该组件的作用是用来选择(由target_metadata_key参数确定)将哪些context发送给llm, 也就是说Replacement组件会从检索到的context中挑选指定的内容发送给llm，所以它具有选择context的功能.

另外postprocessor还有一个叫rerank的子组件,它的作用是对检索到的上下文进行从新排序，从而得到一个精度更高的检索结果，最后Replacement组件会将rerank组件的排序结果发送给llm, 不过这里需要说明一下的是rerank是可选组件，它不是必须的，rerank组件的作用仅仅是为了提高检索的精度。

#创建Replacement组件
postproc = MetadataReplacementPostProcessor(
    target_metadata_key="window"
)
 
#创建rerank组件
# 参考: https://huggingface.co/BAAI/bge-reranker-base
rerank = SentenceTransformerRerank(
    top_n=2, 
    model="BAAI/bge-reranker-base"
)

#创建查询引擎
sentence_window_engine = sentence_index.as_query_engine(
    similarity_top_k=6, 
    node_postprocessors=[postproc, rerank]
)

代码中的BAAI/bge-reranker-base需要运行的在本地的重排模型，这个可选的。

这里创建的Replacement组件中我们设置了target_metadata_key参数为"window", 它的作用是当执行检索操作时会将context中的元数据的“窗口”数据发送给llm。而rerank组件中的top_n=2的作用是对检索到的多个context进行重新排序并选取精度最高前2个context。这里所谓的精度是指相似度计算的精度，所以可以认为经过rerank模型的重新排序后会得到和question相关度更高的context。