RAGとは何なのか。勉強しながら作成してみる

こんにちは。42Tokyoに所属していますyamatoといいます。42Tokyoについては、以下のリンクからご覧ください
https://42tokyo.jp/

では、本題です
githubは
https://github.com/yamato1936/my_rag_project
です。

①RAGとはなにか

ChatGptやGeminiのような大規模言語モデル(LLM)は非常に強力ですが、
・ハルシネーションが発生する
・学習データに含まれない情報には答えられない
・回答の根拠が不明で、事実確認が難しい
などという弱点があります。これらを解決するためのものが、RAG(Retrieval-Augmented Generation)です。RAGを簡単に説明すると、「LLMが回答を生成する『前』に、外部の知識源から関連情報を検索し、それを参考資料としてLLMに渡す仕組み」です。
RAGには、以下のようなメリットがあります
・ハルシネーションの抑制
・特定の知識を外部知識としてLLMに与えられる
・出典の明記と事実確認ができる

②RAGの仕組み

RAGは大きく分けて、2つのフェーズが存在します。

1,準備フェーズ：知識のインデックス化

これは、ユーザーが質問をする前に、一度だけ行っておくオフラインの処理です

①読み込み：PDF、Webページ、テキストファイルなど、知識源としたいドキュメントを読み込みます。
②チャンク化：読み込んだドキュメントを、扱いやすいサイズ（例: 2000文字ごと）の小さな塊（チャンク）に分割します。
③ベクトル化：各チャンクを、Embeddingモデルを使って数値の配列（ベクトル）に変換します。このベクトルは、チャンクの意味的な位置を多次元空間上の座標として表現します。

ここは、少し詳しく解説します。以下のことをしています。
・embedding:テキストの意味を、多次元空間のベクトルに変換する
・ベクトル計算：ベクトル同士の計算で、意味の関係性(例えば、王様ー男性＋女性＝女王)をとらえる
・コサイン類似度：ベクトル間の角度を計算することで、意味の近さを判断し、関連情報を検索する

今回の場合、GoogleGenerativeAIEmbeddings(model="models/embedding-001")を使っていますので、そちらのほうで解説します。このモデルは、文脈を考慮する埋め込みモデルです。他には、単語ごとに座標が決まっているモデルもありますが、現在の主流は文脈を考慮するタイプです。
例えば、
I sat on the river bank. (訳：私は川の土手に座った)
I deposited money at bank.　(訳：私は銀行にお金を預けた)
この2つのbankは同じbankですが、違う意味です。なので、この2つは違う座標が与えられています。それにより、文章全体で一つのベクトルが作成されます。このベクトルを計算していくわけです。
ここの部分は、transformerが大いに関係あります。以下のページでtransformerについてまとめましたので、見ていただけると幸いです。
transformerとembedding

④保存：チャンクのテキストと、それに対応するベクトルをセットでベクトルデータベースに保存します

2,実行フェーズ：検索と生成

これは、ユーザーが質問をするたびに行われるオンラインの処理です

①質問をベクトル化: ユーザーの質問文も、準備フェーズで使ったのと同じEmbeddingモデルでベクトルに変換します。
②検索: 質問のベクトルと最も「意味が近い（距離が近い）」チャンクのベクトルを、ベクトルデータベースからいくつか探し出します。
③プロンプト作成: 「【参考資料】: (検索で見つかったチャンク) 【質問】: (ユーザーの質問) 【回答】:」という形式で、LLMに渡すためのプロンプトを組み立てます。
④生成 (Generate): 組み立てたプロンプトをLLMに送り、参考資料に基づいた回答を生成させます。

コード解説

では実際のコード見ていきましょう。githubにあるので、コード全文を見たい方はそちらからでも見ることができます。

import os
import shutil
from dotenv import load_dotenv
from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.chains import RetrievalQA

os：ファイルやディレクトリの存在確認、一覧表示などに使います。
shutil：より高度なファイル操作に使います。今回のコードでは、古いデータベースフォルダを消すために使用しています
from dotenv import load_dotenv：APIキーのような秘密情報を安全に管理するためのものです。

from langchain_google_genai import GoogleGenerativeAIEmbeddings, ChatGoogleGenerativeAI：GoogleのAIモデル関連の機能。GoogleGenerativeAIEmbeddingsは、embeddingモデルで、ChatGoogleGenerativeAIは、LLMです。

from langchain_community.vectorstores import Chroma：ベクトルデータを管理するためのデータベースです。

from langchain_community.document_loaders import PyPDFLoader：PDFファイルを開いて中のテキストを読み込むためのものです。

from langchain.text_splitter import RecursiveCharacterTextSplitter：テキスト分割機です。PDFの文章をチャンクに分割します。

from langchain.retrievers.multi_query import MultiQueryRetriever：情報取得の精度を高めるための、賢い検索エージェントです。
MultiQueryRetriever：ユーザーの質問をそのまま使うのではなく、まずLLMに質問を様々な角度から書き直させ、その複数の質問で検索をかけることで、より網羅的で精度の高い情報を探し出す役割を担います。

from langchain.chains import RetrievalQA：全ての部品を繋ぎ合わせて、一連の流れを管理する機能です。RetrievalQAは、質問を受け取り、情報を検索し、LLMに回答生成させる一連の流れの管理者のようなものです

def load_and_split_pdfs(docs_path: str):
    """
    指定されたパスからPDFを読み込み、テキストを分割（チャンキング）して
    LangChainのDocumentオブジェクトのリストとして返す。
    """
    all_docs = []
    if not os.path.isdir(docs_path):
        print(f"エラー: '{docs_path}' ディレクトリが見つかりません。")
        return all_docs
        
    print("ドキュメントを読み込んでいます...")
    for pdf_file in [f for f in os.listdir(docs_path) if f.lower().endswith(".pdf")]:
        file_path = os.path.join(docs_path, pdf_file)
        print(f"- {pdf_file} を読み込み中...")
        try:
            loader = PyPDFLoader(file_path)
            pages = loader.load()

            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=2000,  # 論文のような密なテキストはチャンクを大きめに
                chunk_overlap=200
            )
            documents = text_splitter.split_documents(pages)
            
            for doc in documents:
                doc.metadata['source'] = pdf_file
                doc.metadata['page'] = doc.metadata.get('page', 0) + 1

            all_docs.extend(documents)
        except Exception as e:
            print(f"  エラー: {file_path} の読み込み/分割に失敗しました。理由: {e}")
            
    return all_docs

①PDFの読み込み

loader = PyPDFLoader(file_path)
pages = loader.load()

まず、PyPDFLoaderが指定されたPDFファイルを読み込みます。loader.load()メソッドは、PDFの各ページを個別のDocumentオブジェクトとしてリストに格納します。このDocumentオブジェクトには、本文（page_content）と、ページ番号などの情報（metadata）が含まれています。

②テキストの分割

text_splitter = RecursiveCharacterTextSplitter(...)
documents = text_splitter.split_documents(pages)

学術論文の1ページには数千文字が含まれることもあり、そのままでは情報が密すぎるため、ベクトル検索の精度が落ちてしまいます。
そこでRecursiveCharacterTextSplitterを使い、ページごとのかたまりを、さらに意味的にまとまりのある小さなチャンク（このコードでは2000文字）に分割します。
chunk_size=2000: チャンクのおおよその文字サイズ。これを調整することで、LLMに渡すコンテキストの粒度を制御します。
chunk_overlap=200: チャンク間の重なり文字数。文章の途中で分割されて文脈が失われるのを防ぐための重要なパラメータです。
この処理により、検索対象が「ページ」から、より具体的な「段落」や「節」に近い単位になり、質問との関連性を正確に捉えやすくなります。

③メタデータの付与

for doc in documents:
    doc.metadata['source'] = pdf_file
    doc.metadata['page'] = doc.metadata.get('page', 0) + 1

チャンキングで生成された小さなDocumentオブジェクトの一つ一つに、それが**「どのPDFの」「何ページ目」から来たのかというメタデータ**を紐付けています。
この処理があるおかげで、RAGシステムは回答を生成した後に、「この回答は、nature16961.pdfの5ページ目を参考にしました」というように、正確な出典をユーザーに提示できます。これにより、システムの回答の信頼性と透明性が飛躍的に向上します。
doc.metadata.get('page', 0) + 1 という部分は、PyPDFLoaderが元々持っている0から始まるページ番号を取得し、人間にとって分かりやすい1から始まる形式に変換しています。

def main():
    """
    RAGアプリケーションのメイン実行関数
    """
    load_dotenv()

    # 既存のDBがあれば、新しい知識で作り直すために一度削除する
    if os.path.exists(CHROMA_DB_PATH):
        print(f"既存のデータベース'{CHROMA_DB_PATH}'を削除しています...")
        shutil.rmtree(CHROMA_DB_PATH)

    # PDFを読み込み、チャンクに分割する
    documents = load_and_split_pdfs(DOCS_PATH)
    if not documents:
        print(f"処理するドキュメントがありません。'{DOCS_PATH}' フォルダを確認してください。")
        return
    print(f"\n合計 {len(documents)} 個のドキュメントチャンクを準備しました。")
        
    # テキストをベクトル化し、データベースに保存する
    print("\nデータベースを構築しています...（時間がかかります）")
    embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
    db = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=CHROMA_DB_PATH
    )
    print("データベースの構築が完了しました。")

    # 複雑な質問を複数の簡単な質問に分解するMultiQueryRetrieverを準備
    llm = ChatGoogleGenerativeAI(
        model="gemini-1.5-flash", 
        temperature=0, 
        convert_system_message_to_human=True
    )
    retriever = MultiQueryRetriever.from_llm(
        retriever=db.as_retriever(search_kwargs={'k': 7}), # 検索するチャンク数を7に設定
        llm=llm
    )

    # 検索と回答生成を組み合わせたQAチェーンを作成
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )

    print("\nセットアップ完了。質問応答ループを開始します.\n")
    # ユーザーからの質問を受け付け、回答を生成するループ
    while True:
        question = input("質問を入力してください (終了するには 'exit' と入力): ")
        if question.lower() == 'exit':
            print("プログラムを終了します。")
            break
        
        if not question.strip():
            continue
            
        print("\n回答を生成中です...")
        try:
            response = qa_chain.invoke({"query": question})
            
            print("\n■ 回答:")
            print(response["result"])
            
            if response["source_documents"]:
                print("\n--- 出典 ---")
                sources = set()
                for doc in response["source_documents"]:
                    source_str = f"- {doc.metadata['source']} (Page: {doc.metadata['page']})"
                    sources.add(source_str)
                
                for source in sorted(list(sources)):
                    print(source)
            print("\n" + "="*50)

        except Exception as e:
            print(f"\nエラーが発生しました: {e}")

①準備とクリーンアップ

load_dotenv()
if os.path.exists(CHROMA_DB_PATH):
    shutil.rmtree(CHROMA_DB_PATH)

まず初めに、.envファイルからAPIキーを読み込みます。
次に、shutil.rmtreeを使って、もし既存のデータベースフォルダが存在すれば、それを毎回削除しています。これは、docsフォルダ内のPDFを変更した際に、古い情報が残ってしまうのを防ぐためのシンプルな設計です。これにより、プログラムは常に最新のドキュメントでデータベースを構築します。

②ドキュメントの読み込みとベクトル化

documents = load_and_split_pdfs(DOCS_PATH)
# ...
embeddings = GoogleGenerativeAIEmbeddings(...)
db = Chroma.from_documents(...)

前のステップで解説したload_and_split_pdfs関数を呼び出し、全ドキュメントのチャンクを取得します。
その後、Chroma.from_documentsが、それらのチャンクをGoogleGenerativeAIEmbeddingsを使って一つずつベクトルに変換し、ベクトルデータベースChromaDBに保存します。このステップが、アプリケーション起動時で最も時間がかかる処理です。

③ 高度な検索コンポーネントの準備

retriever = MultiQueryRetriever.from_llm(retriever=db.as_retriever(search_kwargs={'k': 7})

ここでは、検索の精度を高めるための「賢い検索エージェント」であるMultiQueryRetrieverを準備しています。db.as_retriever(search_kwargs={'k': 7})で「通常の検索なら上位7件のチャンクを探す」という基本設定を作り、それをMultiQueryRetrieverに渡しています。これにより、リトリーバーはユーザーの質問をまずLLMに解釈させて複数の検索クエリを生成し、より網羅的な検索を実行します。

④ QAチェーンの組み立て

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

これがアプリケーションの本体を組み立てる最終工程です。RetrievalQAチェーンは、これまで準備した各部品を統合し、一連の処理フローを定義します。
llm: 回答を生成するLLMを指定します。
retriever: 情報を検索する方法として、ステップ3で作成した賢いMultiQueryRetrieverを指定します。
chain_type="stuff": 検索で見つかった全てのチャンクを、そのままLLMへのプロンプトに「詰め込む（stuff）」という、シンプルで強力な方法を指定しています。
return_source_documents=True: 回答と合わせて、その根拠となったソースドキュメントも返すように設定します。これが出典表示機能の鍵となります。

⑤対話ループと出典の表示

while True:
    # ...
    response = qa_chain.invoke({"query": question})
    # ...

RAGシステムの司令塔：main関数のステップバイステップ解説

main関数は、これまで解説してきた様々な部品（ローダー、スプリッター、リトリーバー、LLM）を組み合わせ、一つのアプリケーションとして機能させるための、いわば「司令塔」です。その処理の流れを順に見ていきましょう。

def main():
    """
    RAGアプリケーションのメイン実行関数
    """
    load_dotenv()

    # 既存のDBがあれば、新しい知識で作り直すために一度削除する
    if os.path.exists(CHROMA_DB_PATH):
        print(f"既存のデータベース'{CHROMA_DB_PATH}'を削除しています...")
        shutil.rmtree(CHROMA_DB_PATH)

    # PDFを読み込み、チャンクに分割する
    documents = load_and_split_pdfs(DOCS_PATH)
    if not documents:
        print(f"処理するドキュメントがありません。'{DOCS_PATH}' フォルダを確認してください。")
        return
    print(f"\n合計 {len(documents)} 個のドキュメントチャンクを準備しました。")
        
    # テキストをベクトル化し、データベースに保存する
    print("\nデータベースを構築しています...（時間がかかります）")
    embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
    db = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=CHROMA_DB_PATH
    )
    print("データベースの構築が完了しました。")

    # 複雑な質問を複数の簡単な質問に分解するMultiQueryRetrieverを準備
    llm = ChatGoogleGenerativeAI(
        model="gemini-1.5-flash", 
        temperature=0, 
        convert_system_message_to_human=True
    )
    retriever = MultiQueryRetriever.from_llm(
        retriever=db.as_retriever(search_kwargs={'k': 7}), # 検索するチャンク数を7に設定
        llm=llm
    )

    # 検索と回答生成を組み合わせたQAチェーンを作成
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )

    print("\n★★★ セットアップ完了。質問応答ループを開始します。★★★\n")
    # ユーザーからの質問を受け付け、回答を生成するループ
    while True:
        question = input("質問を入力してください (終了するには 'exit' と入力): ")
        if question.lower() == 'exit':
            print("プログラムを終了します。")
            break
        
        if not question.strip():
            continue
            
        print("\n回答を生成中です...")
        try:
            response = qa_chain.invoke({"query": question})
            
            print("\n■ 回答:")
            print(response["result"])
            
            if response["source_documents"]:
                print("\n--- 出典 ---")
                sources = set()
                for doc in response["source_documents"]:
                    source_str = f"- {doc.metadata['source']} (Page: {doc.metadata['page']})"
                    sources.add(source_str)
                
                for source in sorted(list(sources)):
                    print(source)
            print("\n" + "="*50)

        except Exception as e:
            print(f"\nエラーが発生しました: {e}")

1 準備とクリーンアップ

load_dotenv()
if os.path.exists(CHROMA_DB_PATH):
    shutil.rmtree(CHROMA_DB_PATH)

まず初めに、.envファイルからAPIキーを読み込みます。
次に、shutil.rmtreeを使って、もし既存のデータベースフォルダが存在すれば、それを毎回削除しています。これは、docsフォルダ内のPDFを変更した際に、古い情報が残ってしまうのを防ぐためのシンプルな設計です。これにより、プログラムは常に最新のドキュメントでデータベースを構築します。

2 ドキュメントの読み込みとベクトル化

documents = load_and_split_pdfs(DOCS_PATH)
# ...
embeddings = GoogleGenerativeAIEmbeddings(...)
db = Chroma.from_documents(...)

前のステップで解説したload_and_split_pdfs関数を呼び出し、全ドキュメントのチャンクを取得します。
その後、Chroma.from_documentsが、それらのチャンクをGoogleGenerativeAIEmbeddingsを使って一つずつベクトルに変換し、ベクトルデータベースChromaDBに保存します。このステップが、アプリケーション起動時で最も時間がかかる処理です。

3 高度な検索コンポーネントの準備 (MultiQueryRetriever)

retriever = MultiQueryRetriever.from_llm(...)

ここでは、検索の精度を高めるための「賢い検索エージェント」であるMultiQueryRetrieverを準備しています。db.as_retriever(search_kwargs={'k': 7})で「通常の検索なら上位7件のチャンクを探す」という基本設定を作り、それをMultiQueryRetrieverに渡しています。これにより、リトリーバーはユーザーの質問をまずLLMに解釈させて複数の検索クエリを生成し、より網羅的な検索を実行します。

4 QAチェーンの組み立て (RetrievalQA)

qa_chain = RetrievalQA.from_chain_type(...)

これがアプリケーションの本体を組み立てる最終工程です。RetrievalQAチェーンは、これまで準備した各部品を統合し、一連の処理フローを定義します。
llm: 回答を生成するLLMを指定します。
retriever: 情報を検索する方法として、ステップ3で作成した賢いMultiQueryRetrieverを指定します。
chain_type="stuff": 検索で見つかった全てのチャンクを、そのままLLMへのプロンプトに「詰め込む（stuff）」という、シンプルで強力な方法を指定しています。
return_source_documents=True: 回答と合わせて、その根拠となったソースドキュメントも返すように設定します。これが出典表示機能の鍵となります。

5 対話ループと出典の表示

while True:
    # ...
    response = qa_chain.invoke(...)
    # ...

最後に、while True:ループでユーザーからの入力を待ち受けます。
qa_chain.invokeが呼ばれると、これまで設定した全ての処理（質問のベクトル化 → 検索 → 回答生成）が一気に実行されます。
返ってきたresponseオブジェクトから、回答本体（response["result"]）と、出典情報（response["source_documents"]）を取り出して、きれいに表示しています。出典の表示でset()を使っているのは、複数のサブクエリが同じチャンクを重複して見つけてきた場合に、重複を除いてユニークな出典のみを表示するためのテクニックです。

以上です。読んでいただきありがとうございました