Agentic RAG - MCPサーバー、LangGraph とベクトル埋め込みで実現するインテリジェント RAG

Last updated at 2025-07-31Posted at 2025-07-30

この記事は？

このブログの内容は、ウェビナー【生成AI実用化の鍵　エンジニアのためのエージェントRAG／MCPとマルチモーダル】でご紹介させていただきました。アーカイブ動画もありますのでぜひお立ち寄りください。

はじめに - Agentic RAG とは？

従来のRAG（Retrieval-Augmented Generation- 検索拡張生成）システムは決められた1つの知識ベースから情報を検索し、コンテンツを生成するという単一の流れを辿ります。

しかし、実際のビジネスシーンでは、ユーザーの指示や質問の内容によって異なる専門知識や異なる処理方式が必要になることが多くあります。

大規模言語モデル（LLM）が静的な事前学習データに依存しているため企業独自のデータを活用できなかったり、情報が古くなったり、生成されるコンテンツが不正確になったりするという限界を抱える中、RAGはその解決策として登場し、リアルタイムのデータ検索を統合することで文脈に即したコンテンツ生成を可能にしてきました。しかし、従来のRAGシステムは静的なワークフローに制約され、多段階の推論や複雑なタスク管理に必要な適応性に欠けていました。

この限界を乗り越えるのが、Agentic RAG（エージェンティックRAG） です。Agentic RAGは、自律的なAIエージェントをRAGパイプラインに組み込むことで、動的な意思決定、反復的な推論、適応的な検索戦略を可能にします。これにより、エージェンティックRAGシステムは、従来のRAGでは困難だった柔軟性、スケーラビリティ、文脈認識能力を様々なアプリケーションで発揮できるようになります。

エージェンティックRAGは、この適応性を実現するために様々なエージェントワークフローパターンを活用します。これらのパターンは、LLMベースのアプリケーションにおいて、エージェントの振る舞いを最適化し、パフォーマンス、精度、効率を向上させるための構造化された手法です。サーベイ論文「Agentic Retrieval-Augmented Generation: A Survey on Agentic RAG」では以下のようなエージェンティック・ワークフローパターンを紹介しています。

プロンプト・チェイニング (Prompt Chaining)ー複雑なタスクを連続した複数のステップに分解し、各ステップが前のステップに基づいて進むことで精度を向上させます
- 例）段階的・階層的な文書作成ーまずアウトラインを生成し、その内容を検証した上で、完全なテキストを生成する
ルーティング (Routing)ー入力クエリを分類し、適切な専門のプロンプトやプロセスに誘導するパターンです。これにより、異なるクエリやタスクが個別に処理され、効率と応答品質が向上します
- 例）カスタマーサービスの問い合わせを、技術サポート、返金の要求、または一般的な問い合わせなどのカテゴリに振り分ける
- 例）技術サポートへの問い合わせを、障害対応、仕様に関する質問などのカテゴリに振り分ける
並列化 (Parallelization)ータスクを独立したプロセスに分割し、同時に実行することで、遅延を減らしスループットを向上させます。セクショニング（独立したサブタスク）と投票（複数の出力による精度向上）に分類されます
- セクショニンングの例）コンテンツモデレーションのようなタスクを分割し、1つのモデルが入力をスクリーニングし、もう1つのモデルがレスポンスを生成する
- 投票の例）複数のモデルを使用して、コードの脆弱性をクロスチェックする、コンテンツモデレーションの決定を分析する
オーケストレーター・ワーカーズ (Orchestrator-Workers)ー中央のオーケストレーターモデルが動的にタスクをサブタスクに分解し、専門のワーカーモデルに割り当てて結果を集約・統合するワークフローです。並列化とは異なり、入力の複雑さに応じてワークフローを適応的に変化させます
- 例）リクエストされた変更の性質に基づいて、コードベース内の複数のファイルを自動的に修正する
- 例）複数の情報源から関連する情報を収集・統合し、リアルタイムで調査を実施する
エバリュエーター・オプティマイザー (Evaluator-Optimizer)ー初期出力を生成し、評価モデルからのフィードバックに基づいてそれを反復的に改善するワークフローです
- 例）複数の評価と改善サイクルを通じて翻訳品質を向上させる
- 例）複数回にわたるリサーチクエリを実施し、追加の反復によって検索結果を洗練させる

冒頭に挙げた課題意識「質問の内容によって異なる専門知識や異なる処理方式が必要になる」に対し、これらのパターンの中で特に適合するのは、ルーティング (Routing): Directing Inputs to Specialized Processes のようです。

図はAgentic Retrieval-Augmented Generation: A Survey on Agentic RAG より引用

この動的なルーティング機能により、エージェンティックRAGシステムは、単一の固定ワークフローに依存する従来のRAGの制約を克服し、以下のような形で課題を解決できそうです。

多様なクエリタイプへの効率的な対応: 質問が分析処理（例：「2025年度に最も売り上げ金額が大きかった製品は何ですか？」）を必要とするか、文書検索（例：「製品XXXの特長は何ですか？」）を必要とするかをシステムが自動的に判断し、最適なプロセスへ誘導します
専門知識ベースの活用: 各クエリを専用のエージェントや知識ベースへ誘導することで、例えば「売上履歴」「生成AI関連技術」「魔法関連」といった個別の知識領域に特化した情報が、それぞれの知識ベースから最も関連性の高い正確な形で取得されます。これにより、単一の知識ソースに依存することによる検索精度の低下（ノイズの問題）を防ぎ、専門特化した知識ベースが互いにノイズとならず、より高い回答精度を実現します
応答品質と効率の向上: 各クエリがそのタイプに最適化されたプロセスで処理されるため、生成される回答の精度とタイムリーさが向上します

Agentic RAGは、このような適応的なアプローチによって「単一の流れ」という課題や「異なる専門知識、処理方式」へのニーズに根本的に対応し、次世代AIアプリケーションの基盤として期待されています。

そこで、本記事では、このルーティングパターンの応用として、LLM（大規模言語モデル）が事前学習で獲得した知識（"常識"）を使って質問を分析系と文書検索系に分類し、次に文書検索系の質問文と各知識ベースの平均ベクトルとのコサイン距離を計算することで、より精度の高いルーティングを実現する Agentic RAG を MCPと LangGraph の力を借りて実装してみた経験をご紹介しようと思います。

このアプローチは、最新の研究で提唱されている”Adaptive RAG”の考え方に基づいています。

従来のRAGとエージェンティックRAGの違い

従来のRAG

あらかじめ決められた1つの流れをなぞる（単一パス）
固定された知識ベースからの検索のみ
検索は一度だけ
単一のLLMモデルでの応答生成

従来のRAGの問題点

分析系（"2025年度に最も売り上げ金額が大きかった製品は何ですか？"）と文書検索系（"製品XXXの特長は何ですか？"）は、全く別々のシステムで扱う必要がありました。また、文書検索系では、ベクトルインデックスに多くのドキュメントを登録すると互いの情報がノイズとなって検索精度が下がってしまうという問題がありました。

エージェンティックRAG

RAG パイプラインの様々な場所でエージェンティックAIの考え方を活用することが可能ですが、今回は次のようなことを試してみました。

質問内容によって適切なエージェント（分析系エージェント、文書検索系エージェント）を自律選択
各エージェントが専門分野に特化した知識ベースを持つ
ルーターエージェントが最適なエージェントに処理を移譲

この記事では、技術スタックとしては、MCP（Model Context Protocol）対応のSQLclとLangGraphを組み合わせ、さらにベクトル埋め込みによる高精度なルーティングを実装したエージェンティックRAGシステムにチャレンジしてみました。

本実装の最大の特徴は、質問文をLLMによって分析系と文書検索系に一次分類し、次に文書検索系の質問文と各知識ベースの平均ベクトルとのコサイン距離を計算することで、より精度の高いルーティングを実現している二段階のアプローチである点です。

アーキテクチャの全体像

技術スタックと特徴

1. インデータベース・AIエージェント SELECT AI

データベースの検索と、その検索結果を元に回答を生成するエージェントとして Oracle Database 23ai の SELECT AI を使っています。

SELECT AI が、分析系として機能するか文書検索系として機能する（SELECT AI with RAG）かは、セッションに設定される AI Profile で決まります。AI Profile にベクトルインデックスが指定されていると文書検索系（SELECT AI with RAG）として機能します。

今回は、設定も簡単ですぐに使える手軽さなどから SELECT AI を使っていますが、別の手段で検索と回答生成を担うエージェントを用意すればこの記事と同様の考え方で Agentic RAG を実現できます。

2. 知識ベース

データベースには、Oracle Autonomous Databaseを使っています
売上履歴知識ベースは、Autonomous Database にプリインストールされているサンプルスキーマ"Sales History"を使っています。Products、Sales、Customersなどの9つのテーブルが用意されています
生成AI関連知識ベースと魔法関連知識ベースは、DBMS_CLOUD_AI.CREATE_VECTOR_INDEX プロシージャで構築したベクトルインデックスを使用します
生成AI関連知識ベースには、生成AI関連の Wikipedia の記事や Oracle Database AI Vector Search 関連のドキュメント、私の生成AI関連のブログを登録してあります
魔法関連知識ベースには、ハリー・ポッターや魔法少女アニメ関連の Wikipedia の記事を登録してあります
これらの知識ベースを検索して回答を生成するためには、各知識ベースに対応した AI Profile をセッションに設定した上で、SELECT AI narrate を使います
売上知識ベースに対して、分析系の質問（例えば、「最も売上金額が大きかった製品は何ですか？」）をする場合は、EXEC DBMS_CLOUD_AI.SET_PROFILE('SALES_HISTORY_PROFILE')でAI Profileを設定してから、SELECT AI narrate 最も売上金額が大きかった製品は何ですか？を発行します
生成AI関連知識ベースに対して、文書検索系の質問（例えば、「LLMが苦手なことは何ですか？」）をする場合は、EXEC DBMS_CLOUD_AI.SET_PROFILE('GENERATIVE_AI_RAG_INDEX')でAI Profileを設定してから、SELECT AI narrate LLMが苦手なことは何ですか？を発行します
魔法AI関連知識ベースに対して、文書検索系の質問（例えば、「魔性少女まどか☆マギカの主人公は誰？」）をする場合は、EXEC DBMS_CLOUD_AI.SET_PROFILE('MAGIC_RAG_VEC_INDEX')でAI Profileを設定してから、SELECT AI narrate 魔性少女まどか☆マギカの主人公は誰？を発行します

インデータベース AIエージェント＝ SELECT AI による知識ベースへの問い合わせまでの流れ

なお、分析系については、ベクトルインデックスの作成は不要です。

知識ベースの AI Profile とベクトルインデックスの定義例

AI Profile

売上履歴知識ベース

BEGIN
DBMS_CLOUD_AI.CREATE_PROFILE(
  profile_name =>'SALES_HISTORY_PROFILE',
  attributes   =>'{"provider": "oci",
    "credential_name": "OCI_CREDENTIAL",
    "model": "meta.llama-4-scout-17b-16e-instruct",
    "max_tokens": 4000,
    "object_list": [
      {"owner": "SH"}
    ]
  }',
  description => '売上履歴データベースを使って自然言語からSQLを生成したり、そのSQLを実行したり、SQLを実行した結果に基づいてAIを使って質問に対する回答を生成するプロファイル。SQLの生成は、SELECT AI showsql 質問文　の形式。SQLの生成とその実行は、SELECT AI runsql 質問文　の形式。SQLの実行結果に基づいて回答を生成する場合は、SELECT AI narrate 質問文　の形式。質問文は自然文です。'
);
END;

売上履歴は、ベクトルインデックスを使用しませんので事前に知識ベースのRAGパイプラインの構築は不要です。従って、このプロファイルは、SELECT AI実行時にのみ使用されます。

生成AI関連知識ベース

BEGIN
DBMS_CLOUD_AI.CREATE_PROFILE(
  profile_name =>'GENERATIVE_AI_RAG_PROFILE',
  attributes   =>'{"provider": "oci",
    "credential_name": "OCI_CREDENTIAL",
    "embedding_model": "cohere.embed-v4.0",
    "model": "meta.llama-4-scout-17b-16e-instruct",
    "max_tokens": 4000,
    "vector_index_name": "GENERATIVE_AI_RAG_INDEX"
  }',
  description => '生成AI、RAG、Oracle Database の Select AI や DBMS_CLOUD_AI パッケージ（SQLの生成、実行、説明、大規模言語モデルとのチャット、RAG（検索拡張生成）などのタスク）やOracle AI Vector Searchに関する質問に答えるRAGのプロファイル'
);
END;

魔法関連知識ベース

BEGIN
DBMS_CLOUD_AI.CREATE_PROFILE(
  profile_name =>'MAGIC_RAG_PROFILE',
  attributes   =>'{"provider": "oci",
    "credential_name": "OCI_CREDENTIAL",
    "embedding_model": "cohere.embed-v4.0",
    "model": "meta.llama-4-scout-17b-16e-instruct",
    "max_tokens": 4000,
    "vector_index_name": "MAGIC_RAG_VEC_INDEX"    
  }',
  description => '魔法に関する質問に答えるRAGのプロファイル　。SQLの確認は、SELECT AI showsql 質問文　の形式。SQLを実行して検索結果を得るには、SELECT AI runsql 質問文　の形式。SQLを実行してその実行結果に基づいて回答を生成する場合は、SELECT AI narrate 質問文　の形式。質問文は自然文です'
);
END;

ベクトルインデックス

```sql:生成AI関連知識ベース・ベクトルインデックス DBMS_CLOUD_AI.CREATE_VECTOR_INDEX( index_name => 'GENERATIVE_AI_RAG_INDEX', attributes => '{"vector_db_provider": "oracle", "location": "https://orasejapan.objectstorage.ap-tokyo-1.oci.customer-oci.com/n/xxxxxxxxxx/b/genai-bucket/o/", "object_storage_credential_name": "OCI_CREDENTIAL", "profile_name": "GENERATIVE_AI_RAG_PROFILE", "vector_dimension": 1536, "vector_distance_metric": "cosine", "chunk_overlap":25, "chunk_size":250, "match_limit":10, "similarity_threshold":0.3, "refresh_rate":1}', description => '生成AI、Oracle Database の Select AI、 DBMS_CLOUD_AI パッケージなどのAIやベクトル検索やRAGの質問に答えるRAGのベクトルインデックス' ); END; ``` ```sql:魔法関連知識ベース・ベクトルインデックス BEGIN DBMS_CLOUD_AI.CREATE_VECTOR_INDEX( index_name => 'MAGIC_RAG_VEC_INDEX', attributes => '{"vector_db_provider": "oracle", "location": "https://orasejapan.objectstorage.ap-tokyo-1.oci.customer-oci.com/n/xxxxxxxxxx/b/mahou-bucket/o/", "object_storage_credential_name": "OCI_CREDENTIAL", "profile_name": "MAGIC_RAG_PROFILE", "vector_dimension": 1536, "vector_distance_metric": "cosine", "chunk_overlap":25, "chunk_size":250, "match_limit":10, "similarity_threshold":0.3, "refresh_rate":1}' ); END; ```

インデータベース AIエージェント＝ SELECT AI による知識ベースへの問い合わせ

3. MCP対応SQLcl

Oracle Database のコマンドラインインターフェース (CLI) の Oracle SQLcl が MCPサーバーとして動作するようになりました。SQLcl は、sql コマンドに -mcp オプションを付けて起動すると STDIO トランスポートで MCP クライアントと通信する MCP サーバーになり、データベース操作をツールとして公開します。

上図は、Introducing MCP Server for Oracle Databaseより引用

利用可能なツールは、5つあります。

ツール名	概要
list-connections	保存されているデータベース接続の一覧を取得します
connect	指定された接続を使ってデータベースへ接続します
disconnect	データベース接続を切断します
run-sqlcl	SQLclコマンドを実行します
run-sql	SQLクエリを実行します

Cluade Code、Cursor などの AIアプリケーションにおける典型的な設定例

（パスの表記は Windowsの例）

設定ファイル例

{
  "mcpServers": {
    "sqlcl": {
      "command": "<SQLclのパス>\\sql.exe",
      "args": ["-mcp"]
    }
  }
}

MCP クライアント機能の実装

Python + LangChain で MCP クライアント機能を実装する際に SQLcl を MCPサーバーとして使用する例です。
アプリケーション側の MCP のコードはこれですべてです！

STDIO トランスポートを使用するMCPクライアント機能の実装例

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

server_params = StdioServerParameters(
    command=r"<SQLclのパス>\\sql.exe",
    args=["-mcp"],
)
    
async with stdio_client(server_params) as (read, write):
    # MCPクライアントとMCPサーバーとのセッションを確立
    async with ClientSession(read, write) as session:
        # MCPサーバーを初期化
        await session.initialize()
        # MCPサーバーからツールを取得
        tools = await load_mcp_tools(session)

stdio_client()は、StdioServerParametersの commandに指定された MCPサーバーを子プロセスとして起動して、argsで指定したパラメータを渡します
ClientSession()は、stdio_client()が生成したMCPクライアントとサーバーのセッションを管理します
load_mcp_tools()は、すべての利用可能なMCPツールをロードして、LangChainツールへ変換します

4. エージェントのフレームワーク（LangGraph + LangChain）

エージェントの実装にLangGraphのcreate_react_agentを使用し、各エージェントが自律的に判断・実行できる仕組みを構築しています。

LangGraphの Prebuilt ReAct Agent「create_react_agent」は、ReAct（Reasoning＋Acting）の考え方に基づいたエージェントを非常に簡単に構築できる関数です。ReActは、大規模言語モデル（LLM）が推論とアクション（外部ツールの呼び出し）を組み合わせた対話を行う設計思想です。

ReAct（Reasoning and Acting）**とは、AIが「考える（推論：Reasoning）」と「行動する（Acting）」を繰り返すことで、複雑なタスクや問題解決を人間のように柔軟に進めるための仕組みです。

従来のAIアプリケーションは指示に一度だけ答える「一問一答型」として構築される例が多かったかと思います。それに対して、ReActでは次のようなサイクルを繰り返します。

課題や質問について最初に**思考（推論）**します
思考（推論）に基づいて**実際の行動（外部ツール実行や検索など）**を行います
行動の結果を観察・確認して、再度考ます

この「考える→行動→結果を見てまた考える」のループにより、AIエージェントは途中で方針を修正したり追加情報を取得するなど、段階的に最適化された答えを目指します

図は、LangGraph Prebuild Agentより引用

LLMからの応答にツールの起動指示がなくなるまで繰り返します（繰り返し回数のデフォルト上限は10回）。

ReAct エージェントの基本的なコード例（LLMとして OpenAI モデルを使った場合）

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(
    model="gpt-4o", # o4-mini-2025-04-16, gpt-4o, gpt-4.1-mini-2025-04-14 e.t.c.
    api_key="<OPENAI_API_KEY>",
)

agent = create_react_agent(llm, tools, prompt)
agent_response = await agent.ainvoke({"messages": [{"role": "user", "content": "<メッセージ>"}]})

OCI Generative AI サービスの Cohere モデルを使った例

ReAct エージェントの基本的なコード例（OCI Generative AI サービスの Cohere モデルを使った例)

from langchain_community.chat_models.oci_generative_ai import ChatOCIGenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOCIGenAI(
    model_id="cohere.command-a-03-2025", # cohere.command-r-plus-08-2024 e.t.c.
    service_endpoint="<OCI Generative AI エンドポイント>",
    compartment_id="<OCI コンパートメント ID>"
)

agent = create_react_agent(llm, tools, prompt)
agent_response = await agent.ainvoke({"messages": [{"role": "user", "content": "<メッセージ>"}]})

5. 質問文を分析/文書検索系へ分類する手法

今回は、LLMにプロンプトでいくつかの分類例を与えてLLMに分類させるシンプルな方法を取っています（Few-Shot Prompting）。

分析の例	文書検索の例
2024年に売上が最も多かった製品は？	海外出張の際の申請手続きについて教えてください？
全販売地域の中で、売上が最も多かった地域は？	Autonomous Databaseのデータベースのバックアップについて教えてください？
売上が下落傾向にある製品は？	魔女っ子という言葉が使われなくなり魔法少女と呼ばれるようになったのはなぜですか？

分類実行結果例

質問文を質問タイプ（分析/検索）に分類。検索は文書検索の意。

=== 質問 1: ======================================
質問文: 「売上が最も多かった製品は何で、その売上はいくらですか？」
質問タイプ: 分析

=== 質問 2: ======================================
質問文: 「地域別の売上ランキングを教えてください。」
質問タイプ: 分析

=== 質問 3: ======================================
質問文: 「2001年に最も売上高が大きかった地域はどこで、その地域で最も売れた製品カテゴリーは何でしたか？」
質問タイプ: 分析

=== 質問 4: ======================================
質問文: 「販売に最も貢献しているチャネルは？そのチャネルの売上はいくらですか？」
質問タイプ: 分析

=== 質問 5: ======================================
質問文: 「各製品カテゴリーで最も売上が多かった製品は？」
質問タイプ: 分析

=== 質問 6: ======================================
質問文: 「年度ごとの総売上高の推移を教えて？」
質問タイプ: 分析

=== 質問 7: ======================================
質問文: 「大規模言語モデルとは何ですか？」
質問タイプ: 検索

=== 質問 8: ======================================
質問文: 「LLMの主なアーキテクチャは何ですか？」
質問タイプ: 検索

=== 質問 9: ======================================
質問文: 「LLMはどのようなデータで訓練されるのですか？」
質問タイプ: 検索

=== 質問 10: ======================================
質問文: 「ファインチューニングとは何ですか？」
質問タイプ: 検索

=== 質問 11: ======================================
質問文: 「プロンプトエンジニアリングとは何ですか？」
質問タイプ: 検索

=== 質問 12: ======================================
質問文: 「LLMの創発的能力とは何ですか？」
質問タイプ: 検索

=== 質問 13: ======================================
質問文: 「LLMにはどのような課題や限界がありますか？」
質問タイプ: 検索

=== 質問 14: ======================================
質問文: 「単語の埋め込みとは何ですか？」
質問タイプ: 検索

=== 質問 15: ======================================
質問文: 「対照学習とは何ですか？」
質問タイプ: 検索

=== 質問 16: ======================================
質問文: 「単語埋め込みから文埋め込みを生成するにはどのような方法が使われていますか？」
質問タイプ: 検索

=== 質問 17: ======================================
質問文: 「OpenAIのCLIPは、どのように学習したのですか？」
質問タイプ: 検索

=== 質問 18: ======================================
質問文: 「ハリー・ポッターに彼が魔法使いであることを告げたのは誰ですか？それはいつのことでしたか？」
質問タイプ: 検索

=== 質問 19: ======================================
質問文: 「日本の次期大統領は誰ですか？」
質問タイプ: 検索

=== 質問 20: ======================================
質問文: 「宇宙で一番大きな惑星は？」
質問タイプ: 検索

=== 質問 21: ======================================
質問文: 「宇宙全体の元素組成を原子の存在比と総質量で教えてください」
質問タイプ: 検索

=== 質問 22: ======================================
質問文: 「チャンキングと検索件数と精度の関係を教えて？」
質問タイプ: 検索

=== 質問 23: ======================================
質問文: 「画像生成のCFGって何ですか？」
質問タイプ: 検索

=== 質問 24: ======================================
質問文: 「リモートセンシングや医療画像などの解析に用いられるRAGはどんなものですか？」
質問タイプ: 検索

=== 質問 25: ======================================
質問文: 「ショート動画レコメンデーションのコールドスタート問題にはどのような対処が可能ですか？」
質問タイプ: 検索

=== 質問 26: ======================================
質問文: 「環いろはとは誰ですか？」
質問タイプ: 検索

=== 質問 27: ======================================
質問文: 「神崎メグの呪文は？」
質問タイプ: 検索

=== 質問 28: ======================================
質問文: 「テクニク・テクニカ・シャランラーとは？」
質問タイプ: 検索

=== 質問 29: ======================================
質問文: 「錬金術とは何ですか？」
質問タイプ: 検索

=== 質問 30: ======================================
質問文: 「化学とは何ですか？」
質問タイプ: 検索

=== 質問 31: ======================================
質問文: 「化学と錬金術の違いは？」
質問タイプ: 検索

=== 質問 32: ======================================
質問文: 「化学と物理学の違いは？」
質問タイプ: 検索

=== 質問 33: ======================================
質問文: 「イーサン・ハントさんの職業は？」
質問タイプ: 検索

=== 質問 34: ======================================
質問文: 「暁美ほむらの固有魔法はなんですか？」
質問タイプ: 検索

=== 質問 35: ======================================
質問文: 「フェイトが留学していた小学校は？」
質問タイプ: 検索
=== 質問 33: ======================================
質問文: 「イーサン・ハントさんの職業は？」
質問タイプ: 検索

=== 質問 34: ======================================
質問文: 「暁美ほむらの固有魔法はなんですか？」
質問タイプ: 検索

=== 質問 35: ======================================
質問文: 「フェイトが留学していた小学校は？」
質問タイプ: 検索

質問文: 「イーサン・ハントさんの職業は？」
質問タイプ: 検索

=== 質問 34: ======================================
質問文: 「暁美ほむらの固有魔法はなんですか？」
質問タイプ: 検索

=== 質問 35: ======================================
質問文: 「フェイトが留学していた小学校は？」
質問タイプ: 検索

=== 質問 36: ======================================
質問文: 「アースラの艦長は誰？」
質問タイプ: 検索
質問タイプ: 検索

=== 質問 35: ======================================
質問文: 「フェイトが留学していた小学校は？」
質問タイプ: 検索

=== 質問 36: ======================================
質問文: 「アースラの艦長は誰？」
質問タイプ: 検索
質問タイプ: 検索


=== 質問 37: ======================================
質問文: 「なのはの魔法の杖の名前は？」
質問タイプ: 検索

質問タイプ: 検索

=== 質問 37: ======================================
質問文: 「なのはの魔法の杖の名前は？」
質問タイプ: 検索


=== 質問 37: ======================================
質問文: 「なのはの魔法の杖の名前は？」
質問タイプ: 検索

=== 質問 38: ======================================
質問文: 「管理局の空戦魔導士には誰がいますか？」
質問タイプ: 検索

=== 質問 39: ======================================
質問文: 「なのはの最大最強の武器は？」
質問タイプ: 検索

=== 質問 40: ======================================
質問文: 「熱力学の第二法則に縛られないエネルギー源は？それを搾取しているのは誰ですか？」
質問タイプ: 検索

=== 質問 41: ======================================
質問文: 「フリーレンの年齢は？」
質問タイプ: 検索

=== 質問 42: ======================================
質問文: 「フリーレンのパーティーのメンバーは？」
質問タイプ: 検索

=== 質問 43: ======================================
質問文: 「ユフィリアの魔道具の名前は？」
質問タイプ: 検索

=== 質問 44: ======================================
質問文: 「侯爵令嬢だったユフィーが後に女王となった経緯を教えて？」
質問タイプ: 検索

=== 質問 45: ======================================
質問文: 「ホグワーツ特急にはどこで乗れますか？」
質問タイプ: 検索

=== 質問 46: ======================================
質問文: 「木之元桜が鍵を杖に変えるときに唱える呪文は？」
質問タイプ: 検索

Tips
Few-Shot Prompting は、"few"という言葉から例は2,3個で十分、2,3個に留めるべきものという誤解がたまにありますが、数百例まで増やしても精度が上がり続けることが知られています。
最初に GPT-3.5 で Few-Shot を提案した論文でも 100例までテストしても精度が上がり続けることが報告されています。さらに、後の研究では 500例まで増やしても精度が上がり続けるという報告もあります。

Language Models are Few-Shot Learners より引用

6. ベクトル埋め込みによる質問文の知識ベースへの高精度分類

質問文がどの知識ベースへ問い合わせるべき質問であるかを分類するためには、各知識ベースの（チャンクの）平均ベクトルを事前に計算しておいて、質問文とのコサイン距離を使って最適な知識ベースを選択方式を取ります。LLMにLLMが事前学習した"常識"に基づいて判定させるよりも実際にデータベースに格納されているドキュメント群の特徴を反映した分類ができますので、より高い分類精度が期待できます。

埋め込みモデル

ベクトル埋め込みを生成する埋め込みモデルには、Cohere Embed 4 を使用します。
Cohere Embed 4は、コサインで事前学習していますので、今回はベクトル間の距離の指標としてコサインを使用します。なお、Cohere Embed 4は、ベクトルの長さを正規化していますので、内積等を使用してもランキング（ベクトルが似ている順位）は同じになります。

知識ベースの平均ベクトルの計算

今回は、データベースのテーブル上に格納されている知識ベースのチャンク全体の平均をSQLの集計関数 AVG を使用してデータベース内で集計します。大量のベクトル埋め込みをデータベースからローカルにダウンロードするのを避けるためです（ベクトルの次元が1536次元であるため、各チャンクあたり浮動小数点数が1536個あるため転送データ量が大きいためです）。データベースには、Oracle Autonomous Database（23ai）を使用しています。

計算した平均ベクトルは、データベース上に平均ベクトル格納用のテーブルを作成して格納します。

質問文と知識ベースの平均ベクトルのコサイン距離の計算

各知識ベースの平均ベクトルと質問文のベクトル埋め込みのコサイン距離はSQLのVECTOR_DISTANCE関数で計算します。

実装の詳細

1. 知識ベースの平均ベクトル計算

各知識ベース（AI Profile）に含まれる文書のベクトル埋め込みの平均を事前に計算し、AI_PROFILE_VECTOR_SUMMARYテーブルに格納しておきます。

各知識ベースに登録されているドキュメントの平均ベクトル埋め込みを格納するテーブル

AI_PROFILE_VECTOR_SUMMARY

CREATE TABLE AI_PROFILE_VECTOR_SUMMARY (
    AI_PROFILE_NAME VARCHAR2(100),
    VECTOR_INDEX_NAME VARCHAR2(100),
    VECTOR_TABLE_NAME VARCHAR2(100),
    AVERAGE_VECTOR VECTOR(1536, FLOAT32)
);

VECTOR_INDEX_NAME と VECTOR_TABLE_NAME は、今回使いませんでしたので無くて大丈夫です。
また、このテーブルには今回２件しかデータを登録しないため索引も作成していません。

各知識ベースに登録されているドキュメントの平均ベクトル埋め込みを計算してAI_PROFILE_VECTOR_SUMMARY テーブルに登録する SQL

平均ベクトル埋め込み登録 SQL

INSERT INTO AI_PROFILE_VECTOR_SUMMARY (
    AI_PROFILE_NAME, 
    VECTOR_INDEX_NAME, 
    VECTOR_TABLE_NAME, 
    AVERAGE_VECTOR
)
SELECT 
    'MAGIC_RAG_PROFILE' as AI_PROFILE_NAME,
    'MAGIC_RAG_VEC_INDEX' as VECTOR_INDEX_NAME,
    'MAGIC_RAG_VEC_INDEX$VECTAB' as VECTOR_TABLE_NAME,
    AVG(EMBEDDING) as AVERAGE_VECTOR
FROM MAGIC_RAG_VEC_INDEX$VECTAB;

SQLの集計関数 AVG を使用してデータベース内で平均ベクトルを計算して、AI_PROFILE_VECTOR_SUMMARY テーブルに格納しています。
この平均ベクトルが、各知識ベース（AI Profile）の「特徴ベクトル」として機能します。

2. 質問文との距離計算

質問文がどの知識ベースへ問い合わせるべき質問であるかは、質問文がどの知識ベースのドキュメント群と似ているか、関連があるかで判定できると仮定し、似ているか、関連があるかは「質問文のベクトル埋め込み」と「知識ベースのドキュメント群の平均ベクトル埋め込み」の距離（コサイン距離）で判定することにします。この計算は、データベースの中で PL/SQL で実行することにします。なお、これはベクトル検索そのもので、2件全件を取得して距離が近い順にソートしています。ここでは、PL/SQL のファンクションで実現していますが、アプリケーションから MCPサーバー経由などでベクトル検索を行っても同じ結果が得られます。

get_distances_from_profilesファンクションで、質問文のベクトル埋め込みと各知識ベースの平均ベクトルとのコサイン距離を計算します。

get_distances_from_profiles（質問文のベクトル埋め込みと各知識ベースの平均ベクトルとのコサイン距離を計算するファンクション）

-- 知識ベース（AI Profile）の名前と、質問文と知識ベースの平均ベクトルの
-- コサイン距離を格納するオブジェクト型を定義
CREATE OR REPLACE TYPE profile_result AS OBJECT (
    profile_name VARCHAR2(100),
    distance_score NUMBER
);
/

-- profile_result オブジェクトのコレクション型（Nested Table）を定義
-- 複数の知識ベース（AI Profile）の検索結果を格納するためのコレクション
-- ファンクションの戻り値となる
CREATE OR REPLACE TYPE profile_result_table AS TABLE OF profile_result;

-- get_distances_from_profiles（コサイン距離計算ファンクションの本体）
CREATE OR REPLACE FUNCTION get_distances_from_profiles(
    search_text IN VARCHAR2
) RETURN profile_result_table
IS
    params CLOB := '{"provider":"ocigenai",...,"model":"cohere.embed-v4.0"}';
    result_table profile_result_table := profile_result_table();
BEGIN
    WITH query_vec AS (
        SELECT DBMS_VECTOR.UTL_TO_EMBEDDING(search_text, json(params)) AS vec
        FROM dual
    )
    SELECT profile_result(
        sd.AI_PROFILE_NAME,
        VECTOR_DISTANCE(sd.AVERAGE_VECTOR, q.vec, COSINE)
    )
    BULK COLLECT INTO result_table
    FROM AI_PROFILE_VECTOR_SUMMARY sd, query_vec q
    ORDER BY VECTOR_DISTANCE(sd.AVERAGE_VECTOR, q.vec, COSINE) ASC
    FETCH FIRST 2 ROWS ONLY;
    
    RETURN result_table;
END;

UTL_TO_EMBEDDING関数で質問文をベクトルに変換しています
VECTOR_DISTANCE関数で質問文のベクトルと各知識ベースの平均ベクトルのコサイン距離を計算しています

MCP を使って試してみる

"プロンプトを変えると振舞いが変わるのは？" という質問が、生成AI系（GENERATIVE_AI_RAG_PROFILE）と魔法系（MAGIC_RAG_PROFILE）のどちらの知識ベースへ問い合わせるべき質問と分類されるのか確かめてみましょう。

SQLcl の Oracle Database MCPサーバーへ接続した Claude Desktop から

SELECT * FROM TABLE(get_distances_from_profiles('プロンプトを変えると振舞いが変わるのは？'));

というSQLで get_distances_from_profilesファンクションを呼び出します。

"プロンプトを変えると振舞いが変わるのは？"は、生成AI系と判定されました。
期待どおりに判定されているようです。

次に、**"呪文を唱えると姿が変わるのは？"**という質問についても分類してみます。

"呪文を唱えると姿が変わるのは？"は、魔法系と判定されました。
こちらも期待どおりに判定されているようです。

両方のスクリーンショットのクエリーの結果表の後に続く分析を見ていただくと私が説明したかったことを Claude が説明してくれています。（私の出番がない....）

"高度に発達したテクノロジーは魔法と区別がつかない" というアーサー・C・クラーク（だったと思います）の名言がありますが、見事に区別できています。

3. Agentic RAG 本体の実装

Agentic RAG コード全体

AgenticRAG.py

from langchain.globals import set_debug, set_verbose

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
import numpy as np
import re
import json

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent

from langchain_openai import ChatOpenAI
from langchain_community.chat_models.oci_generative_ai import ChatOCIGenAI
from langchain_core.language_models.chat_models import BaseChatModel

import os
from dotenv import load_dotenv, find_dotenv
from typing import Dict, Any, Optional, List
from langchain.tools import Tool

# 環境変数のロード
load_dotenv(find_dotenv())
VERBOSE = os.getenv("VERBOSE", "").lower() == "true"

# LLMの設定
"""
llm = ChatOpenAI(
    model="gpt-4o", # o4-mini-2025-04-16, gpt-4o # gpt-4.1-mini-2025-04-14
    api_key=os.getenv("OPENAI_API_KEY"),
    #temperature=0.0,
    max_tokens=4000,
    seed=1234,
    streaming=False
)
"""
llm = ChatOCIGenAI(
    model_id="cohere.command-a-03-2025", # cohere.command-r-plus-08-2024 e.t.c.
    service_endpoint=os.getenv("OCI_GENAI_ENDPOINT"),
    compartment_id=os.getenv("OCI_COMPARTMENT_ID"),
    auth_type="API_KEY",
    auth_file_location=os.getenv("OCI_CONFIG_FILE_LOCATION"),
    auth_profile=os.getenv("OCI_CONFIG_PROFILE"),
    model_kwargs={"temperature": 0, "max_tokens": 4000, "seed": 1234, "is_stream": False}
)


server_params = StdioServerParameters(
    command=r"D:\\tools\\sqlcl\\bin\\sql.exe",
    args=["-mcp"],
)

async def ConnectDatabaseAgent(connection_name: str, tools: List[Tool], llm: BaseChatModel) -> Dict[str, Any]:
    """
    データベースへ接続する
    
    Args:
        connection_name: 接続名
        tools: MCPサーバーから取得したツール
        llm: 言語モデル
    
    Returns:
        Dict[str, Any]: {
            "response": データベースへの接続結果,
            "has_error": エラーの有無,
            "error_message": エラーの内容（エラーがない場合はNone）
        }
    """
    result = {
            "response": None,
            "has_error": False,
            "error_message": None
        }
    PROMPT_FOR_DB_CONNECTION = """
    # あなたは、データベースに接続するAIエージェントです。
    # データベースへ接続するツールを利用できる場合は、それを使って指定された接続先へ接続してください。
    # データベースへ接続するツールを利用できない場合は、既にデータベース接続が確立されていると判断してください。
    # 以下のフォーマットで応答してください。
    ## データベースへ接続した場合:
    ### フォーマット
    [データベースへ接続しました][接続名][接続先のデータベース名]
    ### 例
    [データベースへ接続しました][ADB - adbuser][ADB001]
    ## 接続するツールを利用できない場合:
    ### フォーマット
    [接続ツールが利用できないため既に接続が確立していると仮定します]
    ### 例
    [接続ツールが利用できないため既に接続が確立していると仮定します]
    """
    try:
        agent = create_react_agent(llm, tools, prompt=PROMPT_FOR_DB_CONNECTION)
        agent_response = await agent.ainvoke({"messages": [{"role": "user", "content": "接続先：" + connection_name}]})
        response = agent_response['messages'][-1].content
        result["response"] = response
                
    except Exception as e:
        result["has_error"] = True
        result["error_message"] = str(e)
    
    return result

async def ClassifyQueryTypeAgent(question: str, llm: BaseChatModel) -> Dict[str, Any]:
    """
    質問文をAnalytics系か文書検索系に分類する
    
    Args:
        question: 分類する質問文
        llm: 言語モデル
        
    Returns:
        dict: 結果を含む辞書
    """
    result = {
        "question": question,
        "question_type": None,
        "has_error": False,
        "error_message": None
    }

    PROMPT_FOR_QUERY_TYPE_CLASSIFIRE = """
    # あなたは、ユーザーが提供する質問文が以下の2つのカテゴリーのいずれに属するかを分類する AIエージェントです。
    # 質問に答えることは禁止します。あくまでも、質問文の分類だけを行います。必ず１つの カテゴリー を選択してください。
    # 適切な カテゴリーがない場合にもどちらかと言えばあてはまるカテゴリーを選択してください。
    # カテゴリー：
    ## カテゴリー1：検索
    - これは、文書に対するベクトル検索や全文検索に対する質問文です。
    ### カテゴリー1の質問文の例：
    - 海外出張の際の申請手続きについて教えてください？
    - Autonomous Databaseのデータベースのバックアップについて教えてください？
    - 魔女っ子という言葉が使われなくなり魔法少女と呼ばれるようになったのはなぜですか？
    ## カテゴリー2：分析
    - これは、RDBに格納されている構造化データに対する分析やデータ分析、集計に対する質問文です。
    ### カテゴリー2の質問文の例：
    - 2024年に売上が最も多かった製品は？
    - 全販売地域の中で、売上が最も多かった地域は？
    - 売上が下落傾向にある製品は？

    回答は、必ず以下のJSON形式で返答してください。JSON以外のテキストは含めないでください：
    ```json
    {
        "categories": ["検索", "分析"],
        "selected_category": "選択したカテゴリー"
    }
    ```

    エラーが発生した場合には、以下のJSON形式で返答してください：
    ```json
    {
        "error": "エラーの内容"
    }
    ```

    例1（検索を選択した場合）：
    ```json
    {
        "categories": ["検索", "分析"],
        "selected_category": "検索"
    }
    ```

    例2（分析を選択した場合）：
    ```json
    {
        "categories": ["検索", "分析"],
        "selected_category": "分析"
    }
    ```

    例3（エラーの場合）：
    ```json
    {
        "error": "分類処理中にエラーが発生しました"
    }
    ```
    """
    
    try:
        # 分類エージェントを作成（ツールなしで動作）
        agent_classifier = create_react_agent(llm, tools=[], prompt=PROMPT_FOR_QUERY_TYPE_CLASSIFIRE)
        
        # エージェントを実行
        agent_response = await agent_classifier.ainvoke({"messages": [{"role": "user", "content": "質問文：" + question}]})
        response_text = agent_response['messages'][-1].content
        
        # JSON応答をパース
        data = parse_json_response(response_text)
        
        if data is None:
            result["has_error"] = True
            result["error_message"] = "JSON応答の解析に失敗しました"
            return result
        
        # エラーチェック
        if "error" in data:
            result["has_error"] = True
            result["error_message"] = data["error"]
            return result
        
        # 分類結果を取得
        if "selected_category" in data:
            result["question_type"] = data["selected_category"]
        else:
            result["has_error"] = True
            result["error_message"] = "分類結果が見つかりません"
            
    except Exception as e:
        result["has_error"] = True
        result["error_message"] = str(e)
    
    return result

def softmax_with_temperature(distances, temperature=10.0):
    """
    温度パラメータ付きsoftmax関数で確信度を計算
    コサイン距離は小さいほど類似しているため、負の値にしてからsoftmaxを適用
    
    Args:
        distances: コサイン距離のリスト
        temperature: 温度パラメータ（大きいほど差が拡大される、デフォルト10.0）
    """
    # 距離を負の値にして、小さい距離ほど高い確率になるようにする
    neg_distances = -np.array(distances)
    # 温度パラメータで差を拡大
    scaled_distances = neg_distances * temperature
    exp_values = np.exp(scaled_distances - np.max(scaled_distances))  # オーバーフロー対策
    probabilities = exp_values / np.sum(exp_values)
    return probabilities * 100  # パーセンテージに変換

def parse_json_response(response_text):
    """
    エージェントの応答からJSONを抽出してパース
    """
    try:
        # JSON部分を抽出（```json と ``` の間のテキスト）
        json_match = re.search(r'```json\s*(.*?)\s*```', response_text, re.DOTALL)
        if json_match:
            json_str = json_match.group(1)
        else:
            # JSONブロックが見つからない場合は全体をJSONとして扱う
            json_str = response_text.strip()
        
        # まず通常のJSON解析を試す
        try:
            data = json.loads(json_str)
            return data
        except json.JSONDecodeError:
            # 失敗した場合、段階的にエスケープ処理を実行
            print(f"初回JSON解析失敗、エスケープ処理を実行中...")
            
            # 1. 基本的なエスケープ処理
            json_str_fixed = fix_basic_escaping(json_str)
            
            try:
                data = json.loads(json_str_fixed)
                return data
            except json.JSONDecodeError:
                # 2. より詳細なエスケープ処理
                json_str_fixed = fix_advanced_escaping(json_str)
                data = json.loads(json_str_fixed)
                return data
            
    except json.JSONDecodeError as e:
        print(f"JSON解析エラー: {e}")
        print(f"問題のあるJSON文字列の先頭200文字: {json_str[:200]}...")
        return None
    except Exception as e:
        print(f"予期しないエラー: {e}")
        return None

def fix_basic_escaping(json_str):
    """
    基本的なエスケープ処理
    """
    # 文字列値の部分のみを対象として処理
    def escape_json_string_value(match):
        key = match.group(1)
        value = match.group(2)
        
        # 基本的なエスケープ
        value = value.replace('\\', '\\\\')  # バックスラッシュ（最初に処理）
        value = value.replace('"', '\\"')    # ダブルクォート
        value = value.replace('\n', '\\n')   # 改行
        value = value.replace('\r', '\\r')   # 復帰文字
        value = value.replace('\t', '\\t')   # タブ
        
        return f'"{key}": "{value}"'
    
    # "key": "value" パターンを検索して修正
    # 複数行にわたる値に対応
    json_str = re.sub(r'"([^"]+)":\s*"([^"]*(?:[^"\\]|\\.)*)"', 
                      escape_json_string_value, json_str, flags=re.DOTALL)
    
    return json_str

def fix_advanced_escaping(json_str):
    """
    より高度なエスケープ処理（複雑なケース用）
    """
    # JSONの構造を保持しながら、agent_response の値部分のみを修正
    
    # agent_response の値部分を抽出
    response_match = re.search(r'"agent_response":\s*"(.*)"', json_str, re.DOTALL)
    if response_match:
        response_value = response_match.group(1)
        
        # 値部分を適切にエスケープ
        escaped_value = response_value.replace('\\', '\\\\')
        escaped_value = escaped_value.replace('"', '\\"')
        escaped_value = escaped_value.replace('\n', '\\n')
        escaped_value = escaped_value.replace('\r', '\\r')
        escaped_value = escaped_value.replace('\t', '\\t')
        
        # 元のJSONで値部分を置換
        json_str = re.sub(r'("agent_response":\s*)".*"', 
                         f'\\1"{escaped_value}"', json_str, flags=re.DOTALL)
    
    # error の値部分も同様に処理
    error_match = re.search(r'"error":\s*"(.*)"', json_str, re.DOTALL)
    if error_match:
        error_value = error_match.group(1)
        escaped_value = error_value.replace('\\', '\\\\')
        escaped_value = escaped_value.replace('"', '\\"')
        escaped_value = escaped_value.replace('\n', '\\n')
        escaped_value = escaped_value.replace('\r', '\\r')
        escaped_value = escaped_value.replace('\t', '\\t')
        
        json_str = re.sub(r'("error":\s*)".*"', 
                         f'\\1"{escaped_value}"', json_str, flags=re.DOTALL)
    
    return json_str

async def ClassifyKnowledgebase(question, connection_name, tools, llm: BaseChatModel)  -> Dict[str, Any]:
    """
    質問文に適した知識ベースを選択する関数
    
    Args:
        question: 分類する質問文
        connection_name: 接続先名
        tools: MCPツール
        llm: 言語モデル
        
    Returns:
        dict: 結果を含む辞書
    """
    result = {
        "question": question,
        "knowledgebase": None,
        "confidence": None,
        "has_error": False,
        "error_message": None
    }

    PROMPT_FOR_KNOWLEDGE_BASE_CLASSIFIRE = """
    # あなたは、既に接続されているデータベースの get_distances_from_profiles ファンクションを使用して、与えられた質問文と AI Profileの距離を計算するAIエージェントです。
    # 与えられた質問文をファンクションに渡して、AI Profile名 とその距離を取得してください。get_distances_from_profiles からの応答以外の情報は使用しないでください。
    # SQL文は以下のフォーマットで実行してください：
    ```sql
    SELECT * FROM TABLE(get_distances_from_profiles(:search_text))
    ```
    # :search_textの部分には与えられた質問文を設定してください。

    パラメータの設定が不足していてエラーとなった場合はパラメータを見直して再実行してください。

    回答は、必ず以下のJSON形式で返答してください。JSON以外のテキストは含めないでください：
    ```json
    {
    "profiles": [
        {
        "name": "Profile名1",
        "distance": 0.1234567890123456
        },
        {
        "name": "Profile名2", 
        "distance": 0.5234567890123456
        }
    ]
    }
    ```

    エラーが発生した場合には、以下のJSON形式で返答してください：
    ```json
    {
    "error": "エラーの内容"
    }
    ```

    例1（正常な場合）：
    ```json
    {
    "profiles": [
        {
        "name": "Literature_Profile",
        "distance": 0.1234567890123456
        },
        {
        "name": "Sales_Transaction_Profile",
        "distance": 0.5234567890123456
        }
    ]
    }
    ```

    例2（エラーの場合）：
    ```json
    {
    "error": "ファンクション get_distances_from_profiles が見つかりません"
    }
    ```
    """
    
    try:
        # ベクトル距離を使用して質問文を分類するエージェントを作成
        agent_knowledge_base_classifier = create_react_agent(llm, tools, prompt=PROMPT_FOR_KNOWLEDGE_BASE_CLASSIFIRE)
        
        # エージェントを実行
        agent_response = await agent_knowledge_base_classifier.ainvoke({"messages": [{"role": "user", "content": "質問文：" + question + "\n接続先：" + connection_name}]})
        response_text = agent_response['messages'][-1].content
        
        # JSON応答をパース
        data = parse_json_response(response_text)
        
        if data is None:
            result["has_error"] = True
            result["error_message"] = "JSON応答の解析に失敗しました"
            print(f"JSON応答の解析に失敗しました: {response_text}")
            return result
        
        # エラーチェック
        if "error" in data:
            result["has_error"] = True
            result["error_message"] = data["error"]
            return result
        
        # プロファイル情報を取得
        if "profiles" in data and len(data["profiles"]) >= 2:
            profiles = [p["name"] for p in data["profiles"]]
            distances = [p["distance"] for p in data["profiles"]]
            
            # 温度10でのSoftmax確信度を計算
            confidences = softmax_with_temperature(distances, temperature=10.0)
            
            # 最も確信度の高いプロファイルを選択
            best_idx = np.argmax(confidences)
            result["knowledgebase"] = profiles[best_idx]
            result["confidence"] = confidences[best_idx]
            
            # デバッグ情報を出力
            if VERBOSE:
                print(f"\n=== 知識ベース分類詳細情報 ===")
                for i, (profile, distance, confidence) in enumerate(zip(profiles, distances, confidences)):
                    print(f"{i+1}. {profile}: 距離={distance:.6f}, 確信度={confidence:.2f}%")
        else:
            result["has_error"] = True
            result["error_message"] = "2つ以上のAI Profileが必要です"
            
    except Exception as e:
        result["has_error"] = True
        result["error_message"] = str(e)
    
    return result


async def RetrieverRouterAgent(question: str, knowledgebase: str, connection_name: str, tools: List[Tool], llm: BaseChatModel) -> Dict[str, Any]:
    """
    質問文を適切な検索AIエージェントにルーティングして回答を得る
    
    Args:
        question: 質問文
        knowledgebase: 知識ベース（AI Profile名）
        connection_name: 接続先名
        tools: MCPサーバーから取得したツール
        llm: 言語モデル
    
    Returns:
        Dict[str, Any]: {
            "question": 質問文,
            "knowledgebase": 知識ベース（AI Profile名）,
            "response": エージェントの回答,
            "has_error": エラーの有無,
            "error_message": エラーの内容（エラーがない場合はNone）
        }
    """
    result = {
        "question": question,
        "knowledgebase": knowledgebase,
        "response": None,
        "has_error": False,
        "error_message": None
    }
    PROMPT_FOR_ROUTER = """
    # あなたは、既に接続されているデータベースにある適切な検索AIエージェントに問い合わせて与えられた質問文への回答を得るルーターAIエージェントです。
    # 重要：データベースには既に接続されていることを前提としてください。

    # 使用可能な SQL コマンド：
    ## 検索AIエージェントの回答を取得する SQL
    ```sql
    EXEC DBMS_CLOUD_AI.SET_PROFILE('AI Profile名');\nSELECT AI NARRATE 質問文;
    ```
    ### SELECT AI の文法の注意事項
    ** 極めて重要: AI NARRATE には、コメントを含めることは禁止です**
    - RAG queries: `SELECT AI NARRATE 質問文` (/* comment */　は使用禁止。NARRATE の後にはスペースを含めてください)
    - その他の SQLは、使用禁止です。

    # **重要** SELECT AI NARRATE で取得した回答だけをそのまま返答してください。SELECT AI NARRATE で取得した回答以外の情報を一切付け加えないでください。SELECT AI NARRATE で取得した回答に「情報がないので回答できない」ことを意味する文言がある場合は「情報がないので回答できません。」と返答してください。
    # **重要** SELECT AI NARRATE で取得した回答やエラーメッセージに「No matching results」、「行が選択されていません」等といった検索条件にマッチしたデータがないことを意味する文言がある場合はそのまま「情報がないので回答できません。」と返答してください。
    # **重要** 取得した回答に含まれない情報の補足は一切行いません。

    回答は、必ず以下のJSON形式で返答してください。JSON以外のテキストは含めないでください：
   **JSON形式の厳格な要件:**
    1. 改行文字は必ず \\n でエスケープ
    2. ダブルクォートは必ず \\" でエスケープ
    3. バックスラッシュは必ず \\\\ でエスケープ
    4. タブ文字は必ず \\t でエスケープ
    5. コードブロック内の文字も全てエスケープ
    6. 一行で出力し、実際の改行は含めない

    正しい形式:
    {"agent_response": "回答内容（全ての特殊文字がエスケープされている）"}

    エラーが発生した場合:
    {"error": "エラーの内容"}

    **重要な注意点:**
    - JSONは必ず一行で出力してください
    - 文字列内に実際の改行文字を含めないでください
    - 全ての特殊文字を適切にエスケープしてください
    - コード例がある場合も全てエスケープしてください

    例1（正常な場合）：
    {"agent_response": "反重力推進エンジンは、重力を制御することで推進力を得る革新的な推進システムです。"}

    例2（改行を含む回答の場合）：
    {"agent_response": "LLMの主なアーキテクチャには以下があります：\\n1. Transformer\\n2. GPT\\n3. BERT"}

    例3（コードを含む場合）：
    {"agent_response": "以下はPythonのコード例です：\\n```python\\nimport torch\\nprint(\\"Hello World\\")\\n```"}

    例4（情報がない場合）：
    {"agent_response": "情報がないので回答できません。"}

    例5（エラーの場合）：
    {"error": "データベース接続エラーが発生しました"}
    """
    
    
    try:        
        # 分類結果に基づいて適切なリトリーバーを選択し、情報検索を実行するエージェントを作成
        agent_router = create_react_agent(llm, tools, prompt=PROMPT_FOR_ROUTER)
        
        # エージェントを実行
        agent_response = await agent_router.ainvoke({"messages": [{"role": "user", "content": "質問文：" + question + "\n知識ベース（AI Profile名）：" + knowledgebase + "\n接続先：" + connection_name}]})
        response_text = agent_response['messages'][-1].content
        
        # JSON応答をパース
        data = parse_json_response(response_text)
        
        if data is None:
            result["has_error"] = True
            result["error_message"] = "JSON応答の解析に失敗しました"
            return result
        
        # エラーチェック
        if "error" in data:
            result["has_error"] = True
            result["error_message"] = data["error"]
            return result
        
        # 検索結果を取得
        if "agent_response" in data:
            result["response"] = f"{data['agent_response']}"
        else:
            result["has_error"] = True
            result["error_message"] = "検索結果が見つかりません"
                
    except Exception as e:
        result["has_error"] = True
        result["error_message"] = str(e)
    
    return result

async def AgenticRAG(connection_name: str, question: str) -> Dict[str, Any]:
    """
    質問文（文書検索）を適切な知識ベースを検索する検索AIエージェントにルーティングして回答を得る
    """
    result = {
        "question_type": None,
        "knowledgebase": None,
        "confidence": None,
        "response": None,
        "has_error": False,
        "error_message": None
    }
    try:
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                # MCPサーバーを初期化
                print("MCPサーバーを初期化します.........................")
                await session.initialize()

                # MCPサーバーからツールを取得
                tools = await load_mcp_tools(session)

                # データベースへ接続するエージェントを実行
                print("データベース接続エージェントを起動します..........")
                connect_result = await ConnectDatabaseAgent(connection_name, tools, llm)
                if connect_result["has_error"]:
                    return connect_result            

                # 質問文をAnalytics系か文書検索系に分類
                print("質問タイプ分類エージェントを起動します............")
                classify_query_type_result = await ClassifyQueryTypeAgent(question, llm)
                if classify_query_type_result["has_error"]:
                    return classify_query_type_result
                
                question_type = classify_query_type_result["question_type"]
                print(f"質問タイプ: {question_type}")
                
                if question_type == "分析":
                    # Analytics系の質問の場合は、固定のプロファイルでRetrieverRouterAgentを実行
                    knowledgebase = "SALES_HISTORY_PROFILE"
                    confidence = "N/A"
                    print(f"知識ベース（AI Profile名）：{knowledgebase} ")
                else:
                    # 文書検索系の質問の場合は、ClassifyKnowledgebaseで分類
                    print("知識ベース分類エージェントを起動します............")
                    classify_kb_result = await ClassifyKnowledgebase(question, connection_name, tools, llm)
                    if classify_kb_result["has_error"]:
                        return classify_kb_result
                    knowledgebase = classify_kb_result["knowledgebase"]
                    confidence = classify_kb_result["confidence"]
                    print(f"知識ベース（AI Profile名）: {knowledgebase} (確信度: {confidence:.2f}%)")

                # 質問文を指定したカテゴリー（知識ベース）の検索AIエージェントにルーティングして回答を取得するエージェントを実行
                print("検索エージェントを起動します......................")
                retriever_router_result = await RetrieverRouterAgent(question, knowledgebase, connection_name, tools, llm)
                if retriever_router_result["has_error"]:
                    return retriever_router_result
                result["response"] = retriever_router_result["response"]
                result["question_type"] = question_type
                result["knowledgebase"] = knowledgebase
                result["confidence"] = confidence
    except Exception as e:
        result["has_error"] = True
        result["error_message"] = str(e)
    return result

async def main():
    import csv
    import os
    from datetime import datetime
    
    # 出力フォルダを作成
    output_dir = "output"
    os.makedirs(output_dir, exist_ok=True)
    
    # ファイル名生成用の日付と連番
    today = datetime.now().strftime("%y%m%d")
    counter = 1
    
    # 既存ファイルの連番をチェック
    while True:
        csv_filename = f"agenticrag_out_{today}_{counter:03d}.csv"
        md_filename = f"agenticrag_out_{today}_{counter:03d}.md"
        csv_path = os.path.join(output_dir, csv_filename)
        md_path = os.path.join(output_dir, md_filename)
        
        if not os.path.exists(csv_path) and not os.path.exists(md_path):
            break
        counter += 1
    
    # 結果を格納するリスト
    results = []
    
    # デバッグモードを有効化
    #set_debug(True)  # 詳細なデバッグ情報を表示
    #set_verbose(True)  # より詳細な情報を表示
    # ルーティング（分類）するテスト質問文のリスト
    QUESTIONS = [
        # Analytics系の質問
        "売上が最も多かった製品は何で、その売上はいくらですか？",
        "地域別の売上ランキングを教えてください。",
        "2001年に最も売上高が大きかった地域はどこで、その地域で最も売れた製品カテゴリーは何でしたか？",
        "販売に最も貢献しているチャネルは？そのチャネルの売上はいくらですか？",
        "各製品カテゴリーで最も売上が多かった製品は？",
        "年度ごとの総売上高の推移を教えて？",
        # 文書検索系の質問
        "大規模言語モデルとは何ですか？",
        "LLMの主なアーキテクチャは何ですか？",
        "LLMはどのようなデータで訓練されるのですか？",
        "ファインチューニングとは何ですか？",
        "プロンプトエンジニアリングとは何ですか？",
        "LLMの創発的能力とは何ですか？",
        "LLMにはどのような課題や限界がありますか？",
        "単語の埋め込みとは何ですか？",
        "対照学習とは何ですか？",
        "単語埋め込みから文埋め込みを生成するにはどのような方法が使われていますか？",
        "OpenAIのCLIPは、どのように学習したのですか？",
        "ハリー・ポッターに彼が魔法使いであることを告げたのは誰ですか？それはいつのことでしたか？",
        "日本の次期大統領は誰ですか？",
        "宇宙で一番大きな惑星は？",
        "宇宙全体の元素組成を原子の存在比と総質量で教えてください",
        "チャンキングと検索件数と精度の関係を教えて？",
        "画像生成のCFGって何ですか？",
        "リモートセンシングや医療画像などの解析に用いられるRAGはどんなものですか？",
        "ショート動画レコメンデーションのコールドスタート問題にはどのような対処が可能ですか？",
        "環いろはとは誰ですか？",
        "神崎メグの呪文は？",
        "テクニク・テクニカ・シャランラーとは？",
        "錬金術とは何ですか？",
        "化学とは何ですか？",
        "化学と錬金術の違いは？",
        "化学と物理学の違いは？",
        "イーサン・ハントさんの職業は？",
        "暁美ほむらの固有魔法はなんですか？",
        "フェイトが留学していた小学校は？",
        "アースラの艦長は誰？",
        "なのはの魔法の杖の名前は？",
        "管理局の空戦魔導士には誰がいますか？",
        "なのはの最大最強の武器は？",
        "熱力学の第二法則に縛られないエネルギー源は？それを搾取しているのは誰ですか？",
        "フリーレンの年齢は？",
        "フリーレンのパーティーのメンバーは？",
        "ユフィリアの魔道具の名前は？",
        "侯爵令嬢だったユフィーが後に女王となった経緯を教えて？",
        "ホグワーツ特急にはどこで乗れますか？",
        "木之元桜が鍵を杖に変えるときに唱える呪文は？",
    ]
    CONNECTION_NAME = "\'Dev Day Tokyo orasejapan - labuser\'"

    for i, question in enumerate(QUESTIONS, 1):
        print(f"\n=== 質問 {i}: ======================================")
        print(f"質問文: 「{question}」")

        result = await AgenticRAG(CONNECTION_NAME, question)
        
        # 結果の処理
        if result["has_error"]:
            print(f"エラーが発生しました: {result['error_message']}")
            
            # エラーの場合の結果
            result_data = {
                "質問番号": i,
                "質問文": question,
                "リトリーバルタイプ": "エラー",
                "知識ベースタイプ": "エラー",
                "知識ベース分類の確信度": "エラー",
                "生成された回答": f"エラー: {result['error_message']}"
            }
        else:
            print(f"AIエージェントの回答: ============================\n{result['response']}")
            
            # 正常な場合の結果
            # リトリーバルタイプの判定（質問から推測）
            if result["question_type"] == "分析":
                retrieval_type = "分析"
            else:
                retrieval_type = "文書検索"
            
            # 知識ベースタイプの取得
            knowledge_base = result["knowledgebase"]
            
            # 確信度の取得（分析の場合は100%とする）
            confidence = result["confidence"]
            
            result_data = {
                "質問番号": i,
                "質問文": question,
                "リトリーバルタイプ": retrieval_type,
                "知識ベースタイプ": knowledge_base,
                "知識ベース分類の確信度": confidence,
                "生成された回答": result['response']
            }
        
        results.append(result_data)
        print("=" * 50)
    
    # CSVファイルに出力
    with open(csv_path, 'w', newline='', encoding='utf-8-sig') as csvfile:
        fieldnames = ["質問番号", "質問文", "リトリーバルタイプ", "知識ベースタイプ", "知識ベース分類の確信度", "生成された回答"]
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        
        writer.writeheader()
        for result_data in results:
            writer.writerow(result_data)
    
    # マークダウンファイルに出力
    with open(md_path, 'w', encoding='utf-8') as mdfile:
        mdfile.write("# Agentic RAG 実行結果\n\n")
        mdfile.write(f"実行日時: {datetime.now().strftime('%Y年%m月%d日 %H:%M:%S')}\n\n")
        mdfile.write("## 結果一覧\n\n")
        
        # テーブルヘッダー
        mdfile.write("| 質問番号 | 質問文 | リトリーバルタイプ | 知識ベースタイプ | 知識ベース分類の確信度 | 生成された回答 |\n")
        mdfile.write("|----------|--------|-------------------|------------------|------------------------|----------------|\n")
        
        # テーブル内容
        for result_data in results:
            # マークダウンテーブル用にパイプ文字をエスケープ
            question_escaped = result_data["質問文"].replace("|", "&#124;")
            answer_escaped = result_data["生成された回答"].replace("|", "&#124;").replace("\n", "<br>")
            
            mdfile.write(f"| {result_data['質問番号']} | {question_escaped} | {result_data['リトリーバルタイプ']} | {result_data['知識ベースタイプ']} | {result_data['知識ベース分類の確信度']} | {answer_escaped} |\n")
        
        # 詳細結果
        mdfile.write("\n## 詳細結果\n\n")
        for i, result_data in enumerate(results, 1):
            mdfile.write(f"### 質問 {i}\n\n")
            mdfile.write(f"**質問文:** {result_data['質問文']}\n\n")
            mdfile.write(f"**リトリーバルタイプ:** {result_data['リトリーバルタイプ']}\n\n")
            mdfile.write(f"**知識ベースタイプ:** {result_data['知識ベースタイプ']}\n\n")
            mdfile.write(f"**知識ベース分類の確信度:** {result_data['知識ベース分類の確信度']}\n\n")
            mdfile.write(f"**生成された回答:**\n\n{result_data['生成された回答']}\n\n")
            mdfile.write("---\n\n")
    
    print(f"\n結果をファイルに出力しました:")
    print(f"CSV: {csv_path}")
    print(f"Markdown: {md_path}")

if __name__ == "__main__":
    asyncio.run(main())

3.1 データベース接続エージェント

データベース接続エージェントのプロンプト

PROMPT_FOR_DB_CONNECTION = """
# あなたは、データベースに接続するAIエージェントです。
# データベースへ接続するツールを利用できる場合は、それを使って指定された接続先へ接続してください。
# データベースへ接続するツールを利用できない場合は、既にデータベース接続が確立されていると
判断してください。
# 以下のフォーマットで応答してください。
## データベースへ接続した場合:
### フォーマット
[データベースへ接続しました][接続名][接続先のデータベース名]
### 例
[データベースへ接続しました][ADB - adbuser][ADB001]
## 接続するツールを利用できない場合:
### フォーマット
[接続ツールが利用できないため既に接続が確立していると仮定します]
### 例
[接続ツールが利用できないため既に接続が確立していると仮定します]
"""

データベース接続エージェントの実行部分

agent = create_react_agent(llm, tools, prompt=PROMPT_FOR_DB_CONNECTION)
agent_response = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "接続先：" + connection_name}]}
)

ainvoke で指定したmessages（この場合は接続先の情報）とcreate_react_agentで指定したtools（MCPサーバーが提供しているツールのリスト）をLangGraphがプロンプトに追加して LLM を呼び出します
LLM はプロンプトに従って、「データベースへ接続するツール」を探して、「ツールの名前」と「必要なパラメータ（接続先など）」をLangGraphへ返します
LangGraphは、LLMから受け取った情報を元に MCPツールを起動します

質問タイプ分類エージェント（分析系/文書検索系）

質問タイプ分類エージェントのプロンプト

PROMPT_FOR_QUERY_TYPE_CLASSIFIRE = """
# あなたは、ユーザーが提供する質問文が以下の2つのカテゴリーのいずれに属するかを分類する
AIエージェントです。
# 質問に答えることは禁止します。あくまでも、質問文の分類だけを行います。
必ず１つの カテゴリー を選択してください。
# 適切な カテゴリーがない場合にもどちらかと言えばあてはまるカテゴリーを選択してください。
# カテゴリー：
## カテゴリー1：検索
- これは、文書に対するベクトル検索や全文検索に対する質問文です。
### カテゴリー1の質問文の例：
- 海外出張の際の申請手続きについて教えてください？
- Autonomous Databaseのデータベースのバックアップについて教えてください？
- 魔女っ子という言葉が使われなくなり魔法少女と呼ばれるようになったのはなぜですか？
## カテゴリー2：分析
- これは、RDBに格納されている構造化データに対する分析やデータ分析、集計に対する質問文です。
### カテゴリー2の質問文の例：
- 2024年に売上が最も多かった製品は？
- 全販売地域の中で、売上が最も多かった地域は？
- 売上が下落傾向にある製品は？

回答は、必ず以下のJSON形式で返答してください。JSON以外のテキストは含めないでください：
{
    "categories": ["検索", "分析"],
    "selected_category": "選択したカテゴリー"
}

エラーが発生した場合には、以下のJSON形式で返答してください：
{
    "error": "エラーの内容"
}

例1（検索を選択した場合）：
{
    "categories": ["検索", "分析"],
    "selected_category": "検索"
}


例2（分析を選択した場合）：
{
    "categories": ["検索", "分析"],
    "selected_category": "分析"
}


例3（エラーの場合）：
{
    "error": "分類処理中にエラーが発生しました"
}

"""

質問タイプ分類エージェントの実行部分

agent_classifier = create_react_agent(
    llm, tools=[], prompt=PROMPT_FOR_QUERY_TYPE_CLASSIFIRE
)
agent_response = await agent_classifier.ainvoke(
    {"messages": [{"role": "user", "content": "質問文：" + question}]}
)

質問タイプ分類エージェントは、今回の実装はLLM自体に推論させています。そのため MCPツールは使用しません。この場合は、LLMを直接呼び出せばよく ReActエージェントを使う必要はありません。将来的には、より複雑な質問文を複数の質問文に分割（サブクエリー）してから個々のサブクエリーを分類したり、データベースのスキーマ情報などを取得してより高度な分類を実装したくなるかもしれません。そんなときにも、プロンプトを書け変えるだけで様々な検証ができるように仮に ReActエージェントで実装しておきます。

ベクトル距離による知識ベース選択エージェント

ベクトル距離による知識ベース選択エージェントのプロンプト

PROMPT_FOR_KNOWLEDGE_BASE_CLASSIFIRE = """
# あなたは、既に接続されているデータベースの get_distances_from_profiles ファンクションを
使用して、与えられた質問文と AI Profileの距離を計算するAIエージェントです。
# 与えられた質問文をファンクションに渡して、AI Profile名 とその距離を取得してください。
get_distances_from_profiles からの応答以外の情報は使用しないでください。
# SQL文は以下のフォーマットで実行してください：

SELECT * FROM TABLE(get_distances_from_profiles(:search_text))

# :search_textの部分には与えられた質問文を設定してください。

パラメータの設定が不足していてエラーとなった場合はパラメータを見直して再実行してください。

回答は、必ず以下のJSON形式で返答してください。JSON以外のテキストは含めないでください：
{
"profiles": [
    {
    "name": "Profile名1",
    "distance": 0.1234567890123456
    },
    {
    "name": "Profile名2", 
    "distance": 0.5234567890123456
    }
]
}

エラーが発生した場合には、以下のJSON形式で返答してください：
{
"error": "エラーの内容"
}


例1（正常な場合）：
{
"profiles": [
    {
    "name": "Literature_Profile",
    "distance": 0.1234567890123456
    },
    {
    "name": "Sales_Transaction_Profile",
    "distance": 0.5234567890123456
    }
]
}

例2（エラーの場合）：
{
"error": "ファンクション get_distances_from_profiles が見つかりません"
}

"""

ベクトル距離による知識ベース選択エージェントの実行部分

agent_knowledge_base_classifier = create_react_agent(
    llm, tools, prompt=PROMPT_FOR_KNOWLEDGE_BASE_CLASSIFIRE
)
agent_response = await agent_knowledge_base_classifier.ainvoke(
    {"messages": [
        {"role": "user", "content": "質問文：" + question + "\n接続先：" + connection_name}
    ]}
)

4. メインのルーティングロジック

メインのルーティングロジック

async def AgenticRAG(connection_name: str, question: str) -> Dict[str, Any]:
    """質問文を適切な知識ベースを検索する検索AIエージェントにルーティングして回答を得る"""
    
    # 1. データベース接続
    result = await ConnectDatabaseAgent(connection_name, tools, llm)
    
    # 2. 質問タイプの分類（分析系/文書検索系）
    classify_result = await ClassifyQueryTypeAgent(question, llm)
    question_type = classify_result["category"]
    
    if question_type == "分析":
        # 分析系は固定の知識ベース（AIプロファイル）
        knowledgebase = "SALES_HISTORY_PROFILE"
    else:
        # 文書検索系はベクトル距離で分類
        result = await ClassifyKnowledgebase(question, connection_name, tools, llm)
        knowledgebase = result["knowledgebase"]
        confidence = result["confidence"]
        print(f"知識ベース（AI Profile名）: {knowledgebase} (確信度: {confidence:.2f}%)")
    
    # 3. 選択されたプロファイルで検索実行
    result = await RetrieverRouterAgent(question, category, connection_name, tools, llm)
    return result

Agentic RAG の利点

1. エージェントが質問を適切な検索エージェント（分析／文書検索）へルーティング

分析系の質問なら Aシステム、技術文書の検索なら Bシステム、社内規定の検索なら Cシステムとユーザーがシステムを使分けるのではなく、ユーザーの意図に沿った検索エージェントをAI自身が自動的に選択して回答を返してくれる

2. エージェントが質問を適切な知識ベースを担当する検索エージェントへルーティング

専門特化したち知識ベースを構築することで互いがノイズとならず高い回答精度を実現できる。

ベクトル距離による質問文の知識ベースへの分類の利点

1. 高精度な分類

各知識ベースの内容を反映した平均ベクトルにより、質問文との意味的な類似性を正確に判定

2. スケーラビリティ・拡張性

新しい知識ベースの追加時も、平均ベクトルを計算するだけで対応可能
プロンプトの修正が不要

3. 透明性

距離と確信度が数値で確認でき、分類の根拠が明確

実行例と結果

# 魔法関連の質問
"高町なのはの最大最強の砲撃魔法は何？"
# → MAGIC_RAG_PROFILE (確信度: 85.3%)

# 技術関連の質問  
"MCPは、アプリ開発者にとってどんなメリットがありますか？"
# → GENERATIVE_AI_RAG_PROFILE (確信度: 92.1%)

# 売上分析の質問
"2001年に最も売上高が大きかった地域はどこで、その地域で最も売れた製品カテゴリーは何でしたか？"
# → SALES_HISTORY_PROFILE (分析系として固定選択)

パイプライン全体の実行結果例

質問番号20,21 は、どの知識ベースにも関連ドキュメントを登録していませんので、分類の確信度が小さくなっている上に「情報がないので回答できません」となっていて正しい挙動です。

今後の展望

1. ハイブリッドアプローチ

ベクトル距離とLLMによる判定を組み合わせた、より堅牢な分類システムの構築。例えば、ベクトル距離に基づく確信度がxx%以下の場合は、LLMに判定させるなど。

2. フォールバック

分類が間違いであったために回答を生成できなかった場合に、別の知識ベースに対して検索を行うなど。これはAdaptive RAGの考え方で、クエリの複雑さに応じて動的に検索戦略を調整します。

3. フィードバックループ

ユーザーフィードバックを基に平均ベクトルを更新し、継続的な精度向上を実現。併せて、Reflectionパターンを活用することで、システムが自己評価・自己改善を行える仕組みの構築。

4. 分析系にも分類を導入

どのテーブル群に対する質問であるかを分類して、NL2SQLの候補となるテーブルを絞り込む。Planningパターンを活用し、複雑なSQL生成タスクを段階的に処理。

5. 複雑な質問への対応

質問をサブクエリーに分割して、それぞれに回答生成した上で統合するなど。これはMulti-Agent Collaborationの応用で、各サブクエリーを専門エージェントが処理し、最終的に統合エージェントがまとめる仕組み。

6. 質問タイプの分類にも質問文データセットを用意してベクトル距離などの分類手法でより高精度に分類する

7. 質問の分類とルーティング以外にもAIを活用

検索結果のチャンク群の中で回答生成に役立つものとノイズになるものを回答生成前に判定してノイズを削減したらり、情報が足りない場合は検索クエリーを修正して再検索を行うなど。最新の研究では、Corrective RAGのように、検索結果の品質を動的に評価し、必要に応じて再検索や修正を行うアプローチが提案されています。

8. インデータベースで実装

セキュリティやレイテンシーなどを重視する場合には、データベース内で稼働する AI エージェントですべてを実装するというアプローチもあるかもしれません。

まとめ

本記事では、MCP対応SQLclとLangGraphを活用し、ベクトル埋め込みによる高精度なルーティングを実装したエージェンティックRAGシステムにチャレンジしてみました。

主な特徴：

ベクトル距離による客観的な分類: 各知識ベースの平均ベクトルを使った意味的類似性の計算
MCPによる柔軟な拡張性: データベース操作をツールとして公開し、エージェントから利用
LangGraph Pre-built ReAct Agentによるシンプルな実装: MCPツールを自動的に選択して実行することでプログラミングをプロンプティングへ変換
LangGraph Pre-built ReAct Agentを単一タスクに限定して使用した確定的なワークフローによる確実な動作

この手法により、従来のプロンプトベースの分類よりも高精度で、かつ説明可能なルーティングが実現できそうです。エージェンティックAIの考え方とベクトル検索技術を組み合わせることで、より実用的で信頼性の高いRAGシステムの構築が可能になっていますので RAG と Agentic AI を活用した生成AIの応用が増えていくのではないでしょうか。

おまけ

この記事を元にしたウェビナー「生成AI実用化の鍵　エンジニアのためのエージェントRAG／MCPとマルチモーダル」では、この記事の内容に続けて Agentic RAG をインデータベースで実現するお話し【エージェンティック AI で RAGを賢く - インデータベースAIエージェント編】とマルチモーダル RAG のデモもさせていただきました。

アーカイブ動画もありますのでぜひ見て行ってください。

参考リソース

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