LangGraph 学習記録5:高度なRAGフロー：自己反映機能を備えた堅牢なLLM検索システムの構築

Posted at 2025-04-05

Part 1: 高度なRAGの導入
- RAGの限界を理解する
- 高度なRAGのコンセプト
Part 2: 高度なRAGシステムの構築
- プロジェクトのセットアップと構造
- データ取り込みパイプライン
Part 3: コアRAGフローの実装
- ドキュメント検索
- ドキュメント評価
Part 4: 高度なRAG拡張機能
Part 5: すべてを統合する
- LangGraphフローの構築
- テストとデバッグ

Part 1: 高度なRAGの導入

RAGの限界を理解する

従来のRAGアーキテクチャ

Retrieval Augmented Generation (RAG)は、大規模言語モデル(LLM)を外部知識で強化する標準的なアプローチとなっています。従来のRAGアーキテクチャは、以下のような単純なプロセスに従います：

ユーザークエリを受け取る
ベクトルデータベースから類似ドキュメントを検索する
検索されたドキュメントを使ってプロンプトを拡張する
LLMが拡張されたプロンプトに基づいて回答を生成する

このアプローチは多くのアプリケーションで効果的ですが、ドキュメント検索の質に大きく依存し、検索されたドキュメントの関連性や生成された回答の正確性を検証するメカニズムが含まれていません。

一般的なRAGの課題

その人気にもかかわらず、従来のRAGはいくつかの課題に直面しています：

検索品質: RAGの効果は関連ドキュメントの検索に大きく依存します。関連性のないドキュメントは、質の低い回答につながります。
情報の欠如: 知識ベースにクエリに関する情報がない場合、従来のRAGにはフォールバックメカニズムがありません。
ハルシネーションのリスク: LLMは、検索されたドキュメントに根拠がなくても、もっともらしく見える回答を生成することがあります。
自己評価の欠如: 従来のRAGは、システムがユーザーの質問に適切に答えたかどうかを評価するメカニズムを提供していません。

これらの制限に対処するため、研究者たちは従来のRAGアーキテクチャにいくつかの強化を提案しており、Self-RAG、Corrective RAG、Adaptive RAGなどの高度なアプローチが生まれました。

高度なRAGのコンセプト

Self-RAGの概要

Self-RAGは、反省メカニズムを追加することで従来のRAGを大幅に改善します。回答を生成した後、システムは以下を評価します：

回答が検索されたドキュメントに根拠があるかどうか
回答がユーザーの質問に適切に答えているかどうか

この自己評価により、システムは潜在的な問題を特定し、回答の再生成や追加情報の取得など、修正アクションを取ることができます。

Corrective RAGの概要

Corrective RAGは、回答を生成する前に検索されたドキュメントの品質向上に焦点を当てています。システムはクエリに対する各ドキュメントの関連性を評価し、関連性のないものをフィルタリングします。

十分な関連ドキュメントが見つからない場合、Corrective RAGは自動的に検索を外部ソース（Webサーチエンジンなど）に拡大し、ベクトルストアからの情報を補完することができます。

Adaptive RAGの概要

Adaptive RAGは、クエリを最も適切な情報ソースに誘導する機能を導入します。常にベクトルストアの検索から始めるのではなく、Adaptive RAGシステムはまず以下を判断します：

クエリはベクトルストアから回答可能か
クエリは外部情報ソースを必要とするか

このアプローチは、不要な検索や外部検索を避けることで、効率を最適化します。

Self-RAG、Corrective RAG、Adaptive RAGの要素を組み合わせることで、従来のRAGアプローチの限界に対処する堅牢なシステムを構築できます。

Part 1 まとめ

高度なRAGアーキテクチャは、従来のRAGシステムを以下の点で大幅に改善します：

ドキュメント関連性の評価
回答品質の検証
外部情報の検索
最適な情報ソースへのクエリルーティング

これらの拡張機能により、ハルシネーションの減少、関連性の向上、そしてベクトルストアに関連情報がない場合でも、より広範囲のクエリに対応できるようになります。

Part 2: 高度なRAGシステムの構築

プロジェクトのセットアップと構造

Poetryを使った環境セットアップ

まず開発環境をセットアップしましょう。依存関係の管理とプロジェクト用の仮想環境作成が容易なPoetryを使用します。

プロジェクト用の新しいディレクトリを作成し、そこに移動します：

mkdir langraph-course
cd langraph-course

Poetry環境を初期化します：

poetry init

必要な依存関係を追加します：

poetry add beautifulsoup4 langchain langchain-community langchain-openai langchain-core langsmith chroma-hnswlib chromadb python-dotenv black isort pytest tavily-python

これにより以下がインストールされます：

BeautifulSoup: Webスクレイピングとドキュメント処理用
LangChain, LangGraph: コアフレームワーク
Chroma: ベクトルストレージ用
tavily-python: Web検索統合用
Pytest, black, isort: テストとコードフォーマット用

APIキーを保存するための.envファイルを作成します：

OPENAI_API_KEY=your_openai_api_key
LANGCHAIN_API_KEY=your_langchain_api_key
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=rag
TAVILY_API_KEY=your_tavily_api_key
PYTHONPATH=.

リポジトリ構造

LangGraphアプリケーションのアーキテクチャを反映するようにコードを整理します。LangGraphはノードとエッジの概念を中心に構築されているため、リポジトリ構造はこれらのコンポーネントを定義して接続しやすいものにする必要があります。

以下の構造を使用します：

langraph-course/
├── .env                  # 環境変数
├── main.py               # エントリーポイント
├── ingestion.py          # ドキュメント取り込みパイプライン
├── graph/                # グラフ定義
│   ├── __init__.py
│   ├── state.py          # グラフ状態定義
│   ├── graph.py          # ノードとエッジの接続
│   ├── consts.py         # 定数
│   └── nodes/            # ノード実装
│       ├── __init__.py
│       ├── retrieve.py
│       ├── grade_documents.py
│       ├── web_search.py
│       └── generate.py
└── chains/               # チェーン定義
    ├── __init__.py
    ├── retrieval_grader.py
    ├── generation.py
    └── tests/            # チェーンテスト
        ├── __init__.py
        └── test_chains.py

この構造は以下を分離します：

グラフ定義（ノードとエッジ）
ノード間を渡す状態
各ノードが実行するチェーン
チェーンのテスト

この構造を作成しましょう：

mkdir -p graph/nodes chains/tests
touch graph/__init__.py graph/state.py graph/graph.py graph/consts.py
touch graph/nodes/__init__.py
touch chains/__init__.py chains/tests/__init__.py
touch chains/tests/test_chains.py
touch main.py ingestion.py

テストインフラの設定

テストは堅牢なLLMアプリケーションを維持するために不可欠です。LLMベースのシステムのテストは非決定的な性質のため独自の課題がありますが、チェーンとノードの全体的な動作を検証するテストを作成することはできます。

chains/tests/test_chains.pyに基本的なテストインフラを設定しましょう：

def test_foo():
    """pytestが機能していることを確認するためのダミーテスト"""
    assert 1 == 1

テスト設定が機能することを確認するためにこのテストを実行します：

pytest . -sv

これは問題なくパスするはずです。アプリケーションを開発していく中で、各チェーンにより意味のあるテストを追加していきます。

データ取り込みパイプライン

RAGワークフローを実装する前に、ドキュメントをベクトルストアに準備する必要があります。以下を行うデータ取り込みパイプラインを作成します：

WebのURLからドキュメントをロードする
これらのドキュメントを小さな部分にチャンク分割する
各チャンクのベクトル埋め込みを作成する
これらの埋め込みをChromaベクトルデータベースに保存する

Webからのドキュメントのロード

LangChainのWebBaseLoaderを使用して、指定したURLからドキュメントをロードします。ingestion.pyに実装しましょう：

from dotenv import load_dotenv
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 環境変数のロード
load_dotenv()

# ロードするURL
urls = [
    "https://lilianweng.github.io/posts/2023-06-23-agent/",
    "https://lilianweng.github.io/posts/2023-03-15-prompt-engineering/",
    "https://lilianweng.github.io/posts/2023-10-25-adv-attack-llm/",
]

# URLからドキュメントをロード
docs = []
for url in urls:
    loader = WebBaseLoader(url)
    docs.extend(loader.load())

print(f"Loaded {len(docs)} documents")

このコードは、エージェントメモリ、プロンプトエンジニアリング、LLMに対する敵対的攻撃に関する3つの記事をロードします。これらのトピックが私たちの知識ベースを形成します。

ドキュメントのチャンク分割

次に、これらのドキュメントを効果的に埋め込みと検索ができる小さなチャンクに分割する必要があります。RecursiveCharacterTextSplitterを使用します：

# ingestion.pyに追加

# ドキュメントをチャンクに分割
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, 
    chunk_overlap=0
)

chunks = text_splitter.split_documents(docs)
print(f"Split into {len(chunks)} chunks")

チャンク間のオーバーラップなしで、250トークンの小さなチャンクサイズを使用しています。本番環境では、異なるチャンク分割戦略を試してみるとよいでしょう。

ベクトル埋め込みの作成

各チャンクの埋め込みを作成し、Chromaベクトルデータベースに保存します：

# ingestion.pyに追加

# ベクトルストアの作成
vectorstore = Chroma.from_documents(
    documents=chunks,
    collection_name="rag_chroma",
    embedding=OpenAIEmbeddings(),
    persist_directory="./.chroma"
)

# ディスクに永続化
vectorstore.persist()

print("Vector store created and persisted")

# 検索用のretrieverを作成
retriever = Chroma(
    collection_name="rag_chroma",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./.chroma"
).as_retriever()

このコードは：

OpenAIの埋め込みモデルを使用して各チャンクの埋め込みを作成
これらの埋め込みをChromaベクトルデータベースに保存
再利用できるようにデータベースをディスクに永続化
データベースを検索するためのretrieverを作成

ベクトルデータベースを作成するためにこのスクリプトを一度実行します：

python ingestion.py

実行後、ベクトルデータベースを含む.chromaディレクトリが表示されるはずです。

Part 2 まとめ

これでプロジェクト環境を設定し、ドキュメントをベクトルストアに取り込みました。設定には以下が含まれます：

Poetryで管理された仮想環境と必要なすべての依存関係
グラフベースのアーキテクチャを反映したよく整理されたリポジトリ構造
基本的なテストインフラ
ドキュメントのロード、チャンク分割、埋め込みを行うドキュメント取り込みパイプライン

この基盤により、次のセクションで高度なRAGワークフローを実装することができます。

Part 3: コアRAGフローの実装

ドキュメント検索

RAGフローの最初のステップは、ユーザーのクエリに基づいてベクトルストアから関連ドキュメントを取得することです。これをLangGraphフローのノードとして実装します。

検索ノードの実装

まず、graph/state.pyでグラフの状態を定義しましょう。この状態はグラフの実行中にノード間で渡されます：

from typing import List, TypedDict

class GraphState(TypedDict):
    """グラフ実行のための状態"""
    question: str
    generation: str
    web_search: bool
    documents: List[str]

状態に含まれるもの：

question: ユーザーの元の質問
generation: 生成された回答
web_search: オンライン検索が必要かどうかを示すフラグ
documents: 検索されたドキュメントまたは検索から取得したドキュメントのリスト

次に、graph/nodes/retrieve.pyに検索ノードを実装しましょう：

from typing import Any, Dict
from ..state import GraphState
from ingestion import retriever

def retrieve(state: GraphState) -> Dict:
    """質問に関連するドキュメントを検索する"""
    print("Retrieving...")
    
    # 状態から質問を抽出
    question = state["question"]
    
    # ドキュメントを検索
    documents = retriever.invoke(question)
    
    # 検索されたドキュメントで状態を更新
    return {
        "documents": [doc.page_content for doc in documents],
        "question": question  # 質問を通過させる
    }

このノードは：

現在のグラフ状態を取得
質問を抽出
retrieverを使用して関連ドキュメントを検索
状態を更新するために検索されたドキュメントと元の質問を返す

検索プロセスのテスト

検索プロセスが正しく機能することを確認するために、直接テストできます：

# テストファイルに追加するか、スクリプトで実行可能
from graph.nodes.retrieve import retrieve
from graph.state import GraphState

# テスト状態の作成
state = GraphState(
    question="エージェントメモリとは何ですか？",
    generation="",
    web_search=False,
    documents=[]
)

# 検索ノードを実行
result = retrieve(state)

# 結果を確認
print(f"Retrieved {len(result['documents'])} documents")

このテストは、retrieverが機能していて、エージェントメモリに関するクエリに関連するドキュメントを見つけられることを確認します。

ドキュメント評価

高度なRAGシステムの主要な拡張機能の1つは、検索されたドキュメントの品質を評価する能力です。検索された各ドキュメントがクエリに本当に関連しているかどうかを評価するドキュメント評価者を実装します。

ドキュメント評価チェーンの構築

chains/retrieval_grader.pyにドキュメント評価チェーンを実装しましょう：

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

# 出力スキーマを定義
class GradeDocument(BaseModel):
    """ドキュメント関連性のバイナリスコア"""
    binary_score: str = Field(..., description="ドキュメントが質問に関連しているか（yes/no）")

# LLMを作成
llm = ChatOpenAI(temperature=0)

# 構造化出力のLLMを作成
structured_llm = llm.with_structured_output(GradeDocument)

# システムプロンプトを作成
system_prompt = """
あなたはユーザーの質問に対する検索ドキュメントの関連性を評価する評価者です。
ドキュメントに質問に関連するキーワードや意味的な内容が含まれている場合、関連性があると評価してください。
ドキュメントが質問に関連しているかどうかを示すバイナリスコア（yes/no）を与えてください。
"""

# プロンプトテンプレートを作成
grade_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "ドキュメント: {document}\n\n質問: {question}")
])

# チェーンを作成
retrieval_grader = grade_prompt | structured_llm

このチェーンは：

ドキュメントと質問を入力として受け取る
LLMにドキュメントの質問に対する関連性を評価するよう促す
yes/noのバイナリスコアを持つ構造化出力を返す

ドキュメント評価ノードの実装

次に、graph/nodes/grade_documents.pyにドキュメント評価ノードを実装しましょう：

from typing import Dict
from ..state import GraphState
from chains.retrieval_grader import retrieval_grader

def grade_documents(state: GraphState) -> Dict:
    """検索されたドキュメントの関連性を評価する"""
    print("ドキュメントの関連性を確認中...")
    
    # 状態から質問とドキュメントを抽出
    question = state["question"]
    documents = state["documents"]
    
    # 関連するドキュメントをフィルタリング
    filtered_docs = []
    web_search = False
    
    for doc in documents:
        # ドキュメントを評価
        score = retrieval_grader.invoke({
            "question": question,
            "document": doc
        })
        
        # 関連性があれば保持
        if score.binary_score.lower() == "yes":
            filtered_docs.append(doc)
        else:
            # ドキュメントが関連性がない場合、Web検索をトリガー
            web_search = True
    
    # フィルタリングされたドキュメントとWeb検索フラグで状態を更新
    return {
        "documents": filtered_docs,
        "question": question,
        "web_search": web_search
    }

このノードは：

検索された各ドキュメントの関連性を評価
関連性のあるドキュメントのみを保持
ドキュメントが関連性がないと判断された場合、Web検索をトリガーするフラグを設定

ドキュメント評価のテスト

ドキュメント評価者をテストしましょう：

# chains/tests/test_chains.pyに追加
from chains.retrieval_grader import retrieval_grader
from ingestion import retriever

def test_retrieval_grader_answer_yes():
    """関連ドキュメントが正しく評価されることをテスト"""
    # 関連ドキュメントを取得
    question = "agent memory"
    docs = retriever.invoke(question)
    document = docs[0].page_content
    
    # ドキュメントを評価
    result = retrieval_grader.invoke({
        "question": question,
        "document": document
    })
    
    # 関連性があることをアサート
    assert result.binary_score.lower() == "yes"

def test_retrieval_grader_answer_no():
    """関連性のないドキュメントが正しく評価されることをテスト"""
    # エージェントに関するドキュメントを取得
    docs = retriever.invoke("agent memory")
    document = docs[0].page_content
    
    # 無関係な質問で評価
    result = retrieval_grader.invoke({
        "question": "ピザの作り方",
        "document": document
    })
    
    # 関連性がないことをアサート
    assert result.binary_score.lower() == "no"

これらのテストは、ドキュメント評価者が関連性のあるドキュメントと関連性のないドキュメントの両方を正しく識別できることを確認します。

Part 3 まとめ

RAGフローの主要コンポーネントを実装しました：

セマンティック検索を使用して関連ドキュメントを検索するドキュメント検索ノード
各ドキュメントの関連性を評価するドキュメント評価チェーン
関連性のないドキュメントをフィルタリングし、必要に応じてWeb検索をトリガーするドキュメント評価ノード

この強化された検索プロセスにより、プロンプト拡張に真に関連性のあるドキュメントのみを使用することが保証され、生成される回答の品質が向上します。

Part 4: 高度なRAG拡張機能

Web検索の統合

ベクトルストアに十分な関連情報が含まれていない場合、追加のコンテキストをWeb検索で強化できます。Tavilyサーチエンジンを使用するWeb検索ノードを実装します。

Web検索ノードの実装

graph/nodes/web_search.pyにWeb検索ノードを実装しましょう：

from typing import Dict
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.documents import Document
from ..state import GraphState

# Tavily検索ツールを初期化
search_tool = TavilySearchResults(max_results=3)

def web_search(state: GraphState) -> Dict:
    """質問に関する情報をWebで検索する"""
    print("Webを検索中...")
    
    # 状態から質問とドキュメントを抽出
    question = state["question"]
    documents = state["documents"]
    
    # Webを検索
    search_results = search_tool.invoke({"query": question})
    
    # すべての結果を1つの文字列に結合
    joined_results = "\n".join([result["content"] for result in search_results])
    
    # 検索結果からドキュメントを作成
    web_results = Document(page_content=joined_results)
    
    # Web結果をドキュメントに追加
    if documents:
        documents.append(web_results.page_content)
    else:
        documents = [web_results.page_content]
    
    # Web結果を含むドキュメントで状態を更新
    return {
        "documents": documents,
        "question": question
    }

このノードは：

ユーザーの質問を取得
Tavilyを使用してWebを検索
検索結果を1つのドキュメントに結合
このドキュメントを既存のドキュメントに追加

Tavily検索の統合

Tavilyサーチエンジンは、LLMアプリケーション向けに特別に設計されており、一般的な検索エンジンよりも焦点を絞った関連性の高い検索結果を提供します。Tavilyを使用するには、.envファイルに追加したAPIキーが必要です。

LangChainのTavilySearchResultsツールがAPI統合を処理し、Web検索をワークフローに簡単に組み込むことができます。

回答生成と検証

関連ドキュメント（ベクトルストアまたはWeb検索から）を取得した後、回答を生成し、その品質を検証する必要があります。

生成ノードの実装

chains/generation.pyに生成チェーンを実装しましょう：

from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# LLMを作成
llm = ChatOpenAI(temperature=0)

# hubからReActプロンプトを取得
prompt = hub.pull("langchain-ai/react-chat")

# チェーンを作成
generation_chain = prompt | llm | StrOutputParser()

このチェーンは、LLMが最終的な回答を生成する前にステップバイステップで考えるように導くLangChain hubからのReAct（Reasoning and Acting）プロンプトを使用します。

次に、graph/nodes/generate.pyに生成ノードを実装しましょう：

from typing import Dict
from ..state import GraphState
from chains.generation import generation_chain

def generate(state: GraphState) -> Dict:
    """検索されたドキュメントに基づいて回答を生成する"""
    print("回答を生成中...")
    
    # 状態から質問とドキュメントを抽出
    question = state["question"]
    documents = state["documents"]
    
    # ドキュメントをコンテキストとしてフォーマット
    context = "\n\n".join(documents)
    
    # 回答を生成
    generation = generation_chain.invoke({
        "context": context,
        "question": question
    })
    
    # 生成された回答で状態を更新
    return {
        "generation": generation,
        "documents": documents,
        "question": question
    }

このノードは：

すべてのドキュメントをコンテキスト文字列に結合
このコンテキストと元の質問で生成チェーンを呼び出す
生成された回答で状態を更新

ハルシネーション評価器の構築

Self-RAGの重要な革新の1つは、生成された回答が検索されたドキュメントに根拠があるかどうかを確認することです。chains/hallucination_grader.pyにハルシネーション評価器を実装しましょう：

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

# 出力スキーマを定義
class GradeHallucination(BaseModel):
    """生成された回答のハルシネーションに対するバイナリスコア"""
    binary_score: bool = Field(..., description="回答が事実に根拠付けられているか（yes/no）")

# LLMを作成
llm = ChatOpenAI(temperature=0)

# 構造化出力のLLMを作成
structured_llm = llm.with_structured_output(GradeHallucination)

# システムプロンプトを作成
system_prompt = """
あなたはLLM生成が一連のドキュメントに根拠付けられている/支持されているかどうかを評価する評価者です。
回答が事実に根拠付けられている/支持されていることを意味するyesのバイナリスコア（yes/no）を与えてください。
"""

# プロンプトテンプレートを作成
hallucination_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "事実のセット:\n{documents}\n\nLLM生成:\n{generation}")
])

# チェーンを作成
hallucination_grader = hallucination_prompt | structured_llm

このチェーンは、生成された回答が提供されたドキュメントに事実的に根拠付けられているかどうかを評価します。

回答評価器の構築

ハルシネーションのチェックに加えて、回答がユーザーの質問に適切に対応しているかどうかも確認したいと思います。chains/answer_grader.pyに回答評価器を実装しましょう：

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

# 出力スキーマを定義
class GradeAnswer(BaseModel):
    """回答品質のバイナリスコア"""
    binary_score: bool = Field(..., description="回答が質問に対応しているか（yes/no）")

# LLMを作成
llm = ChatOpenAI(temperature=0)

# 構造化出力のLLMを作成
structured_llm = llm.with_structured_output(GradeAnswer)

# システムプロンプトを作成
system_prompt = """
あなたは回答が質問に対応し解決しているかどうかを評価する評価者です。
回答が質問を解決することを意味するyesのバイナリスコア（yes/no）を与えてください。
"""

# プロンプトテンプレートを作成
answer_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "質問:\n{question}\n\n回答:\n{generation}")
])

# チェーンを作成
answer_grader = answer_prompt | structured_llm

このチェーンは、生成された回答がユーザーの質問に適切に対応しているかどうかを評価します。

適応型ルーティング

Adaptive RAGは、クエリを最も適切な情報ソースにルーティングする概念を導入します。ベクトルストアまたはWeb検索を使用するかどうかを決定する質問ルーターを実装します。

質問ルーターの実装

chains/router.pyに質問ルーターを実装しましょう：

from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# 出力スキーマを定義
class RouteQuery(BaseModel):
    """どのデータソースを使用するかを決定するルーター"""
    data_source: Literal["vector_store", "web_search"] = Field(
        ..., 
        description="ユーザーの質問を与えられたとき、web_searchまたはvector_storeにルーティングする選択"
    )

# LLMを作成
llm = ChatOpenAI(temperature=0)

# 構造化出力のLLMを作成
structured_llm = llm.with_structured_output(RouteQuery)

# システムプロンプトを作成
system_prompt = """
あなたはユーザーの質問をvector_storeまたはweb_searchにルーティングする専門家です。
vector storeにはエージェント、プロンプトエンジニアリング、敵対的攻撃に関連するドキュメントが含まれています。
これらのトピックに関する質問にはvector_storeを使用してください。
それ以外のすべてにはweb_searchを使用してください。
"""

# プロンプトテンプレートを作成
route_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{question}")
])

# チェーンを作成
question_router = route_prompt | structured_llm

このチェーンは、ユーザーの質問を分析し、ベクトルストアまたは直接Web検索にルーティングするかどうかを決定します。

条件付きエントリーポイントの設定

このルーターを使用して、グラフに条件付きエントリーポイントを作成し、質問タイプに基づいて異なる実行パスを可能にします。

ルーターをテストしましょう：

# chains/tests/test_chains.pyに追加
from chains.router import question_router

def test_router_to_vector_store():
    """関連する質問をベクトルストアにルーティングするテスト"""
    result = question_router.invoke({
        "question": "エージェントメモリとは何ですか？"
    })
    assert result.data_source == "vector_store"

def test_router_to_web_search():
    """無関係な質問をWeb検索にルーティングするテスト"""
    result = question_router.invoke({
        "question": "ピザの作り方は？"
    })
    assert result.data_source == "web_search"

これらのテストは、ルーターが関連する質問をベクトルストアに、その他の質問をWeb検索に正しく誘導することを確認します。

Part 4 まとめ

RAGシステムにいくつかの高度な拡張機能を実装しました：

ベクトルストアに関連情報がない場合のWeb検索統合
ReActプロンプトを使用して回答を生成する生成ノード
ハルシネーションと回答の質をチェックする評価器
クエリを適切な情報ソースに誘導する質問ルーター

これらのコンポーネントにより、必要に応じて外部知識にアクセスし、生成された回答の品質を検証することで、システムはより質の高い回答を提供できます。

Part 5: すべてを統合する

LangGraphフローの構築

高度なRAGシステムのすべてのコンポーネントを実装したので、それらを完全なLangGraphフローに接続しましょう。graph/graph.pyでグラフを定義します：

from dotenv import load_dotenv
from langgraph.graph import StateGraph
from langgraph.graph.edges import add_edge

# 環境変数をロード
load_dotenv()

# 定数をインポート
from .consts import (
    RETRIEVE, GRADE_DOCUMENTS, GENERATE, WEB_SEARCH
)

# ノードをインポート
from .nodes import retrieve, grade_documents, generate, web_search

# 状態定義をインポート
from .state import GraphState

# 評価器とルーターをインポート
from chains.hallucination_grader import hallucination_grader
from chains.answer_grader import answer_grader
from chains.router import question_router

# ルーティング機能を定義
def route_question(state: GraphState) -> str:
    """ベクトルストアかWeb検索を使用するかを決定する"""
    print("質問をルーティング中...")
    
    question = state["question"]
    
    # 質問をルーティング
    source = question_router.invoke({"question": question})
    
    # 適切なノードを返す
    if source.data_source == "web_search":
        return WEB_SEARCH
    else:
        return RETRIEVE

# ドキュメント評価ルーティング機能を定義
def decide_to_generate(state: GraphState) -> str:
    """生成するかWeb検索するかを決定する"""
    if state["web_search"]:
        return WEB_SEARCH
    else:
        return GENERATE

# 回答品質ルーティング機能を定義
def grade_generation(state: GraphState) -> str:
    """ハルシネーションと回答の質を評価する"""
    print("生成を評価中...")
    
    question = state["question"]
    documents = state["documents"]
    generation = state["generation"]
    
    # ハルシネーションをチェック
    hallucination_score = hallucination_grader.invoke({
        "documents": "\n\n".join(documents),
        "generation": generation
    })
    
    # 回答がドキュメントに根拠があれば
    if hallucination_score.binary_score:
        # 質問に答えているかチェック
        answer_score = answer_grader.invoke({
            "question": question,
            "generation": generation
        })
        
        # 質問に答えていれば完了
        if answer_score.binary_score:
            return "useful"
        else:
            # 質問に答えていなければWebを検索
            return "not_useful"
    else:
        # 根拠がなければ再生成
        return "not_supported"

# グラフを作成
workflow = StateGraph(GraphState)

# ノードを追加
workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEB_SEARCH, web_search)

# 条件付きエントリーポイントを設定
workflow.set_conditional_entry_point(
    route_question,
    {
        WEB_SEARCH: WEB_SEARCH,
        RETRIEVE: RETRIEVE
    }
)

# エッジを追加
workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edge(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEB_SEARCH: WEB_SEARCH,
        GENERATE: GENERATE
    }
)
workflow.add_edge(WEB_SEARCH, GENERATE)
workflow.add_conditional_edge(
    GENERATE,
    grade_generation,
    {
        "not_supported": GENERATE,     # ハルシネーションがあれば再生成
        "useful": "end",               # 有用であれば終了
        "not_useful": WEB_SEARCH       # 有用でなければWebを検索
    }
)

# グラフをコンパイル
workflow.compile()

# グラフを視覚化
workflow.print_graph("graph.png")

このグラフは：

質問ルーターを使用して初期ノード（検索またはWeb検索）を決定
検索後、ドキュメントを評価し、生成するかWeb検索するかを決定
生成後、ハルシネーションと関連性について回答を評価
この評価に基づいて、回答を返す、再生成する、またはWebを検索する

最後に、main.pyファイルを更新してこのグラフを使用しましょう：

from dotenv import load_dotenv
from graph.graph import workflow

# 環境変数をロード
load_dotenv()

def main():
    """高度なRAGワークフローを実行する"""
    # 質問を定義
    question = "エージェントメモリとは何ですか？"
    
    # グラフ状態を初期化
    state = {"question": question}
    
    # グラフを実行
    result = workflow.invoke(state)
    
    # 結果を表示
    print("\n最終回答:")
    print(result["generation"])

if __name__ == "__main__":
    main()

このスクリプトを実行して、高度なRAGシステムの動作を確認しましょう：

python main.py

テストとデバッグ

エンドツーエンドテスト

異なるタイプの質問でシステムをテストしましょう：

ベクトルストアのトピックに関する質問：
- "エージェントメモリとは何ですか？"
- "プロンプトエンジニアリング技術を説明してください。"
- "LLMに対する敵対的攻撃はどのように機能しますか？"
Web検索が必要な質問：
- "フランスの首都は何ですか？"
- "ピザの作り方は？"
- "2023年の主要なイベントは何でしたか？"

各質問について、以下を観察できます：

クエリがグラフのどのパスを通るか
ドキュメントが関連性でフィルタリングされるか
Web検索がトリガーされるか
最終回答の品質

LangSmithでのトレース

LangSmithは、LangGraphアプリケーションの理解とデバッグに役立つ強力なトレース機能を提供します。LangSmithトレースが有効（環境変数を通じて）になっていると、以下が可能です：

グラフを通る実行パスの表示
各ノードの入力と出力の検査
各LLM呼び出しのプロンプトと完了の検査
ワークフローのボトルネックや問題の特定

トレースを表示するには、LangSmithダッシュボードにアクセスし、プロジェクト（設定では "rag"）を選択します。各トレースは、特定のクエリに対するグラフの完全な実行を示します。

Part 5 まとめ

すべてのコンポーネントを完全な高度なRAGシステムに統合することに成功しました：

トピックに基づいてクエリを適応的にルーティングするグラフ
ドキュメントと回答の品質に基づいて異なる実行パスを可能にする条件付きエッジ
回答が根拠付けられ関連性があることを確保する包括的な反省メカニズム
知識ベース外のクエリに対するWeb検索との統合

このシステムは従来のRAGアプローチから大幅に改善されており、より広範囲のクエリに対してより高品質で信頼性の高い回答を提供します。

結論

この記事では、いくつかの最先端の研究論文からテクニックを組み合わせた高度なRAGシステムを構築しました：

Self-RAG: 回答品質を検証するための反省の追加
Corrective RAG: 検索されたドキュメントのフィルタリングとキュレーション
Adaptive RAG: 最適な情報ソースへのクエリルーティング

私たちの実装は、LangGraphの強力なグラフベースのアーキテクチャを活用して、柔軟で保守可能、そしてテスト可能なシステムを作成しました。コードをノードとエッジの周りに構造化することで、理解、デバッグ、拡張が容易なシステムを作成しました。

このアプローチの主な利点は以下の通りです：

検索品質の向上: クエリに本当に関連するドキュメントのみを使用
ハルシネーションの削減: 回答が検索されたドキュメントに根拠があることを検証
知識の拡大: ベクトルストアに情報がない場合、自動的にWebを検索
回答検証: 回答が実際にユーザーの質問に対応していることを確認

この高度なRAGシステムは、検索拡張生成における最新の技術を代表し、従来のアプローチよりも信頼性が高く有用な回答を提供します。

これらのテクニックを適切に構造化された本番指向のコードベースに実装することで、特定のアプリケーションやドメインに拡張しカスタマイズできる堅牢な基盤を作成しました。

🔍 このブログ記事は、Self-RAG、Corrective RAG、Adaptive RAGといった最新の研究論文に基づいて、LangGraphを使用した高度なRAG（検索拡張生成）システムの実装方法を段階的に解説しています。通常のRAGシステムに比べ、ドキュメントの関連性評価、回答の妥当性検証、Web検索の統合、質問ルーティングなどの機能を追加することで、より高品質な回答を得ることができます。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

LangGraph 学習記録5:高度なRAGフロー：自己反映機能を備えた堅牢なLLM検索システムの構築

目次

Part 1: 高度なRAGの導入

RAGの限界を理解する

従来のRAGアーキテクチャ

一般的なRAGの課題

高度なRAGのコンセプト

Self-RAGの概要

Corrective RAGの概要

Adaptive RAGの概要

Part 1 まとめ

Part 2: 高度なRAGシステムの構築

プロジェクトのセットアップと構造

Poetryを使った環境セットアップ

リポジトリ構造

テストインフラの設定

データ取り込みパイプライン

Webからのドキュメントのロード

ドキュメントのチャンク分割

ベクトル埋め込みの作成

Part 2 まとめ

Part 3: コアRAGフローの実装

ドキュメント検索

検索ノードの実装

検索プロセスのテスト

ドキュメント評価

ドキュメント評価チェーンの構築

ドキュメント評価ノードの実装

ドキュメント評価のテスト

Part 3 まとめ

Part 4: 高度なRAG拡張機能

Web検索の統合

Web検索ノードの実装

Tavily検索の統合

回答生成と検証

生成ノードの実装

ハルシネーション評価器の構築

回答評価器の構築

適応型ルーティング

質問ルーターの実装

条件付きエントリーポイントの設定

Part 4 まとめ

Part 5: すべてを統合する

LangGraphフローの構築

テストとデバッグ

エンドツーエンドテスト

LangSmithでのトレース

Part 5 まとめ

結論