Chainlit集成LlamaIndex实现知识库高级检索（自动合并检索）

检索原理

自动合并检索
自动合并检索原理，和我的上一篇文章的检索方案：将文本分割成512大小（一般对应段落大小）和128（一般对句子大小不是严格的句子长度）大小两种分别存储到索引库，再用llama_index的简单融合寻回器，分别从这里个向量库查询。将查询结果融合排序后交给LLM的方式十分类似，不同点是检索能将子小块的内容合成大块文本返回。上一篇文章《Chainlit集成LlamaIndex实现知识库高级检索（简单融合寻回器）》
自动合并检索主要是将文档按照块大小拆分成不同层级的节点，这些节点包括父节点和子节点，然后在检索过程中找到相似度高的叶子节点，如果一个父节点中有多个子节点被检索到，那么这个父节点就会被自动合并，最终将父节点的所有文档都作为上下文发送给 LLM（大语言模型），下面是自动合并检索的示意图：

在这里插入图片描述

该检索技术的优缺点

LlamaIndex是一个用于将大型语言模型（LLMs）与外部数据连接的工具，它提供了一系列的功能，使得从外部数据源检索信息变得更加容易和高效。其中，自动合并检索（Auto-merging Retrieval）是LlamaIndex的一项重要功能，它能够在检索过程中自动合并相关的小文本片段，形成更大的上下文，以便更好地服务于后续的信息合成或问题回答任务。以下是关于LlamaIndex自动合并检索的一些优缺点分析：

优点

提高检索质量
自动合并检索能够递归地“合并”引用父节点超过给定阈值的叶节点子集，从而将潜在不同的、较小的上下文合并成一个较大的上下文。这意味着，当用户提出一个问题或者需要获取相关信息时，系统不仅仅提供孤立的信息片段，而是将相关的信息整合起来，形成更加完整和连贯的答案，提高了检索结果的相关性和质量。

优化文本合成
自动合并检索不仅有助于提高检索的准确性，还可以优化文本的合成过程。通过合并相关的上下文，系统可以更好地理解用户的需求，并且在合成信息时考虑到更多的背景信息，使得最终生成的文本更加符合用户的预期。

灵活性和高效性
LlamaIndex作为一个文本检索工具，以其灵活性和高效性著称。自动合并检索作为其一部分，同样继承了这些特点，使得用户可以在不同的应用场景下灵活地调整检索参数，以达到最佳的检索效果。

缺点

可能增加计算成本
虽然自动合并检索提高了检索质量，但是由于需要合并多个相关片段，可能会导致计算资源的消耗增加。特别是在处理大量数据的情况下，如果不对合并的阈值和策略进行合理的设定，可能会导致不必要的计算开销。

复杂性提升
相比于简单的检索方式，自动合并检索增加了系统的复杂性。为了实现高效的合并，需要设计合理的层次结构和切割策略，这对于用户来说意味着更高的学习成本和技术门槛。同时，这也要求系统设计者需要对数据结构有深入的理解。

潜在的信息冗余
尽管自动合并检索旨在提供更完整的上下文，但是在某些情况下，这种合并可能会引入不必要的信息冗余。如果合并策略不够精确，可能会导致检索结果中包含过多无关紧要的细节，反而影响了信息的清晰度和可读性。

综上所述，LlamaIndex的自动合并检索功能为提高检索质量和文本合成提供了强大的支持，但在实际应用中也需要考虑其可能带来的额外开销和复杂性。正确地配置和使用这一功能，对于最大化其优势同时减轻潜在的缺点至关重要。总体来看，只要文档切割分块的每个块的内容完整，使用自动合并检索的这种方式优于我上一篇写的文章《Chainlit集成LlamaIndex实现知识库高级检索（简单融合寻回器）》的方案，当然你也可以使用简单融合寻回器和自动合并检索结合，利用QueryFusionRetriever 的num_queries参数来生成多个相似问题，使用retrievers参数设置，自动合并检索器和其他检索器整合，提高检索精度。

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, HierarchicalNodeParser, get_leaf_nodes, get_root_nodes, \
    get_child_nodes
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.storage.docstore import SimpleDocumentStore
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():
    storage_dir = "./storage_auto"
    if os.path.exists(storage_dir):
        # rebuild storage context
        storage_context = StorageContext.from_defaults(persist_dir=storage_dir)
        # load index
        vector_store_index = load_index_from_storage(storage_context)
    else:
        documents = SimpleDirectoryReader("./data_file").load_data(show_progress=True)
        print(f"documents: {len(documents)}")
        chunk_sizes = [512, 128]
        node_parser_ids = [f"chunk_size_{chunk_size}" for chunk_size in chunk_sizes]
        node_parser_map = {}
        for chunk_size, node_parser_id in zip(chunk_sizes, node_parser_ids):
            if chunk_size == 128:
                chunk_overlap = 10
            else:
                chunk_overlap = 20
            print(chunk_size, chunk_overlap)
            node_parser_map[node_parser_id] = SentenceSplitter(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
            )
        node_parser = HierarchicalNodeParser.from_defaults(
            node_parser_ids=node_parser_ids, node_parser_map=node_parser_map
        )
        nodes = node_parser.get_nodes_from_documents(documents)
        print(f"nodes: {len(nodes)}")
        root_nodes = get_root_nodes(nodes)
        print(f"root_nodes: {len(root_nodes)}")
        middle_nodes = get_child_nodes(root_nodes, all_nodes=nodes)
        print(f"middle_nodes: {len(middle_nodes)}")
        leaf_nodes = get_leaf_nodes(nodes)
        print(f"leaf_nodes: {len(leaf_nodes)}")
        doc_store = SimpleDocumentStore()
        doc_store.add_documents(nodes)
        storage_context = StorageContext.from_defaults(docstore=doc_store)
        vector_store_index = VectorStoreIndex(
            nodes=leaf_nodes, storage_context=storage_context
        )
        vector_store_index.storage_context.persist(persist_dir=storage_dir)
    return vector_store_index


vector_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()
    vector_retriever = vector_index.as_retriever(similarity_top_k=20)
    retriever = AutoMergingRetriever(
        vector_retriever=vector_retriever, storage_context=vector_index.storage_context,simple_ratio_thresh=0.4,verbose=True
    )
    query_engine = RetrieverQueryEngine.from_args(
        retriever, streaming=True,
    )
    print(f"代码执行时间1: {time.time() - start_time} 秒")
    msg = cl.Message(content="", author="Assistant")
    res = await query_engine.aquery(message.content)
    print(f"代码执行时间1: {time.time() - start_time} 秒")
    async for token in res.response_gen:
        await msg.stream_token(token)
    print(f"代码执行时间3: {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.
代码中chunk_size是将长文档分割的文本块的大小，chunk_overlap 是和上下文本块的重合文本的大小。
代码中 node_parser = HierarchicalNodeParser.from_defaults( node_parser_ids=node_parser_ids, node_parser_map=node_parser_map ) 可以简写为 node_parser = HierarchicalNodeParser.from_defaults() 会按照 [2048,512,128]三种层次分割，经过我测试不使用默认的效果会更好
similarity_top_k=20 返回20条最相关的数据
simple_ratio_thresh，它的默认值是 0.5，表示自动合并文档的阀值，如果在一个父节点中，子节点被检索到的比例小于这个阀值，那么自动合并功能将不会生效，这样提交给 LLM 的上下文就只会包含检索到的叶子节点。反之如果大于这个阀值，文档就会自动合并，最终提交给 LLM 的上下文就会包含这个父节点的内容。比如父节点有 4 个子节点，检索时发现只有 1 个子节点，那么子节点被检索到的比例就是 0.25（1/4），小于阀值 0.5，所以自动合并功能不会生效，最终提交给 LLM 的上下文就只会包含那个检索到的子节点。

代码解读

这段代码展示了一个使用 chainlit 和 llama_index 库来创建一个基于向量存储索引的问答系统的过程。下面是对这段代码的关键部分进行的解读：

导入必要的库：
- 导入了 os 和 time 这两个Python标准库。
- 导入了来自 llama_index 库的核心组件，如设置、向量存储索引、文档读取器等。
- 导入了 chainlit 库用于构建交互式聊天应用。
初始化LLM（Large Language Model）和嵌入模型：
- 设置了使用的LLM为 DashScope 提供的 Qwen Max 模型，并配置了API密钥等参数。
- 嵌入模型也选择了 DashScope 的 TEXT_EMBEDDING_V2 模型。
定义获取向量存储索引的方法：
- 如果存储目录存在，则加载已有的索引；否则，从文档中创建新的索引。
- 使用 HierarchicalNodeParser 来处理文档，将其拆分为不同粒度的节点，并构建层次结构。
- 创建 SimpleDocumentStore 并将所有节点存储进去。
- 创建并保存向量存储索引。
定义聊天启动函数：
- 使用 chainlit 的 on_chat_start 装饰器来定义当聊天开始时发送的消息。
定义消息处理函数：
- 使用 chainlit 的 on_message 装饰器来定义接收用户输入后执行的操作。
- 创建一个向量检索器，并基于此创建一个自动合并检索器。
- 创建一个查询引擎来处理检索到的信息，并通过流式传输的方式返回结果。
- 处理查询引擎返回的结果，并通过 chainlit 发送回给用户。

这段代码主要展示了如何构建一个基于文档知识库的问答系统，并且利用 chainlit 来提供用户界面进行交互。它包括了从文档加载到索引构建，再到查询处理和结果展示的整个流程。需要注意的是，为了使这段代码运行，你需要确保安装了所有必要的依赖库，并且拥有正确的API密钥。此外，代码中的路径（例如 ./data_file 和 ./storage_auto）需要根据实际情况调整。