Self-RAGの概要と実装方法

こんにちは、酒井です！
株式会社 EGGHEAD（エッグヘッド）という「製造業で生成 AI を活用したシステム開発」をしている会社の代表をしております。

https://egghead.co.jp?utm_source=zenn&utm_medium=social&utm_campaign=normal

はじめに

本記事では、検索結果と生成内容を自己評価する「Self-RAG（Self-Reflective Retrieval-Augmented Generation）」について解説し、LangGraph を使った実装方法を紹介します。

https://arxiv.org/abs/2310.11511

Self-RAG とは

従来のRAG(Naive RAG)の問題点

まず、検索拡張生成（Retrieval-Augmented Generation、RAG）は、ドキュメントをエンべディングを行ってベクトルデータベースに保存して、ユーザーのクエリから近しい文脈を取り出し、LLM に関連するコンテキストとして提供し外部知識に基づいた回答を行うという技術です。

しかしながら単純なこの手法では、検索をワンショットで行うため、正しい文脈が取得できていない可能性があります。Naive RAG では取得されたコンテキストが正しいかどうかの検証は行いません。

Self-RAG の特徴

Self-RAG（Self-Reflective Retrieval-Augmented Generation）は、Naive RAG の問題点を解決します。その特徴は以下の2点です。

オンデマンド検索（on-demand retrieval）

モデル自身が「この質問に対して外部知識が必要かどうか」を判断し、必要だと判断した場合のみ検索を実行するという仕組みです。これにより、必要のないときに検索を行いノイズになってしまうのを防ぐことができます。

自己反省（self-reflection）

検索結果や生成した内容を自ら評価し、検索結果との整合性や回答の有用性を検証するプロセスです。検証の結果、不十分または不正確と判断した場合は、検索のやり直しや回答の再生成を行います。これにより精度の高い回答を得ることができます。

従来の RAG との違い

従来の RAG システム(Naive RAG)と比較した Self-RAG の主な違いは以下の通りです。

特徴	従来のRAG	Self-RAG
検索の実行	必ず検索を実行	LLMが「検索が必要か」を判断し、必要な場合のみ検索を実行
検索結果の評価	検索結果をそのまま使用	「この情報は質問に関係あるか」を評価
情報の充分性の評価	検索は1回のみ	情報が足りない場合、追加で検索
回答の品質チェック	生成した回答の正確性を検証しない	「答えは正しいか」を自己評価
回答の改善有無	生成した回答をそのまま使用	回答が不十分と判断したら再生成
情報の取捨選択	検索した情報をすべて使用	関連性の低い情報を除外

Self-RAG の仕組み

Self-RAGのワークフロー

Self-RAGの処理フローを以下のようになっています。

1. ユーザーからの質問受付
   ユーザーが質問を入力します。

2. 検索が必要かどうかの判断
   LLMは「自分の知識だけで答えられるか、それとも外部情報が必要か」を考えます。

   • 外部情報が必要だと判断したら、検索を始めます
   • 自分の知識だけで答えられると判断したら、回答を作ります

3. 関連情報の検索と関連性があるかどうかを確認
   検索が必要な場合、関連情報を検索して「この情報は質問に関係あるか」を確認します。

   • 関係ない情報しか見つからない場合、検索キーワードを変えてもう一度探します

4. 情報が十分かどうかを評価
   取得した情報について「この情報だけで答えられるか」を評価します。

   • 情報が足りない場合、さらに関連文書を検索します。

5. 回答作成と自己チェック
   集めた情報をもとに回答を作り、「答えは正しいか」「情報は十分か」「ユーザーの知りたいことに答えているか」を自分でチェックします。

   • 回答が適切だと判断したら、そのままユーザーに返却します。
   • まだ不充分だと判断したら、もう一度回答を生成し直します。

6. 最終回答の返却
   自己チェックを通過した、信頼できる回答をユーザーに返します。

このように、Self-RAGはNaive RAGと異なり、検索の必要性判断から回答の品質評価まで、すべてのステップでモデル自身が主体的に判断・評価を行うことで、より正確で信頼性の高い回答を実現します。

Self-RAG の実装

実装準備

Google Colab での実装を前提とします。なお、ここからはLangGraphのチュートリアルを踏襲しています。

https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_self_rag

必要なライブラリのインストール

Self-RAG を実装するために、以下のライブラリをインストールします：

%pip install -U langchain_community tiktoken langchain-openai langchainhub chromadb langchain langgraph

API キーの設定

OpenAI API と Tavily Search API のキーを設定します。

import getpass
import os

def _set_env(key: str):
    if key not in os.environ:
        os.environ[key] = getpass.getpass(f"{key}:")

_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")

リトリーバーの作成

テスト用のデータとして、自分のブログ記事をインデックス化します。

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

urls = [
   "https://zenn.dev/egghead/articles/e106a34b0bb271",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

# ベクトルDBに追加
vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

Self-RAG を実装するためのコンポーネント

検索結果の評価機能

検索結果がユーザーの質問に関連するかを評価するためのコンポーネントを実装します。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# データモデル
class GradeDocuments(BaseModel):
    """検索された文書の関連性チェックのためのバイナリスコア"""
    binary_score: str = Field(
        description="文書が質問に関連しているか、'yes'または'no'で回答"
    )

# 関数呼び出し付きLLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)

# プロンプト
system = """あなたは検索された文書がユーザーの質問に関連しているかを評価する採点者です。\n
    厳密なテストである必要はありません。目的は誤った検索結果をフィルタリングすることです。\n
    文書がユーザーの質問に関連するキーワードや意味的な内容を含んでいる場合は、関連性があると評価してください。\n
    文書が質問に関連しているかどうかを示すために、'yes'または'no'のバイナリスコアを付けてください。"""
grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "検索された文書: \n\n {document} \n\n ユーザーの質問: {question}"),
    ]
)

retrieval_grader = grade_prompt | structured_llm_grader

回答生成機能

検索結果を基に回答を生成する機能を実装します。

from langchain import hub
from langchain_core.output_parsers import StrOutputParser

# プロンプト
prompt = hub.pull("rlm/rag-prompt")

# LLM
llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)

# 後処理
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# チェーン
rag_chain = prompt | llm | StrOutputParser()

ハルシネーションチェック

生成された回答が検索された文書に基づいているかを評価する機能を実装します。

# データモデル
class GradeHallucinations(BaseModel):
    """生成された回答にハルシネーションが存在するかのバイナリスコア"""
    binary_score: str = Field(
        description="回答が事実に基づいているか、'yes'または'no'"
    )

# LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeHallucinations)

# プロンプト
system = """あなたはLLMの生成内容が取得された事実のセットに基づいている/サポートされているかを評価する採点者です。\n
     'yes'または'no'のバイナリスコアを付けてください。'Yes'は回答が事実のセットに基づいている/サポートされていることを意味します。"""
hallucination_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "事実のセット: \n\n {documents} \n\n LLM生成: {generation}"),
    ]
)
hallucination_grader = hallucination_prompt | structured_llm_grader

回答評価機能

生成された回答がユーザーの質問に適切に対応しているかを評価する機能を実装します。

# データモデル
class GradeAnswer(BaseModel):
    """回答が質問に対応しているかを評価するバイナリスコア"""
    binary_score: str = Field(
        description="回答が質問に対応しているか、'yes'または'no'"
    )

# LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeAnswer)

# プロンプト
system = """あなたは回答が質問に対応している/解決しているかを評価する採点者です。\n
     'yes'または'no'のバイナリスコアを付けてください。'Yes'は回答が質問を解決していることを意味します。"""
answer_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "ユーザーの質問: \n\n {question} \n\n LLM生成: {generation}"),
    ]
)
answer_grader = answer_prompt | structured_llm_grader

クエリ書き換え機能

検索結果が不十分な場合に、より効果的な検索のためにクエリを最適化します。

# LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# プロンプト
system = """あなたは入力された質問をベクトルストア検索に最適化された、より良いバージョンに変換する質問リライターです。\n
     入力を見て、基本的な意味的意図/意味について推論してみてください。"""
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        (
            "human",
            "これが最初の質問です: \n\n {question} \n 改善された質問を作成してください。",
        ),
    ]
)
question_rewriter = re_write_prompt | llm | StrOutputParser()

LangGraph による Self-RAG の実装

グラフ状態の定義

LangGraph を使用して Self-RAG のワークフローを実装するために、まずグラフの状態を定義します。

from typing import List
from typing_extensions import TypedDict

class GraphState(TypedDict):
    """
    グラフの状態を表します。
    属性：
        question: 質問
        generation: LLM生成
        documents: 文書のリスト
    """
    question: str
    generation: str
    documents: List[str]

ノードの実装

グラフのノードとなる各関数を定義していきます。実装には先ほど作成したコンポーネントを使っていきます。

def retrieve(state):
    """
    文書を検索する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        state (dict): 検索された文書を含む新しいキー documents が状態に追加される
    """
    print("---検索---")
    question = state["question"]

    # 検索
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}


def generate(state):
    """
    回答を生成する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        state (dict): LLM生成結果を含む新しいキー generation が状態に追加される
    """
    print("---回答生成---")
    question = state["question"]
    documents = state["documents"]

    # RAG生成
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}


def grade_documents(state):
    """
    検索された文書が質問に関連しているかどうかを判断する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        state (dict): フィルタリングされた関連文書のみで documents キーを更新する
    """

    print("---文書の質問への関連性チェック---")
    question = state["question"]
    documents = state["documents"]

    # 各文書を評価
    filtered_docs = []
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        if grade == "yes":
            print("---評価: 文書は関連あり---")
            filtered_docs.append(d)
        else:
            print("---評価: 文書は関連なし---")
            continue
    return {"documents": filtered_docs, "question": question}


def transform_query(state):
    """
    より良い質問を生成するためにクエリを変換する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        state (dict): 言い換えられた質問で question キーを更新する
    """

    print("---クエリ変換---")
    question = state["question"]
    documents = state["documents"]

    # 質問を書き直す
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}

条件分岐付きエッジの設定

ノード間の遷移を決定する条件分岐を実装します。

def decide_to_generate(state):
    """
    回答を生成するか、質問を再生成するかを決定する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        str: 次に呼び出すノードのバイナリ決定
    """

    print("---評価済み文書の査定---")
    state["question"]
    filtered_documents = state["documents"]

    if not filtered_documents:
        # すべての文書が関連性チェックでフィルタリングされた
        # 新しいクエリを再生成する
        print(
            "---決定: すべての文書が質問に関連していないため、クエリを変換---"
        )
        return "transform_query"
    else:
        # 関連する文書があるので、回答を生成する
        print("---決定: 生成---")
        return "generate"


def grade_generation_v_documents_and_question(state):
    """
    生成結果が文書に基づいており、質問に答えているかどうかを判断する

    引数:
        state (dict): 現在のグラフ状態

    戻り値:
        str: 次に呼び出すノードの決定
    """

    print("---ハルシネーションチェック---")
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]

    score = hallucination_grader.invoke(
        {"documents": documents, "generation": generation}
    )
    grade = score.binary_score

    # ハルシネーションチェック
    if grade == "yes":
        print("---決定: 生成結果は文書に基づいている---")
        # 質問応答チェック
        print("---生成結果と質問の評価---")
        score = answer_grader.invoke({"question": question, "generation": generation})
        grade = score.binary_score
        if grade == "yes":
            print("---決定: 生成結果は質問に対応している---")
            return "useful"
        else:
            print("---決定: 生成結果は質問に対応していない---")
            return "not useful"
    else:
        print("---決定: 生成結果は文書に基づいていない、再試行---")
        return "not supported"

グラフのコンパイルと実行

最後に、定義したノードとエッジを使用してグラフを構築し、コンパイルします。
回答を生成した後評価を行っていて、回答が質問に関連していない場合、検索のためのクエリを変換し再度ベクトル検索を行う仕様になっています。一方、ハルシネーションしている(検索結果に基づかない回答)場合、回答生成からやり直しています。

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

# ノードを定義
workflow.add_node("retrieve", retrieve)  # 検索
workflow.add_node("grade_documents", grade_documents)  # 文書の評価
workflow.add_node("generate", generate)  # 生成
workflow.add_node("transform_query", transform_query)  # クエリ変換

# グラフを構築
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",  # 質問を変換する
        "generate": "generate",  # 回答を生成する
    },
)
workflow.add_edge("transform_query", "retrieve")
workflow.add_conditional_edges(
    "generate",
    grade_generation_v_documents_and_question,
    {
        "not supported": "generate",  # 文書にサポートされていない場合は再生成
        "useful": END,  # 役立つ回答の場合は終了
        "not useful": "transform_query",  # 役立たない回答の場合はクエリを変換
    },
)

app = workflow.compile()

実行例と評価

関連のある質問への対応

関連のある質問をしてみます。正しく検索ができて回答が生成されていることが分かります。モデルの精度が低いため、回答が微妙ですね。

inputs = {"question": "Corrective RAGについて教えて"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"ノード '{key}':")
    pprint("\n---\n")

# 最終的な生成結果
pprint(value["generation"])

実行結果：

---検索---
"ノード 'retrieve':"
'\n---\n'
---文書の質問への関連性チェック---
---評価: 文書は関連あり---
---評価: 文書は関連あり---
---評価: 文書は関連あり---
---評価: 文書は関連あり---
---評価済み文書の査定---
---決定: 生成---
"ノード 'grade_documents':"
'\n---\n'
---生成---
---幻覚チェック---
---決定: 生成結果は文書に基づいている---
---生成結果と質問の評価---
---決定: 生成結果は質問に対応している---
"ノード 'generate':"
'\n---\n'
('Corrective '
 'RAG（CRAG）は、情報検索と生成を組み合わせた手法で、特にLangGraphを用いた実装方法が紹介されています。この手法は、必要な情報を自ら検索し、活用することを目的としています。詳細については、関連する文献を参照してください。')

関連のない質問への対応

関連のない質問をしてみます。当然天気の情報は登録したドキュメントに載ってないので、無限ループしてしまってますね。

from pprint import pprint

# 実行
inputs = {"question": "明日の天気を教えてください"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"ノード '{key}':")
    pprint("\n---\n")

# 最終的な生成結果
pprint(value["generation"])

実行結果：

---検索---
"ノード 'retrieve':"
'\n---\n'
---文書の質問への関連性チェック---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価済み文書の査定---
---決定: すべての文書が質問に関連していないため、クエリを変換---
"ノード 'grade_documents':"
'\n---\n'
---クエリ変換---
"ノード 'transform_query':"
'\n---\n'
---検索---
"ノード 'retrieve':"
'\n---\n'
---文書の質問への関連性チェック---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価済み文書の査定---
---決定: すべての文書が質問に関連していないため、クエリを変換---
"ノード 'grade_documents':"
'\n---\n'
---クエリ変換---
"ノード 'transform_query':"
'\n---\n'
---検索---
"ノード 'retrieve':"
'\n---\n'
---文書の質問への関連性チェック---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価済み文書の査定---
---決定: すべての文書が質問に関連していないため、クエリを変換---
"ノード 'grade_documents':"
'\n---\n'
---クエリ変換---
"ノード 'transform_query':"
'\n---\n'
---検索---
"ノード 'retrieve':"
'\n---\n'
---文書の質問への関連性チェック---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価: 文書は関連なし---
---評価済み文書の査定---
---決定: すべての文書が質問に関連していないため、クエリを変換---
"ノード 'grade_documents':"

まとめ

ここまで読んでいただきありがとうございました！
今回はRAGが必要かどうか判断する分岐を入れてないので、無限ループが発生してしまう作りになってしまっていて改善の余地がありますが、LangGraphのチュートリアルを通じてSelf-RAGの実装方法を知ることができました。

【お仕事のご依頼】
弊社では生成 AI を用いたシステム開発を支援しています。
もしお仕事の相談があればこちらフォームよりお気軽にご相談ください！
https://forms.gle/RW1Kpaxwx64RBmgL6

メールでの問い合わせはこちらからよろしくお願いいたします。
[e-mail]