在本文中，我們將介紹一種提高RAG（Retrieval-Augmented Generation）模型檢索效果的高階技巧，即窗口上下文檢索。我們將首先回顧一下基礎(chǔ)RAG的檢索流程和存在的問(wèn)題，然后介紹窗口上下文檢索的原理和實(shí)現(xiàn)方法，最后通過(guò)一個(gè)實(shí)例展示其效果。

圖片 基礎(chǔ)RAG存在的問(wèn)題及解決方案 基礎(chǔ)RAG檢索流程 RAG是一種結(jié)合了檢索和生成的AI應(yīng)用落地的方案，它可以根據(jù)給定的問(wèn)題生成回答，同時(shí)利用外部知識(shí)庫(kù)（例如維基百科）來(lái)增強(qiáng)生成的質(zhì)量和多樣性。RAG的核心思想是將問(wèn)題和知識(shí)庫(kù)中的文檔進(jìn)行匹配，然后將匹配到的文檔作為生成模型的輸入，從而生成更加相關(guān)和豐富的回答。

圖片 RAG的檢索流程可以分為以下幾個(gè)步驟：

load：加載文檔，將各種格式的文件加載后轉(zhuǎn)化為文檔，例如將pdf加載為文本數(shù)據(jù)，或者將表格轉(zhuǎn)換為多個(gè)鍵值對(duì)。 split：將文檔拆分為適合向量存儲(chǔ)的較小單元，以便于與向量存儲(chǔ)，以及檢索時(shí)的文檔匹配，例如將“我是kxc。我喜歡唱跳，rap，和籃球?！辈鸱譃椤拔沂莐xc?！焙汀拔蚁矚g唱跳，rap，和籃?！眱蓚€(gè)數(shù)據(jù)分塊（一般稱之為chunk）。 embedding：將文檔用向量表示，例如使用BERT或TF-IDF等模型進(jìn)行向量化。 store: 將向量化后的數(shù)據(jù)分塊，存入向量數(shù)據(jù)庫(kù)。 retrive：根據(jù)問(wèn)題和文檔的向量，計(jì)算它們之間的相似度，然后根據(jù)相似度的高低，選擇最相關(guān)的文檔作為檢索結(jié)果，例如使用余弦相似度或點(diǎn)積等度量進(jìn)行排序。 query：將檢索到的文檔作為生成模型的輸入，根據(jù)問(wèn)題生成回答，例如使用GPT-3或T5等模型進(jìn)行生成。 基礎(chǔ)RAG存在的問(wèn)題 圖片 基礎(chǔ)RAG的檢索流程雖然簡(jiǎn)單，但是也存在一些問(wèn)題，主要是在split和retrive兩個(gè)步驟中。這些問(wèn)題會(huì)影響RAG的檢索效果，從而導(dǎo)致生成的回答不準(zhǔn)確或不完整。

split拆分的塊太大，在retrive時(shí)，同一塊中非相關(guān)的內(nèi)容就越多，對(duì)問(wèn)題的檢索匹配度影響越大，會(huì)導(dǎo)致檢索的不準(zhǔn)確。例如，如果我們將維基百科中的一篇文章作為一個(gè)文檔，那么這個(gè)文檔可能包含很多不同的主題和細(xì)節(jié)，與問(wèn)題的相關(guān)性會(huì)很低。如果我們將這個(gè)文檔作為檢索結(jié)果，那么生成模型可能會(huì)從中提取出一些無(wú)關(guān)或錯(cuò)誤的信息，從而影響回答的質(zhì)量。

split拆分的塊太小，檢索的匹配度會(huì)提高，然而在最后的query環(huán)節(jié)，提供給llm使用的信息會(huì)由于缺少上下文的信息支撐，導(dǎo)致回答不準(zhǔn)確。例如，如果我們將維基百科中的一篇文章拆分為多個(gè)句子，那么每個(gè)句子可能只包含一小部分的信息，與問(wèn)題的相關(guān)性會(huì)很高。如果我們將這些句子作為檢索結(jié)果，那么生成模型可能會(huì)從中提取出一些有用的信息，但是也可能會(huì)忽略一些重要的上下文信息，從而影響回答的完整性。

解決方案-窗口上下文檢索 圖片 解決這個(gè)問(wèn)題一般采取的方案是，在split拆分時(shí)，盡量將文本切分到最小的語(yǔ)義單元。retrive時(shí)，不直接使用匹配到doc，而是通過(guò)匹配到的doc拓展其上下文內(nèi)容后整合到一起，在投遞到LLM使用。這樣，既可以提高檢索的精度，又可以保留上下文的完整性，從而提高生成的質(zhì)量和多樣性。

具體來(lái)說(shuō)，這種方案的實(shí)現(xiàn)步驟如下：

在split拆分時(shí)，將文本切分為最小的語(yǔ)義單元，例如句子或段落，并給每個(gè)單元分配一個(gè)唯一的編號(hào)，作為其在文本中的位置信息。 在retrive檢索時(shí)，根據(jù)問(wèn)題和文檔的向量，計(jì)算它們之間的相似度，然后選擇最相關(guān)的文檔作為檢索結(jié)果，同時(shí)記錄下它們的編號(hào)。 在query生成時(shí)，根據(jù)檢索結(jié)果的編號(hào)，從文本中獲取它們的上下文信息，例如前后若干個(gè)單元，然后將它們拼接成一個(gè)完整的文檔，作為生成模型的輸入，根據(jù)問(wèn)題生成回答。 窗口上下文檢索實(shí)踐 上下文檢索實(shí)現(xiàn)思路 我們從最終要實(shí)現(xiàn)的目標(biāo)著手，也就是在retrive時(shí)要能通過(guò)匹配到doc拓展出與這個(gè)doc內(nèi)容相關(guān)的上下文。要想實(shí)現(xiàn)這個(gè)目標(biāo)，我們就必須建立每個(gè)doc與其上下文的關(guān)聯(lián)關(guān)系。這個(gè)關(guān)系的建立其實(shí)十分簡(jiǎn)單，只需要按順序給拆分出來(lái)的每個(gè)doc進(jìn)行編號(hào)，在檢索時(shí)通過(guò)當(dāng)前文檔的編號(hào)就能匹配到相關(guān)上下文doc的編號(hào)，進(jìn)而獲取上下文的內(nèi)容。

基于chroma向量庫(kù)的代碼實(shí)踐 想要實(shí)踐以上思路，需要在split環(huán)節(jié)基于文檔順序，將文檔編碼寫(xiě)入元數(shù)據(jù)。在檢索時(shí)，則通過(guò)元數(shù)據(jù)中的順序分塊編碼來(lái)查找上下文。具體的代碼如下：

1.split時(shí)對(duì)分塊編碼并寫(xiě)入元數(shù)據(jù) import bs4,uuid from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.document_loaders import WebBaseLoader from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings

Load, chunk and index the contents of the blog.

loader = WebBaseLoader( web_paths=("

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) docs = text_splitter.split_documents(doc)

這里給每個(gè)docs片段的metadata里注入file_id

file_id = uuid.uuid4().hex chunk_id_counter = 0 for doc in docs: doc.metadata["file_id"] = file_id doc.metadata["chunk_id"] = f'{file_id}_{chunk_id_counter}' # 添加chunk_id到metadata chunk_id_counter += 1 for key,value in doc.metadata.items(): if not isinstance(value, (str, int, float, bool)): doc.metadata[key] = str(value)

vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())

2.retrive時(shí)通過(guò)元數(shù)據(jù)中的順序分塊編碼來(lái)查找上下文 def expand_doc(group): new_cands = [] group.sort(key=lambda x: int(x.metadata['chunk_id'].split('_')[-1])) id_set = set() file_id = group[0].metadata['file_id']

group_scores_map = {}
# 先找出該文件所有需要搜索的chunk_id
cand_chunks = []
for cand_doc in group:
    current_chunk_id = int(cand_doc.metadata['chunk_id'].split('_')[-1])
    group_scores_map[current_chunk_id] = cand_doc.metadata['score']
    for i in range(current_chunk_id - 200, current_chunk_id + 200):
        need_search_id = file_id + '_' + str(i)
        if need_search_id not in cand_chunks:
            cand_chunks.append(need_search_id)
where = {"chunk_id": {"$in": cand_chunks}}

ids,group_relative_chunks = get(where)

group_chunk_map = {int(item.metadata['chunk_id'].split('_')[-1]): item.page_content for item in group_relative_chunks}
group_file_chunk_num = list(group_chunk_map.keys())
for cand_doc in group:
    current_chunk_id = int(cand_doc.metadata['chunk_id'].split('_')[-1])
    doc = copy.deepcopy(cand_doc)
    id_set.add(current_chunk_id)
    docs_len = len(doc.page_content)
    for k in range(1, 200):
        break_flag = False
        for expand_index in [current_chunk_id + k, current_chunk_id - k]:
            if expand_index in group_file_chunk_num:
                merge_content = group_chunk_map[expand_index]
                if docs_len + len(merge_content) > CHUNK_SIZE:
                    break_flag = True
                    break
                else:
                    docs_len += len(merge_content)
                    id_set.add(expand_index)
        if break_flag:
            break

id_list = sorted(list(id_set))
id_lists = seperate_list(id_list)
for id_seq in id_lists:
    for id in id_seq:
        if id == id_seq[0]:
            doc = Document(page_content=group_chunk_map[id],
                            metadata={"score": 0, "file_id": file_id})
        else:
            doc.page_content += " " + group_chunk_map[id]
    doc_score = min([group_scores_map[id] for id in id_seq if id in group_scores_map])
    doc.metadata["score"] = doc_score
    new_cands.append(doc)
return new_cands

總結(jié) 在本文中，我們介紹了提高RAG模型檢索效果的高階技巧-窗口上下文檢索。我們首先回顧了基礎(chǔ)RAG的檢索流程和存在的問(wèn)題，然后介紹了窗口上下文檢索的原理和實(shí)現(xiàn)方法，最后通過(guò)一個(gè)實(shí)例展示了其效果。