RAG知識庫之多表示索引

      在樸素RAG中通常會對文檔、文本進行分塊后進行文檔嵌入，對所有文件、文本都沒有經過采用Chunk方法可能有時候效果不是和好，盡管有著各種分塊策略有針對大文件的、針對小文件的策略，但都難免可能會造成上下文語義丟失。
      分塊通常有兩個非常重要的參數chunk_size、chunk_overlap，分別代表塊大小與塊與塊之間的重疊量，并不好確定這兩個值的具體數字，只能通過不斷實驗確定該值。

基本原理

      在分塊效果不好時或許可以試試多表示索引（Multi-representation indexing），在多表示索引中并不會對整個文檔分塊后進行文檔嵌入。而是通過為每個文檔都生成一個文檔摘要，為每個文檔摘要與文檔生成一個唯一ID，將摘要與ID關聯嵌入到Vectorstore中，將ID與文檔關聯存儲到獨立文檔存儲中。

      在用戶提問時會先根據提出的問題先在多向量檢索器中檢索相似度最高的文檔摘要，獲取的文檔摘要后也得到了所關聯的id，再拿文檔id到文檔存儲中獲取所對應的完整文檔。
      此文檔作為用戶提問問題的上下文，同時將問題與文檔上下文提交到LLM。

      經過多表示索引的使用嵌入空間大小下降很多，相似度搜索性能也會提升，使用完整文檔用作上下文LLM響應的準確性也會有所提高。

      適用于小文檔結構，當文檔太大時會超出LLM上下文長度，超大文檔并不適用多表示索引。

代碼示例

def loadDocs():
    loader = WebBaseLoader("https://www.hinews.cn/news/system/2024/07/04/033173743.shtml")
    docs = loader.load()
    loader = WebBaseLoader("https://www.hinews.cn/news/system/2024/07/04/033173745.shtml")
    docs.extend(loader.load())
    return docs
	
def processSummaries():
    chain = (
    {"doc": lambda x: x.page_content}
    | ChatPromptTemplate.from_template("請總結以下文檔:\n\n{doc}")
    | llm
    | StrOutputParser())
    docs=loadDocs()
    summaries = chain.batch(docs, {"max_concurrency": 5})
    return summaries,docs

def get_retriever():
    summaries, docs= processSummaries()
    docstore = init_docstore() 
    vectorstore = Chroma(collection_name="summaries", embedding_function=initEmbedding())
    #創建用于映射概要與文檔的id數組
    doc_ids = [str(uuid.uuid4()) for _ in docs]
    #創建概要文檔并關聯ID 
    summary_docs = [Document(page_content=s, metadata={"doc_id": doc_id}) for s, doc_id in zip(summaries, doc_ids)]
    #創建多向量檢索器
    retriever = MultiVectorRetriever(
        vectorstore=vectorstore,
        byte_store=docstore,
        id_key="doc_id",
        search_kwargs={'k': 1}
    )
    #將概要文檔添加到向量存儲
    retriever.vectorstore.add_documents(summary_docs)
    #文檔與id關聯
    lists = list(zip(doc_ids, docs))

    #文檔添加到檢索器的文檔存儲中
    retriever.docstore.mset(lists)

    return retriever

      創建好多向量檢索器后即可使用該檢索器，或配合RetrievalQA使用：

query="機場在哪里"
retriever= get_retriever()
retriever.invoke(query)

#問答鏈
qa_chain = RetrievalQA.from_chain_type(llm, retriever=retriever, chain_type_kwargs={"prompt": PROMPT})
resp =qa_chain.invoke(query)

      從多向量檢索器的實現源碼可看到，其先從vectorstore中查詢到相似度最高的N個概要文檔，然后獲取概要文檔ID，根據ID去查詢完整文檔。

sub_docs = await self.vectorstore.asimilarity_search(
                query, **self.search_kwargs
            )
ids = []
for d in sub_docs:
    if self.id_key in d.metadata and d.metadata[self.id_key] not in ids:
        ids.append(d.metadata[self.id_key])
docs = await self.docstore.amget(ids)