0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

LLMを自律実行!Agentic RAGで複雑な推論を解くLangGraph実装

0
Posted at

多くのRAGシステム開発者が直面する課題、それは「単一の検索では複雑な質問に答えきれない」「LLMが情報をうまく活用できない」という壁ではないでしょうか。特に、複数の情報源を横断したり、反復的な推論が必要なビジネスロジックを実装しようとすると、従来のRAGでは限界が見えてきます。

この記事では、そんな課題を解決する「Agentic RAG」の概念と、それをLangChainエコシステムで最も柔軟に実装できるフレームワーク「LangGraph」を用いた具体的な構築方法を解説します。LLMが自律的に思考し、必要な情報を取得・評価しながら複雑な推論を重ねていくAgentic RAGの仕組みを理解し、実用的なLLMアプリケーション開発の一歩を踏み出しましょう。

Agentic RAGとは?従来のRAGとの決定的な違い

このセクションでは、Agentic RAGがどのようなもので、なぜ従来のRAGでは対応できなかった課題を解決できるのかを解説します。

従来のRetrieval-Augmented Generation (RAG) は、ユーザーの質問に対して一度の検索を行い、その検索結果を基にLLMが回答を生成するシンプルなワークフローが主流でした。しかし、このアプローチでは以下のような限界がありました。

  • 単一検索の限界: 質問が複雑で、複数の情報源から断片的な情報を集めて統合する必要がある場合、一度の検索では必要な情報を網羅できない。
  • 固定的なワークフロー: 質問のタイプや検索結果の内容に応じて動的に戦略を変えることができず、常に同じ手順で処理されるため、汎用性に欠ける。
  • 推論の浅さ: LLMが検索結果を「そのまま」与えられた情報として扱うため、深い推論や情報評価が難しい。

これらの課題を解決するために登場したのが「Agentic RAG」です。Agentic RAGは、RAGに自律的なAIエージェントの概念を組み込むことで、LLMが能動的に思考し、動的な意思決定、反復的な推論、適応的な検索戦略を可能にする先進的なアーキテクチャです。

LLMエージェントが自律的に動く仕組み

Agentic RAGにおけるLLMは、単なるテキスト生成器ではなく、以下のような役割を担う「エージェント」として機能します。

  1. 問題の分析: ユーザーの質問を分析し、どのような情報が必要か、どのツールを使うべきかを判断します。
  2. ツール選択と実行: ベクトルデータベースからの情報検索、Web検索、SQLクエリ実行など、利用可能なツールの中から最適なものを選択し、実行します。
  3. 情報評価と反復: ツールから得られた情報を評価し、質問に答えるのに十分かを判断します。情報が不足している場合や、さらに深掘りが必要な場合は、再び問題分析に戻り、異なるツールやクエリで情報を収集します。
  4. 最終回答の生成: 必要な情報が揃ったと判断した時点で、それらを統合し、最終的な回答を生成します。

この一連のプロセスは、まるで人間が調査・思考する過程に似ています。LLMが自律的に思考(Reasoning)と行動(Action)を繰り返すパターンは「ReActパターン」として知られており、Agentic RAGの基本的なアーキテクチャの一つです。

出典: LangChain Blog - Agentic RAG: The Next Evolution of RAG

LangGraphでAgentic RAGを構築する理由と基本概念

このセクションでは、なぜLangGraphがAgentic RAGの構築に最適なのか、その基本的な概念と構成要素を説明します。

Agentic RAGのような複雑な推論プロセスを実装するには、エージェントの状態管理、条件分岐、ループ処理を柔軟に記述できるフレームワークが必要です。そこで登場するのが「LangGraph」です。

出典: LangGraph Documentation

LangGraphは、LangChainを拡張したステートフルなAIエージェントをグラフとして構築するための低レベルなオーケストレーションフレームワークです。LangChain v0.1のリリースと共に、自律的なエージェントサイクルを動かすエンジンとして登場しました。

LangGraphの主要な構成要素

LangGraphは、エージェントが複雑なタスクを確実に処理できるよう、以下の主要な概念で構成されます。

  • グラフの状態 (State): エージェントの現在の状況を表す情報。通常、これまでの会話履歴(メッセージのリスト)や、ツールから取得した検索結果などが含まれます。LangGraphではTypedDictで定義します。
  • ノード (Node): グラフにおける処理の単位です。LLMの呼び出し、ツール実行、情報評価など、特定のタスクを実行します。各ノードは状態を受け取り、状態を更新して次のノードに渡します。
  • エッジ (Edge): ノード間の接続を表します。エッジには「通常のエッジ」と「条件付きエッジ」があります。
    • 通常のエッジ: あるノードの処理が完了したら、常に次の特定のノードに進みます。
    • 条件付きエッジ: ノードの処理結果に基づいて、次に進むべきノードを動的に決定します。これにより、エージェントは状況に応じた柔軟な意思決定が可能になります。

出典: LangGraph Documentation - Core Concepts

これらの要素を組み合わせることで、LLMが思考し、ツールを呼び出し、その結果を評価して次のアクションを決定するという、Agentic RAGの動的なワークフローをグラフとして視覚的かつ論理的に構築できます。

LangGraphによるAgentic RAGの実装例

このセクションでは、LangGraphを用いたAgentic RAGの具体的な実装例をステップバイステップで解説します。

ここでは、ユーザーの質問に対して、LLMが「ベクトルデータベース検索」と「Web検索」という2つのツールを自律的に使い分け、最終的な回答を生成するAgentic RAGシステムを構築します。

前提・環境構築

  • Python 3.9以上
  • LangChain, LangGraphライブラリ
  • OpenAI APIキー (LLM呼び出し用)
  • Tavily APIキー (Web検索ツール用、オプション)

必要なライブラリをインストールします。

pip install -qU langchain langchain-openai langgraph langchain_community

環境変数を設定します(実際には.envファイルなどから読み込むのが推奨されます)。

import os

# 環境変数の設定 (実際には .env ファイルなどから読み込む)
# os.environ["OPENAI_API_KEY"] = "sk-..." # ご自身のOpenAI APIキーを設定
# os.environ["TAVILY_API_KEY"] = "tvly-..." # Tavily APIキーを設定

グラフの状態定義

LangGraphのグラフ全体で共有される状態を定義します。ここでは、会話履歴を保持するmessagesリストを定義します。

from typing import TypedDict, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

class AgentState(TypedDict):
    """
    LangGraphの状態を定義するTypedDict。
    メッセージのリストと、必要に応じて検索結果などを保持する。
    """
    messages: List[BaseMessage]
    # 検索結果や評価結果などを追加する場合はここに定義

ツール定義

Agentic RAGでは、LLMエージェントが利用するツールが重要です。ここでは、ダミーのベクトル検索ツールとWeb検索ツールを定義します。実際には、ChromaやPineconeなどのベクトルDB、Tavily APIなどを利用して実装します。

# LLMの初期化
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ツール定義の例 (実際にはベクトルDBやWeb検索ツールをここに実装)
# ここではダミーのツールを定義
@tool
def retrieve_documents(query: str) -> str:
    """
    ベクトルデータベースから関連文書を検索するツール。
    例: 内部ドキュメント、FAQなど。
    """
    print(f"DEBUG: Retrieving documents for query: {query}")
    # 実際にはPineconeやChromaから検索するロジック
    if "GPU" in query:
        return "GPUはGraphics Processing Unitの略で、画像処理や並列計算に特化したプロセッサです。大量のデータを同時に処理できるため、AI/MLの学習にも広く使われます。"
    return "関連文書は見つかりませんでした。"

@tool
def web_search(query: str) -> str:
    """
    Tavily APIなどの外部サービスを使用してWeb検索を実行するツール。
    最新情報や一般的な知識の検索に利用されます。
    """
    print(f"DEBUG: Performing web search for query: {query}")
    # 実際にはTavily APIを呼び出すロジック
    # from langchain_community.tools.tavily_research import TavilySearchResults
    # search = TavilySearchResults(max_results=3)
    # return search.run(query)
    if "GPU" in query and "仕組み" in query:
        return "Web検索結果: GPUは、多数の小さなコア(ALU)を持ち、これらが並列に動作することで、画像レンダリングや科学計算などの複雑なタスクを高速に処理します。CPUが少数の強力なコアで逐次処理を得意とするのに対し、GPUは多数の単純な処理を同時に行うことに優れています。"
    return "Web検索結果は見つかりませんでした。"

# LLMにツールを認識させるためにバインド
# LLMはプロンプトに応じてこれらのツールを呼び出すか判断するようになる
llm_with_tools = llm.bind_tools([retrieve_documents, web_search])

ノードと条件付きエッジの定義

LangGraphの核となる、各処理ステップ(ノード)と、その間の遷移ロジック(エッジ)を定義します。

# エージェントのノード定義
def call_model(state: AgentState):
    """
    LLMを呼び出し、思考とアクション(ツール呼び出しまたは最終回答)を生成するノード。
    """
    messages = state["messages"]
    response = llm_with_tools.invoke(messages) # ツールをバインドしたLLMを呼び出す
    return {"messages": messages + [response]}

def should_continue(state: AgentState) -> str:
    """
    LLMの応答に基づいて、次のアクションを決定する条件分岐ノード。
    LLMがツール呼び出しを提案した場合はツール実行ノードへ、
    そうでなければ最終回答生成ノードへ遷移します。
    """
    last_message = state["messages"][-1]
    # LLMがツール呼び出しを提案しているかチェック
    if last_message.tool_calls:
        print(f"DEBUG: LLM wants to call tools: {last_message.tool_calls}")
        return "call_tool"
    else:
        print("DEBUG: LLM wants to generate final answer.")
        return "end"

def call_tool_node(state: AgentState):
    """
    LLMが提案したツールを実行し、結果を状態に追加するノード。
    """
    last_message = state["messages"][-1]
    tool_outputs = []
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        
        # ツール名に基づいて適切なツールを呼び出す
        # ここではツール名を文字列で比較していますが、より堅牢な方法として
        # ツール名をマッピングした辞書を使うこともできます。
        if tool_name == "retrieve_documents":
            output = retrieve_documents.invoke(tool_args)
        elif tool_name == "web_search":
            output = web_search.invoke(tool_args)
        else:
            output = f"Unknown tool: {tool_name}"
        
        # ツール実行結果をAIMessageとして状態に追加
        tool_outputs.append(AIMessage(content=f"Tool output for {tool_name}: {output}"))
    
    return {"messages": state["messages"] + tool_outputs}

グラフの構築と実行

定義したノードとエッジを使って、Agentic RAGのワークフローをグラフとして構築します。

# グラフの構築
workflow = StateGraph(AgentState)

# ノードの追加
workflow.add_node("llm_node", call_model)
workflow.add_node("tool_node", call_tool_node)

# エントリーポイントの設定 (最初にLLMノードから開始)
workflow.set_entry_point("llm_node")

# エッジの追加
# LLMノードから、条件付きエッジでツール呼び出しor終了を判断
workflow.add_conditional_edges(
    "llm_node",
    should_continue,
    {
        "call_tool": "tool_node", # ツール呼び出しが必要ならツールノードへ
        "end": END,                # 最終回答なら終了
    },
)
# ツールノードの実行後、再度LLMノードに戻って思考を継続
workflow.add_edge("tool_node", "llm_node")

# グラフをコンパイルしてアプリケーションとして利用可能にする
app = workflow.compile()

# 実行例
if __name__ == "__main__":
    print("--- Agentic RAG Execution Start (Complex Query) ---")
    inputs = {"messages": [HumanMessage(content="GPUの仕組みを教えて")]}
    
    # stream()で逐次的に状態の更新を見ることができる
    for s in app.stream(inputs):
        if "__end__" not in s:
            print(s)
            print("---")
    
    # 最終的な回答を取得
    final_state = app.invoke(inputs)
    print("\nFinal Answer (Complex Query):")
    print(final_state["messages"][-1].content)

    print("\n--- Agentic RAG Execution Start (Simple Query) ---")
    inputs_simple = {"messages": [HumanMessage(content="日本の首都は?")]}
    
    for s in app.stream(inputs_simple):
        if "__end__" not in s:
            print(s)
            print("---")
    
    final_state_simple = app.invoke(inputs_simple)
    print("\nFinal Answer (Simple Query):")
    print(final_state_simple["messages"][-1].content)

このコードを実行すると、GPUの仕組みを教えてという複雑な質問に対しては、LLMが自律的にretrieve_documentsweb_searchツールを呼び出し、情報を収集してから回答を生成する様子が確認できます。一方、日本の首都は?というシンプルな質問に対しては、LLMが直接回答を生成し、ツール呼び出しをスキップする振る舞いが見られます。

データ前準備と質問応答CLIの例(概念)

実際のAgentic RAGアプリケーションでは、事前にドキュメントをベクトル化してデータベースに格納するステップが必要です。

# データ前準備のコマンド例 (概念)
# このスクリプトは、dataset.jsonlを読み込み、RecursiveCharacterTextSplitterで分割、
# OpenAI Embeddingでベクトル化し、Pineconeインデックスなどに保存する処理を実行します。
# 実際の実装はLangChainのドキュメントや各ベクトルDBのSDKを参照してください。
# python load2vector_cli.py dataset.jsonl

上記のLangGraphコードをCLIとしてラップする形で、質問応答アプリケーションを作成できます。

# 質問応答CLIの例 (概念)
# このスクリプトは、Agentic RAGシステムを初期化し、ユーザーの質問を処理します。
# python agentic_rag_cli.py "GPU の仕組みを教えて"

Agentic RAGにおけるよくある課題と解決策

このセクションでは、Agentic RAGを開発・運用する上で直面しやすい課題と、それらに対する具体的な回避策やベストプラクティスを解説します。

1. LLMのハルシネーション(幻覚)と誤ったツール選択

ハマりどころ: Agentic RAGはLLMが自律的に判断するため、不正確な情報に基づいて誤った推論や回答を生成するハルシネーションのリスクが依然として存在します。特に、検索結果が不十分な場合や、LLMの判断が誤った場合に、不適切なツールを選択したり、誤ったクエリを生成したりすることがあります。

回避策:

  • 自己評価(Reflection/Self-Critique)の導入: 検索結果やドラフト回答を別のLLM(または同じLLMの別ロール)が評価し、「この検索結果は質問に答えるのに十分か」「回答に矛盾はないか」を判定するステップをグラフに組み込みます。不十分と判断された場合は、再検索や回答の書き直しに進みます。
  • 厳密なプロンプトエンジニアリング: ツール利用の指示、思考プロセス、出力フォーマットなどを明確に定義したシステムプロンプトを使用し、LLMが意図した通りに振る舞うように誘導します。ツールの説明文も具体的に記述し、LLMがツールの用途を正確に理解できるようにします。
  • ガードレールの設定: エージェントの思考プロセスやアクションに厳格なガードレールを設定し、予期せぬ動作や不適切な回答を防ぎます。例えば、特定のキーワードが含まれる場合は警告を出す、特定のツールは特定の条件下でのみ使用可能にする、といったルールを設けます。
  • Human-in-the-Loop: 重要な判断や最終回答の生成前に人間による介入(承認、修正など)の仕組みを導入し、品質を保証します。特に、機密情報を取り扱うシステムや、高精度が求められるビジネスロジックにおいては有効です。

2. レイテンシとコストの増大

ハマりどころ: Agentic RAGは、LLMの呼び出し回数が増え、複数のステップやループを伴うため、従来のRAGと比較して応答速度(レイテンシ)が長くなり、API利用コストが増大する傾向があります。特に、複雑なクエリや、複数のツールを繰り返し利用するシナリオで顕著になります。

回避策:

  • Adaptive RAGの導入: 質問の複雑さに応じて、適切なRAGパターン(単純な質問には従来のRAG、複雑な質問にはAgentic RAG)を動的に切り替えることで、精度とコスト・レイテンシのバランスを取ります。
  • 並列処理の活用: ドキュメント評価や複数のツール呼び出しなど、並列化可能な処理にはThreadPoolExecutorなどを利用して処理速度を向上させます。
  • キャッシュ戦略: 頻繁にアクセスされる情報や、計算コストの高い中間結果(例: 特定のクエリに対するベクトル検索結果)をキャッシュすることで、LLMの呼び出し回数を減らします。
  • 軽量なLLMの活用: 特定の判断ノードや簡単なタスクには、より高速で安価なLLM(例: GPT-3.5 Turbo)を使用し、複雑な推論が必要な場合にのみ高性能なLLM(例: GPT-4o)を利用するハイブリッド戦略を検討します。

3. 複雑なアーキテクチャと開発・運用負荷

ハマりどころ: Agentic RAGは、ノード、エッジ、状態管理、ツール連携など、従来のRAGよりもアーキテクチャが複雑になり、開発および運用における負荷が増大します。デバッグが困難になったり、変更の影響範囲が広くなったりする可能性があります。

回避策:

  • LangGraphなどのフレームワーク活用: LangGraphのようなグラフベースのオーケストレーションフレームワークを積極的に活用し、モジュール化された設計を心がけることで、複雑性を管理しやすくします。各ノードの役割を明確にし、単一責任の原則に従います。
  • 段階的な導入: 最初から完全なAgentic RAGを目指すのではなく、Classic RAGやHybrid RAGで基盤を固めた後、Corrective RAG、Self-RAG、Agentic RAGと段階的に導入する設計パターンが推奨されます。これにより、各段階で学習と改善を重ねることができます。
  • 詳細なモニタリングとロギング: エージェントの思考プロセスやアクション、各ノードの入出力、ツール呼び出しの詳細などを詳細にログに記録し、モニタリングすることで、問題発生時に迅速に原因を特定し、デバッグを容易にします。LangSmithのような可視化ツールとの連携も有効です。
  • 堅牢なテスト戦略: 各ノードの単体テスト、グラフ全体の統合テスト、エンドツーエンドのシナリオテストを導入し、変更がシステム全体に与える影響を最小限に抑えます。

設計上のトレードオフとベストプラクティス

このセクションでは、Agentic RAGシステムを設計する上で考慮すべきトレードオフと、開発を成功させるためのベストプラクティスをまとめます。

設計上のトレードオフ

Agentic RAGは強力ですが、万能ではありません。システムの要件に応じて、以下のトレードオフを考慮する必要があります。

  • 精度 vs コスト・レイテンシ:
    • 考慮: Agentic RAGは高精度な回答を期待できますが、LLM呼び出し回数が増えるため、コストとレイテンシが増大します。全てのクエリにAgentic RAGを適用するのではなく、クエリの複雑性や重要度に応じてRAGのパターンを使い分ける「Adaptive RAG」の導入を検討します。
  • 自律性 vs 制御性:
    • 考慮: エージェントの自律性を高めるほど、予期せぬ動作やコスト高騰のリスクが増します。エージェントの思考プロセスやアクションを詳細にログに記録し、モニタリングすることで、問題発生時に迅速に原因を特定し、制御できるようにします。必要に応じて人間による介入(Human-in-the-Loop)の仕組みを導入します。
  • 複雑性 vs 開発・運用負荷:
    • 考慮: Agentic RAGは従来のRAGよりもアーキテクチャが複雑になり、開発・運用負荷が増大します。LangGraphなどのフレームワークを活用し、モジュール化された設計を心がけ、CI/CDパイプラインに自動テストと評価を組み込み、運用負荷を軽減します。

ベストプラクティス

Agentic RAGを実用的なLLMアプリケーションとして成功させるために、以下のベストプラクティスを推奨します。

  • 段階的な導入: まずはClassic RAGやHybrid RAGで検索品質と評価基盤を固め、その後、Corrective RAG、Self-RAG、Agentic RAGを段階的に導入する設計パターンが推奨されます。これにより、リスクを管理しながら徐々に高度な機能を取り入れられます。
  • LLMの役割とプロンプト設計: LLMを単なる回答生成器ではなく、全体の意思決定を行う「頭脳役」として位置づけ、システムプロンプトやチェーンの設計で明確な指示を与え、エージェントとして振る舞う人格を持たせることが重要です。思考プロセスをステップバイステップで出力させることで、デバッグや理解が容易になります。
  • 多様なツール利用: ベクトル検索だけでなく、SQLクエリ、Web検索(Tavilyなど)、グラフDB探索、社内APIなど、複数のツールを使い分け、問いの性質に応じて検索方法を切り替えられるようにします。ツールは明確な目的と入出力を持つように設計します。
  • 自己評価(Reflection/Self-Critique): 検索結果やドラフト回答をLLM自身が評価し、情報が不十分な場合は再検索や回答の書き直しに進む自己改善ループを組み込みます。これにより、回答の信頼性と精度が向上します。
  • 状態管理とグラフ構造: LangGraphのようなグラフ構造を用いた状態管理により、エージェントの多段階処理や条件分岐、ループを柔軟に制御します。状態は最小限かつ必要な情報のみを保持するように設計し、可読性と保守性を高めます。
  • データ前処理と評価基盤: 徹底したデータ前処理(チャンキング、埋め込みなど)を行い、テストクエリセットを作成してデータやチャンキング戦略の変更のたびにA/Bテストで効果を定量的に検証する評価基盤を構築します。これにより、システムの改善効果を客観的に測定できます。
  • 定期的なデータ更新とメンテナンス: 検索対象データを常に最新かつ正確に保つためのプロセスを確立します。データの鮮度がRAGシステムの品質に直結するため、自動更新や手動レビューの仕組みを導入します。

まとめ

本記事では、複雑な推論やマルチステップな情報検索を可能にするAgentic RAGの概念と、その強力な実装フレームワークであるLangGraphを用いた具体的な構築方法を解説しました。

Agentic RAGは、従来のRAGの限界を超え、LLMが自律的に思考し、ツールを使いこなし、動的に情報を収集・評価する能力を提供します。これにより、より高度で実用的なLLMアプリケーションの実現が可能になります。

実装例を通じて、LangGraphのノード、エッジ、状態管理といった主要な概念を理解し、実際に動くAgentic RAGの骨格を構築できました。また、開発・運用における課題(ハルシネーション、コスト、複雑性)とその回避策、そして設計上のトレードオフとベストプラクティスについても深掘りしました。

LangGraphとAgentic RAGの組み合わせは、これからのLLMアプリケーション開発の主流となるでしょう。ぜひ本記事を参考に、あなたのアプリケーションにAgenticな能力を組み込んでみてください。さらに詳細な情報や応用例については、LangGraphの公式ドキュメントを参照することをお勧めします。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?