RAG 上級スキル - ウィンドウコンテキスト取得の実装方法

この記事では、RAG (Retrieval-Augmented Generation) モデルの検索パフォーマンスを向上させる高度な手法、つまりウィンドウコンテキスト検索を紹介します。まず、基本的な RAG 検索プロセスと既存の問題を確認し、次にウィンドウコンテキスト検索の原理と実装方法を紹介し、最後に例を通してその効果を示します。

画像の基礎RAG の問題と解決策RAG の基本的な検索プロセスRAG は、検索と生成を組み合わせた AI アプリケーションソリューションです。外部の知識ベース (Wikipedia など) を活用して、生成の品質と多様性を高めながら、与えられた質問に基づいて回答を生成できます。 RAG の中心的な考え方は、質問を知識ベース内のドキュメントと照合し、照合されたドキュメントを生成モデルへの入力として使用して、より関連性が高く豊富な回答を生成することです。

画像 RAG の取得プロセスは、次のステップに分けられます。

load: ドキュメントを読み込み、PDF をテキストデータとして読み込んだり、テーブルを複数のキーと値のペアに変換したりするなど、さまざまな形式のファイルをドキュメントに変換します。分割: ドキュメントをベクトルストレージに適した小さな単位に分割して、検索時にベクトルストレージとドキュメントのマッチングを容易にします。たとえば、「私は kxc です。歌、ダンス、ラップ、バスケットボールが好きです。」を「私は kxc です。」と「歌、ダンス、ラップ、バスケットボールが好きです。」という 2 つのデータブロック (通常はチャンクと呼ばれます) に分割します。埋め込み: ベクトル化に BERT や TF-IDF などのモデルを使用するなど、ドキュメントをベクトルとして表現します。 store: ベクトル化されたデータをブロックに分割し、ベクトルデータベースに保存します。取得: 質問とドキュメントのベクトルに基づいて、それらの間の類似度を計算し、類似度に基づいて最も関連性の高いドキュメントを検索結果として選択します。たとえば、コサイン類似度やドット積などのメトリックを使用して並べ替えます。クエリ: 取得したドキュメントを生成モデルへの入力として使用し、GPT-3 や T5 などのモデルを使用して、質問に基づいて回答を生成します。基本的な RAG の問題基本的な RAG の取得プロセスは単純ですが、主に分割と取得の手順でいくつかの問題が残っています。これらの問題は RAG の検索パフォーマンスに影響を及ぼし、不正確または不完全な応答を引き起こす可能性があります。

ブロックの分割が大きすぎると、検索時に同じブロック内に無関係なコンテンツが多くなり、質問の検索一致度に大きな影響を与え、不正確な検索につながります。たとえば、Wikipedia の記事を文書として取り上げると、この文書にはさまざまなトピックや詳細が含まれており、質問との関連性がほとんどない可能性があります。この文書を検索結果として使用すると、生成モデルによって無関係な情報や誤った情報が抽出され、回答の品質に影響する可能性があります。

分割ブロックが小さすぎると、検索一致度は向上しますが、最終クエリ段階では、コンテキスト情報のサポートが不足しているため、LLM に提供される情報は不正確な回答につながります。たとえば、Wikipedia の記事を複数の文に分割すると、各文には質問に関連性の高い情報のごく一部しか含まれない可能性があります。これらの文を検索結果として使用すると、生成モデルはそこから有用な情報を抽出できる可能性がありますが、重要なコンテキスト情報の一部を無視する可能性があり、その結果、回答の完全性に影響を及ぼします。

解決策 - ウィンドウコンテキスト取得イメージこの問題の一般的な解決策は、分割時にテキストを最小の意味単位に分割することです。取得時に、一致したドキュメントは直接使用されませんが、一致したドキュメントのコンテキストコンテンツは拡張され、統合されてから LLM に配信され、使用されます。このようにして、コンテキストの整合性を維持しながら検索の精度を向上させることができ、生成の品質と多様性が向上します。

具体的には、このソリューションの実装手順は次のとおりです。

分割中、テキストは文や段落などの最小の意味単位に分割され、各単位にはテキスト内の位置情報として一意の番号が割り当てられます。検索時には、質問と文書のベクトルに基づいて類似度を計算し、最も関連性の高い文書を検索結果として選択し、その番号を記録します。クエリを生成する際には、検索結果の数値（前後の単位数など）に応じてテキストからコンテキスト情報を取得し、それらを完全なドキュメントに連結して生成モデルの入力とし、質問に基づいた回答を生成します。ウィンドウコンテキスト検索の実践コンテキスト検索の実装のアイデア達成すべき最終目標から始めます。つまり、検索時に、ドキュメントを照合することによって、ドキュメントコンテンツに関連するコンテキストを拡張できなければなりません。この目標を達成するには、各ドキュメントとそのコンテキスト間の関連付けを確立する必要があります。この関係を確立するのは実はとても簡単です。分割された各ドキュメントに順番に番号を付けるだけです。検索時に、関連するコンテキストドキュメントの番号を現在のドキュメントの番号と照合し、コンテキストの内容を取得できます。

クロマベクトルライブラリに基づくコードの実践上記のアイデアを実践するには、分割段階でドキュメントの順序に基づいてドキュメントのエンコーディングをメタデータに書き込む必要があります。検索時には、メタデータ内の連続ブロックエンコーディングによってコンテキストが検索されます。具体的なコードは次のとおりです。

1. ブロックをエンコードし、分割時にメタデータを書き込みます。 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

ブログのコンテンツを読み込み、チャンク化し、インデックスを作成します。

loader = WebBaseLoader( web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",), bs_kwargs=dict( parse_only=bs4.SoupStrainer( class_=("post-content", "post-title", "post-header") ) ), ) doc = loader.load()

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

ここで、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 をメタデータに追加 chunk_id_counter += 1 for key,value in doc.metadata.items(): if not isinstance(value, (str, int, float, bool)): doc.metadata[key] = str(value)

ベクトルストア = Chroma.from_documents(ドキュメント = 分割、埋め込み = OpenAIEmbeddings())

2. 取得時に、メタデータ内の連続チャンクエンコーディングによってコンテキストを検索します。 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