浅析RAG系统文档块结构化索引构建 - 文章 - 开发者社区

“ 一篇旧文分享，也是RAG系统的进阶技巧把；上一篇文章咱们聊过“忘记RAG，拥抱大模型”，是从一个大的宏观层面来讲系统设计，但是不管是naive RAG，还是adavance RAG，或者是最新研究中被称为modular RAG的系统。只要是涉及到大规模知识库的问答场景，召回都是很重要的一个环节，而召回的第一步是文档的解析，文档的结构化。所以陆续分享一下之前写的调研文章，本文为文档块结构化索引的构建种类。

列表索引
关键词索引
树索引
组合索引
pandas/sql 索引
文档摘要索引
知识图谱索引

文档索引是想解决什么事情呢？一个naive的RAG系统的基本流程：

将源文件分成文本块。
将文本块存储在向量数据库中。
在问答时，利用相似度和/或关键词过滤来检索文本块。
使用大模型生成

这个流程的弊端不限于以下几点：

文本块没有完整的全局上下文，这严重限制了问答过程。
需要研究调整 top-k / 相似度分数阈值，太小的值可能会导致相关上下文被忽略，而太大的值则可能会增加与无关上下文的成本和系统延迟。
向量召回结果可能并不总是最合适上下文，因为这个过程本质上是分别确定文本和上下文。

不管用哪种方式读取文本，哪种方式切分文本之后，都可以将非结构化的文档分解成一个一个的文本块，因为文本块之间可能有一些关联关系，所以这里定义文本块为Node(节点)。节点可以是文本块、图标或者别的内容，每个节点还回保存各种元信息、关键信息。

列表索引 ：假如我们对一个文档按照纯文本读取，然后按照固定长度切分，我们可以得到最简单的索引结构，一般称为list index，如下图所示：

picture.image

在搜索过程中，只需要根据一定的检索规则，例如keyword过滤，得到过滤后的节点合并到生成模块。当然这里最常见的还是使用embedding来召回，这种情况需要在节点中会包含向量特征。

picture.image

树索引 ：当有大量的文档，或者可以明确提取到文档的大纲、标题信息的时候，可以采取树索引，树索引是一种树形结构的索引，其中每个父节点都是子节点的总结。在索引构建过程中，树是自下而上地构建的，直到我们最终得到一组根节点。 picture.image

对于树索引的检索，涉及从根节点向下遍历到叶节点。默认情况下（child_branch_factor=1），查询在给定父节点的情况下选择一个子节点。如果child_branch_factor=2，则查询在每层选择两个子节点。

picture.image 在llama index中这种索引的简单代码如下


        
          
from llama_index import GPTTreeIndex  
  
new_index = GPTTreeIndex.from_documents(documents)  
response = query_engine.query("hello world")  
display(Markdown(f"<b>{response}</b>"))

关键词表索引，这对于将用户query路由到不同的数据源非常有用。有点类似es的倒排索引的概念，对每个节点提取关键词，共享相同的关键词的节点会联系到一起。

在搜索时，从用户query中提取相关关键词，并将其与预先提取的节点关键词进行匹配，以获取相应的节点。提取的节点将传递到生成模块。

picture.image

图索引 - 复合索引

例如：可以为每个文档内的文本创建树形索引；然后生成一个列表索引，涵盖整个文档集合的所有树形索引。

一个简单的代码例子如下，涉及如下步骤

从多个文档创建树索引
从树形索引生成摘要（树形索引对于总结文档集合非常有用）
在树形索引之上创建一个列表索引的图（列表索引适用于综合多个数据源的信息来合成答案）


        
          
读取企业财报  
  
years = ['Q2-2023', 'Q3-2023']  
UnstructuredReader = download_loader('UnstructuredReader', refresh_cache=True)  
  
loader = UnstructuredReader()  
doc_set = {}  
all_docs = []  
  
for year in years:  
    year_docs = loader.load_data(f'./data\_{year}.pdf', split_documents=False)  
    for d in year_docs:  
        d.extra_info = {"quarter": year.split("-")[0],   
                        "year": year.split("-")[1],  
                        "q":year.split("-")[0]}  
    doc_set[year] = year_docs  
    all_docs.extend(year_docs)  
  
建立向量索引  
service_context = ServiceContext.from_defaults(chunk_size_limit=512)  
index_set = {}  
for year in years:  
    storage_context = StorageContext.from_defaults()  
    cur_index = GPTVectorStoreIndex.from_documents(  
        documents=doc_set[year],  
        service_context=service_context,  
        storage_context=storage_context  
    )  
    index_set[year] = cur_index  
    storage_context.persist(f'./storage\_index/{year}')  
      
生成树索引的总结  
index_summary = [index_set[year].as_query_engine().query("Summary this document in 100 words").response for year in years]  
  
构图+搜索  
from llama_index.indices.composability import ComposableGraph  
  
from langchain.chat_models import ChatOpenAI  
from llama_index import LLMPredictor  
  
llm_predictor = LLMPredictor(llm=ChatOpenAI(temperature=0, max_tokens=512, model_name='gpt-3.5-turbo'))  
service_context = ServiceContext.from_defaults(llm_predictor=llm_predictor)  
storage_context = StorageContext.from_defaults()  
  
  
graph = ComposableGraph.from_indices(  
    GPTListIndex,  
    [index_set[y] for y in years],  
    index_summaries=index_summary,  
    service_context=service_context,  
    storage_context=storage_context  
)  
root_id = graph.root_id  
  
storage_context.persist(f'./storage\_index/root')  
  
custom_query_engines = {  
    index_set[year].index_id: index_set[year].as_query_engine() for year in years  
}  
  
query_engine = graph.as_query_engine(  
    custom_query_engines=custom_query_engines  
)  
  
response = query_engine.query("hello")  
response.response

pandas index & sql index ：用于结构化数据

pandas index


        
          
from llama_index.indices.struct_store import GPTPandasIndex  
import pandas as pd  
  
df = pd.read_csv("titanic\_train.csv")  
  
index = GPTPandasIndex(df=df)  
  
query_engine = index.as_query_engine(  
    verbose=True  
)  
response = query_engine.query(  
    "What is the correlation between survival and age?",  
)  
response

picture.image

sql index


        
          
from llama_index import SimpleDirectoryReader, WikipediaReader  
from sqlalchemy import create_engine, MetaData, Table, Column, String, Integer, select, column  
  
wiki_docs = WikipediaReader().load_data(pages=['Toronto', 'Berlin', 'Tokyo'])  
  
engine = create_engine("sqlite:///:memory:")  
metadata_obj = MetaData()  
  
table_name = "city\_stats"  
city_stats_table = Table(  
    table_name,  
    metadata_obj,  
    Column("city\_name", String(16), primary_key=True),  
    Column("population", Integer),  
    Column("country", String(16), nullable=False),  
)  
metadata_obj.create_all(engine)  
  
from llama_index import GPTSQLStructStoreIndex, SQLDatabase, ServiceContext  
from langchain import OpenAI  
from llama_index import LLMPredictor  
  
llm_predictor = LLMPredictor(llm=LLMPredictor(llm=ChatOpenAI(temperature=0, max_tokens=512, model_name='gpt-3.5-turbo')))  
service_context = ServiceContext.from_defaults(llm_predictor=llm_predictor)  
  
sql_database = SQLDatabase(engine, include_tables=["city\_stats"])  
sql_database.table_info  
  
index = GPTSQLStructStoreIndex.from_documents(  
    wiki_docs,   
    sql_database=sql_database,   
    table_name="city\_stats",  
    service_context=service_context  
)  
  
stmt = select(  
    city_stats_table.c["city\_name", "population", "country"]  
).select_from(city_stats_table)  
  
with engine.connect() as connection:  
    results = connection.execute(stmt).fetchall()  
    print(results)  
  
query_engine = index.as_query_engine(  
    query_mode="nl"  
)  
response = query_engine.query("Which city has the highest population?")

picture.image

文档摘要索引 ，通过提取每个文档的非结构化文本摘要，从而提高检索性能。该索引包含的信息比单个文本块更多，比关键词标签具有更多的语义含义。它还允许灵活的检索，包括LLM和基于向量的方法。

在构建时，文档摘要索引使用LLM从每个文档中提取摘要。在问答时，它根据以下方法使用摘要检索相关文档：

基于LLM的检索：获取文档摘要集合，并请求LLM识别与用户query相关文档+相关性评分
基于向量的检索：利用摘要向量的相似性来检索相关文档，取top k的结果

代码示例如下：


        
          
import nest_asyncio  
nest_asyncio.apply()  
  
from llama_index import (  
    SimpleDirectoryReader,  
    LLMPredictor,  
    ServiceContext,  
    ResponseSynthesizer  
)  
from llama_index.indices.document_summary import GPTDocumentSummaryIndex  
from langchain.chat_models import ChatOpenAI  
  
wiki_titles = ["Toronto", "Seattle", "Chicago", "Boston", "Houston"]  
  
from pathlib import Path  
  
import requests  
for title in wiki_titles:  
    response = requests.get(  
        'https://en.wikipedia.org/w/api.php',  
        params={  
            'action': 'query',  
            'format': 'json',  
            'titles': title,  
            'prop': 'extracts',  
            'explaintext': True,  
        }  
    ).json()  
    page = next(iter(response['query']['pages'].values()))  
    wiki_text = page['extract']  
  
    data_path = Path('data')  
    if not data_path.exists():  
        Path.mkdir(data_path)  
  
    with open(data_path / f"{title}.txt", 'w') as fp:  
        fp.write(wiki_text)  
  
city_docs = []  
for wiki_title in wiki_titles:  
    docs = SimpleDirectoryReader(input_files=[f"data/{wiki\_title}.txt"]).load_data()  
    docs[0].doc_id = wiki_title  
    city_docs.extend(docs)  
  
llm_predictor_chatgpt = LLMPredictor(llm=ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo"))  
service_context = ServiceContext.from_defaults(llm_predictor=llm_predictor_chatgpt, chunk_size_limit=1024)  
  
response_synthesizer = ResponseSynthesizer.from_args(response_mode="tree\_summarize", use_async=True)  
doc_summary_index = GPTDocumentSummaryIndex.from_documents(  
    city_docs,   
    service_context=service_context,  
    response_synthesizer=response_synthesizer  
)  
  
doc_summary_index.get_document_summary("Boston")

知识图谱索引 ：通过从文档中提取知识三元组（主语，谓语，宾语）来构建索引。一个可参考的英文模型如下：


        
          
Babelscape/rebel-large

在查询时，常见的为2种，一种是类似于pandas、sql index生成查询语句。或者利用节点的文本信息利用其他的检索方式搜索相关的信息。

例子如下：


        
          
from llama_index import (  
    SimpleDirectoryReader,  
    ServiceContext,  
    KnowledgeGraphIndex,  
)  
from llama_index.graph_stores import SimpleGraphStore  
  
from llama_index.llms import OpenAI  
from IPython.display import Markdown, display  
documents = SimpleDirectoryReader(  
    "../../../../examples/paul\_graham\_essay/data"  
).load_data()  
  
llm = OpenAI(temperature=0, model="text-davinci-002")  
service_context = ServiceContext.from_defaults(llm=llm, chunk_size=512)  
  
from llama_index.storage.storage_context import StorageContext  
  
graph_store = SimpleGraphStore()  
storage_context = StorageContext.from_defaults(graph_store=graph_store)  
  
index = KnowledgeGraphIndex.from_documents(  
    documents,  
    max_triplets_per_chunk=2,  
    storage_context=storage_context,  
    service_context=service_context,  
)  
  
query_engine = index.as_query_engine(  
    include_text=False, response_mode="tree\_summarize"  
)  
response = query_engine.query(  
    "Tell me more about Interleaf",  
)

picture.image

总结：

向量索引用起来很简单，可以很容易实现在大规模的知识库下的检索
列表索引，适用于组合多个数据源的信息来生成答案
树索引，适用于文档的结构化
关键词表索引，适用于对用户问题路由到不同的数据源
组合索引，适用于多种索引的组合
pandas、sql索引，适用于结构化数据
文档摘要索引，适用于文档的检索