実務の現場では、PoC（概念実証）段階ではうまく動いていたRAG（検索拡張生成）システムが、実運用に向けて機能を追加していく過程で、コードが複雑化していくという課題が頻繁に観察されます。
例えば、
- ユーザーの質問タイプによって検索するインデックスを変えたい
- キーワード検索とベクトル検索を併用して、結果をリルート（Re-ranking）したい
- 検索結果が0件だった場合は、Web検索にフォールバックさせたい
といった要件を従来のif-else文や手続き的なコードで実装していくと、保守困難な「スパゲッティコード」になりがちです。ロジックの流れが追いにくく、デバッグも困難になり、アジャイルな改善が難しくなることもあります。
そこで今回は、LlamaIndexが提供する「Query Pipeline API」に焦点を当てます。これは、複雑なクエリ処理フローをDAG（有向非巡回グラフ）として宣言的に定義できる仕組みです。技術の本質を見抜き、ビジネス要件へ最短距離で到達するための強力な武器となります。
Pipelineが必要な理由から、条件分岐を含む高度なロジックの実装まで、ステップバイステップで解説していきます。まずは動くプロトタイプをイメージしながら、堅牢なAIパイプラインの構築手法を見ていきましょう。
なぜ従来のChainやQueryEngineでは限界が来るのか
LlamaIndexの導入初期において、VectorStoreIndex.as_query_engine()のような高レベルAPIは非常に強力なツールです。数行のコードでRAGシステムを立ち上げ、ドキュメントとの対話を実現できます。しかし、ビジネス要件が高度化し、カスタマイズの要求が高まると、この「隠蔽された便利さ」が逆に制約となるケースは珍しくありません。
単純なシーケンシャル処理の限界
従来のQueryEngineや、LangChainにおける初期のChain構造は、基本的に「入力 A → 処理 B → 出力 C」という直線的なシーケンシャル（順次）処理に最適化されていました。
しかし、現在のAIアプリケーション開発、特に「進化型RAG」や「自律型エージェント」の文脈では、現実はもっと複雑です。例えば、以下のようなフローを想像してください。
- ユーザーの質問を意図分類モデルで分析する
- 質問が「要約」ならLLM単体処理、「詳細検索」ならRAGプロセスへ分岐
- RAGプロセスでは、社内ベクトルDB検索とWeb検索を並列実行
- 検索結果をリランキング（再順位付け）し、コンテキストとして統合
- 回答生成後、事実確認（Fact Check）を行い、不十分なら再検索ループに入る
これを単一の関数や手続き的なコードで記述しようとすると、条件分岐や状態管理が複雑に絡み合い、いわゆる「スパゲッティコード」になりがちです。業界全体としても、LangChainがLangGraphのようなグラフベースのアプローチへシフトしているように、複雑なロジックをグラフ構造で管理する手法が標準になりつつあります。
宣言的記述 vs 手続き的記述
Query Pipelineの最大の利点は、処理フローを「宣言的（Declarative）」に記述できる点にあります。
- 手続き的（従来のコード）: 「まずAを実行して、その結果を変数xに入れ、次にif文で分岐して...」と、処理の具体的な手順を命令として記述します。
- 宣言的（Pipeline）: 「Aの出力はBの入力につながる。BとCは並列に動作する」と、コンポーネント間の関係性（トポロジー）を定義します。
宣言的に記述することで、システム全体がDAG（有向非巡回グラフ）として可視化されやすくなります。これは、DevOpsやMLOpsの観点からも極めて重要です。データの流れが詰まっている箇所や、エラーが発生しているノードを構造的に把握できるため、トラブルシューティングの効率が格段に向上します。
Query Pipelineが解決する「拡張性」の課題
当初は単純なQAボットとして始まったプロジェクトが、最終的に「契約書レビュー」「リスク分析」「条項検索」を兼ね備えた複合的なAIアシ কূটনীতিকへと進化することはよくあります。
初期段階で手続き的な実装に固執していた場合、この拡張に耐えられず、大規模なリファクタリングを余儀なくされる可能性があります。経営者視点で見れば、これは開発スピードの低下とコスト増大に直結します。Query Pipelineを採用していれば、新しい処理ブロック（モジュール）を既存のグラフに追加するだけで機能を拡張でき、既存ロジックへの影響を最小限に抑えることが可能です。
また、最新のRAG評価フレームワーク（例えばRagasの最新版など）と組み合わせる際も、パイプラインの各ノードが明確に定義されているため、コンポーネントごとの精度評価や改善が容易になるというメリットもあります。
開発環境の準備と基本コンポーネントの理解
それでは、実際に手を動かしていきましょう。プロトタイプ思考で、まずは動くものを作って検証していくことが重要です。ここでは環境構築と、Query Pipelineを構成する基本的な要素について解説します。
LlamaIndexのアップデートと必要なライブラリ
LlamaIndexのエコシステムは急速に進化しています。特にv0.10以降では設計が刷新され、コア機能とLLMプロバイダーごとの統合機能が分離されました。本記事では、この新しい名前空間パッケージを使用します。
以下のコマンドで、コアライブラリとOpenAI用の連携パッケージをインストールします。
pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai
専門家からのアドバイス:
OpenAI等のLLMプロバイダーは頻繁にモデルを更新しています（例：ChatGPTやヘルスケア特化機能など）。これらの新機能に即座に対応し、仮説検証のサイクルを高速に回すためにも、ライブラリは常に最新版を使用することを強く推奨します。
InputComponentとOutputComponentの役割
Query Pipelineは、大きく分けてModules（モジュール）とLinks（リンク）で構成されます。
- Modules: 処理の最小単位です。LLM、プロンプトテンプレート、Retriever（検索器）、あるいは任意のPython関数などがこれに該当します。
- Links: モジュール間をつなぐデータの通り道です。あるモジュールの出力を、次のモジュールの特定の入力引数へ渡します。
パイプラインの設計において特に重要なのが、データの入り口と出口を定義する以下のコンポーネントです。
InputComponent: パイプラインの始点となります。ユーザーからのクエリや追加のパラメータなど、外部からの入力を受け付けます。必要に応じて複数のキーを持たせることも可能です。OutputComponent: パイプラインの終点です。処理された最終結果を外部へ返却します。
Linkオブジェクトによるデータフローの定義
グラフ構造（DAG）を定義する際、開発者は「どのモジュールからどのモジュールへデータを流すか」を明示的に設計します。LlamaIndexの QueryPipeline クラスでは、単純な直列接続には add_chain を、複雑な分岐や合流には add_link を使用します。
データの流れは、概念的に以下のようなイメージになります。
graph LR
Input[Input: query_str] --> Retriever
Retriever -->|nodes| Synthesizer[ResponseSynthesizer]
Input -->|query_str| Synthesizer
Synthesizer --> Output
この図が示すように、ResponseSynthesizer（回答生成器）は、Retriever が取得した検索結果（nodes）だけでなく、元のユーザーの質問（query_str）も必要とします。Query Pipelineを使用すれば、このように「入力を維持したまま下流のコンポーネントへ渡す（飛び越えリンク）」設計も、コード上で明確に記述できます。
Step 1: シンプルな「検索→回答」パイプラインの構築
まずは最小限の構成で動くプロトタイプを作り、感覚を掴んでいきましょう。
RetrieverとResponseSynthesizerの定義
通常のLlamaIndexの構築と同様に、まずは必要なコンポーネントを準備します。ここでは、推論能力の高いOpenAIの最新モデルを利用する構成を例にします。
import os
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI
from llama_index.core.response_synthesizers import TreeSummarize
# APIキーの設定（環境変数に設定されている前提）
# os.environ["OPENAI_API_KEY"] = "sk-..."
# 1. データの読み込みとインデックス作成
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
# 2. コンポーネントの準備
retriever = index.as_retriever(similarity_top_k=3)
# モデルの指定（最新の安定版モデルを推奨）
# 公式ドキュメントを参照し、要件に合ったモデル（ChatGPTなど）を選択してください
llm = OpenAI(model="別のAIサービス")
# Response Synthesizer（検索結果と質問から回答を作る部品）
summarizer = TreeSummarize(llm=llm)
add_modulesによるコンポーネント登録
次に、QueryPipelineを初期化し、モジュールを登録します。verbose=Trueを設定することで、実行時の各ステップの入出力をログで確認できるため、デバッグ時に非常に有用です。
from llama_index.core.query_pipeline import QueryPipeline, InputComponent
# パイプラインの初期化（verbose=Trueで実行ログを表示）
qp = QueryPipeline(verbose=True)
# モジュールの登録
qp.add_modules({
"input": InputComponent(),
"retriever": retriever,
"summarizer": summarizer,
})
add_linkによる処理順序の指定
ここがポイントです。モジュール同士を接続し、データの流れ（DAG）を定義します。
# 1. 入力 -> Retriever (query_strを渡す)
qp.add_link("input", "retriever")
# 2. Retriever -> Summarizer (nodesを渡す)
qp.add_link("retriever", "summarizer", dest_key="nodes")
# 3. 入力 -> Summarizer (query_strを渡す)
# Summarizerはnodes（検索結果）とquery_str（質問文）の両方が必要
qp.add_link("input", "summarizer", dest_key="query_str")
これでパイプラインの完成です。実行してみましょう。
response = qp.run(input="このプロジェクトの主な目的は何ですか？")
print(str(response))
index.as_query_engine().query()を呼ぶ従来の方法よりもコード量が増えているように見えるかもしれませんが、データの流れ（特にsummarizerが2つの入力を必要としている点）がコード上で明確になっていることに注目してください。この明示的な構造こそが、将来的に複雑なロジックを組む際の保守性を高める鍵となります。
Step 2: 複数の入力を扱う「ハイブリッド検索」への拡張
Query Pipelineの真価が発揮されるのは、まさにここからです。単一の検索処理ではなく、特性の異なる複数の検索手法を組み合わせる「ハイブリッド検索」の実装を通して、その柔軟性を見ていきましょう。
一般的に、意味的な類似性を捉える「ベクトル検索」と、正確なキーワード一致を重視する「キーワード検索（BM25など）」を併用することで、RAGの回答精度は飛躍的に向上します。ChatGPTのような高性能なLLMをバックエンドに据える場合でも、入力されるコンテキストの質が低ければ、ハルシネーションのリスクは排除できません。
ここでは、並列処理と結果の統合を行うフローを構築します。
複数のRetrieverを並列に配置する
まず、ベクトル検索器（Vector Retriever）とキーワード検索器（BM25 Retriever）を定義し、それらをパイプラインのモジュールとして追加します。
from llama_index.retrievers.bm25 import BM25Retriever
# BM25 Retrieverの準備
# 既存のdocstoreからBM25インデックスを構築する場合
bm25_retriever = BM25Retriever.from_defaults(docstore=index.docstore, similarity_top_k=3)
# パイプラインへの追加
# input: クエリ文字列を受け取る入口
# vector_retriever: 意味検索を担当
# bm25_retriever: キーワード検索を担当
qp.add_modules({
"input": InputComponent(),
"vector_retriever": retriever,
"bm25_retriever": bm25_retriever,
"join_results": CustomJoinComponent(), # 後述する統合用コンポーネント
"summarizer": summarizer
})
カスタムコンポーネントによる検索結果の統合（Merge）
Pipeline内で独自のロジックを実行したい場合、Python関数をコンポーネント化するのが最も手軽で強力な方法です。LlamaIndexの CustomQueryComponent を継承し、2つの検索結果リストをマージして重複を排除するコンポーネントを作成します。
from llama_index.core.query_pipeline import CustomQueryComponent
from typing import List, Dict, Any, Set
from llama_index.core.schema import NodeWithScore
class ResultMergeComponent(CustomQueryComponent):
"""2つの検索結果リストを結合し、IDベースで重複を除去するコンポーネント"""
model_config = {"arbitrary_types_allowed": True}
def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
# 本番環境では厳密な型チェックを推奨
return input
@property
def _input_keys(self) -> Set[str]:
# 2つの入力を受け取ることを定義
return {"nodes_a", "nodes_b"}
@property
def _output_keys(self) -> Set[str]:
return {"nodes"}
def _run_component(self, **kwargs) -> Dict[str, Any]:
nodes_a = kwargs.get("nodes_a", [])
nodes_b = kwargs.get("nodes_b", [])
# 単純な結合と重複排除（Node IDベース）
seen_ids = set()
merged_nodes = []
# 優先順位に応じた結合ロジックをここに実装可能
for node in nodes_a + nodes_b:
if node.node.node_id not in seen_ids:
merged_nodes.append(node)
seen_ids.add(node.node.node_id)
return {"nodes": merged_nodes}
分岐と結合のロジック実装（DAGの定義）
コンポーネントが揃ったら、それらをリンクさせてデータフロー（DAG）を定義します。ここでのポイントは、入力（クエリ）が2つのRetrieverに「分岐」し、その出力がMergeコンポーネントで「合流」するという流れです。
# 1. 入力から各Retrieverへ分岐（同じクエリを両方に渡す）
qp.add_link("input", "vector_retriever")
qp.add_link("input", "bm25_retriever")
# 2. 各Retrieverの出力をMergeコンポーネントの特定の入力キーへ接続
qp.add_link("vector_retriever", "join_results", dest_key="nodes_a")
qp.add_link("bm25_retriever", "join_results", dest_key="nodes_b")
# 3. 統合された結果（nodes）をSummarizer（LLM）へ渡す
qp.add_link("join_results", "summarizer", dest_key="nodes")
# 4. 元のクエリ文字列もSummarizerへ渡す必要がある
qp.add_link("input", "summarizer", dest_key="query_str")
このように、Query Pipelineを使用すれば、複雑になりがちなハイブリッド検索のロジックを、視覚的に理解しやすいグラフ構造としてコード化できます。
さらに精度を高めたい場合は、Mergeコンポーネントの後に CohereRerank などのRerankingモデルを挟むことも容易です。コンポーネントを一つ追加し、リンクを繋ぎ変えるだけで、システム全体のアーキテクチャを大きく変更することなく機能拡張が可能になります。これは、変化の速いAI技術トレンドに追従する上で非常に重要な特性と言えるでしょう。
Step 3: 条件分岐（Router）を含む高度なロジックの実装
さらに実用的なシナリオとして、「質問の内容に応じて処理を変える」ルーター機能を実装します。静的なパイプラインから、状況判断を行う動的なパイプラインへの進化です。これはQuery Pipelineの真価を発揮する強力な機能の一つと言えます。
ユーザーの意図分類を行うRouterComponentの配置
LlamaIndexにはRouterComponentとSelectorが用意されており、これらを使って動的なパス切り替えを実現します。
特に最近のLLM（ChatGPTなど）は推論能力が飛躍的に向上しているため、ユーザーの曖昧な意図を分類する精度も実用レベルに達しています。例えば、医療分野の質問なら専門の推論モデルへ、一般的な質問なら軽量モデルへ振り分けるといった高度な制御も、このコンポーネントで実現可能です。
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.query_pipeline import RouterComponent
# 選択肢の定義
# 最新のLLMを使用することで、よりニュアンスを含んだ分類が可能になります
choices = [
"要約: 文書全体の要約や概要を知りたい場合",
"詳細検索: 特定の事実や数値、詳細な情報を検索したい場合"
]
# Routerコンポーネントの作成
# llmには推論能力の高いモデル（ChatGPTやClaudeの最新モデル等）を指定することを推奨します
router_component = RouterComponent(
selector=LLMSingleSelector.from_defaults(llm=llm),
choices=choices,
components=["summary_pipeline", "detail_pipeline"] # 分岐先のキー
)
条件に応じたサブパイプラインの実行
Pipelineの中に別のPipeline（サブパイプライン）を組み込むことも可能です。これにより、複雑な処理をモジュール化して管理できます。ここではシンプルにするために、同じPipeline内で分岐を作ってみましょう。
Routerは、LLMを使ってユーザーの質問を分析し、適切なキー（ここではsummary_pipelineかdetail_pipeline）を選択して、その方向へデータを流します。
# 各ルート用のコンポーネント（例としてPromptTemplateを変えるなど）
# ...（省略：summary_promptとdetail_promptを定義）...
qp = QueryPipeline(verbose=True)
qp.add_modules({
"input": InputComponent(),
"router": router_component,
"summary_chain": summary_chain, # 要約用の処理チェーン
"detail_chain": detail_chain, # 詳細検索用の処理チェーン
})
# 入力をRouterへ
qp.add_link("input", "router", dest_key="query_str")
# Routerからの分岐定義
# RouterComponentは、選択されたキーに対応する出力エッジを活性化します
qp.add_link("router", "summary_chain", src_key="summary_pipeline")
qp.add_link("router", "detail_chain", src_key="detail_pipeline")
If-Elseロジックをグラフ構造で表現する
プログラム上のif-else文を、グラフ上の「分岐（Branching）」として表現することで、ロジックが視覚化され、チームでの共有やデバッグが容易になります。
また、システム思考の観点からは「エッジケース」への対応も重要です。「どちらの条件にも当てはまらない場合」や「エラーが発生した場合」のフォールバック処理をDAG（有向非巡回グラフ）上のパスとして明示的に定義することで、予期せぬ挙動を防ぎ、堅牢なアプリケーションを構築できます。複雑化するAIアプリケーションにおいて、このようにロジックを可視化・構造化できる点は、保守性の観点からも非常に大きなメリットとなります。
トラブルシューティングと可視化テクニック
複雑なRAGパイプラインやエージェントワークフローを構築すると、「どこでデータの受け渡しが失敗しているのか」「なぜ意図した回答にならないのか」といった問題に直面することは珍しくありません。ここでは、LlamaIndex Query Pipelineを効率的にデバッグし、構造を把握するための実践的なテクニックを解説します。
パイプライン構造の可視化（グラフ描画）
コードだけでDAG（有向非巡回グラフ）の構造を把握するのは困難です。構築したパイプラインが設計通りに接続されているかを確認するために、可視化機能を活用しましょう。pyvisなどのライブラリと連携することで、パイプラインの構造をインタラクティブなHTMLグラフとして出力できます。
from llama_index.core.query_pipeline import QueryPipeline
# ...パイプライン構築後...
# パイプラインの構造をHTMLファイルとして保存
qp.save_net_to_file("pipeline_graph.html")
出力されたHTMLをブラウザで開くと、各コンポーネント（ノード）とそれらを繋ぐリンクが可視化されます。特に、条件分岐や複数の入力を持つノードがある場合、リンクの向きや接続先が正しいかを視覚的に検証できるため、ロジックのミスを早期に発見できます。
中間出力の確認とデバッグ方法
パイプラインがエラーなく実行されても、最終出力が期待通りでない場合があります。その際、ブラックボックス化を防ぐために中間ステップの挙動を確認することが重要です。
最も基本的な方法は、verbose=Trueを設定することです。これにより、各ステップの入出力や実行状況がコンソールに出力されます。
# 開発・デバッグ時はverboseモードを有効化
qp = QueryPipeline(verbose=True, modules=...)
さらに高度なデバッグが必要な場合は、以下の手法も有効です：
- 段階的な実行: パイプライン全体を実行するのではなく、特定のコンポーネントまでを実行し、その時点での出力を検証します。
- ダミーコンポーネントの挿入: デバッグ用の
OutputComponentを中間に挟み、データフローを一時的にキャプチャして内容を確認します。 - トレーシングツールの活用: 本番環境に近い開発では、Arize PhoenixやLlamaTraceといった可観測性（Observability）ツールと連携し、レイテンシやトークン使用量、プロンプトの履歴を詳細に追跡することをお勧めします。
よくあるエラーと対処法
パイプライン構築時によく遭遇するエラーとその解決策をまとめました。
- ValueError: No path found
- 原因: 定義したリンクが途切れており、入力から出力までのルートが存在しない場合に発生します。
- 対策: 可視化グラフを確認し、孤立しているノードや接続されていないポートがないかチェックしてください。
- KeyError
- 原因:
add_linkメソッドでのsrc_key（出力キー）やdest_key（入力キー）の指定ミスです。 - 対策: 各コンポーネントが要求する正確なキー名（例:
query_str,nodes,responseなど）をドキュメントやソースコードで確認しましょう。特にカスタムコンポーネントを使用する場合、入出力キーの定義漏れに注意が必要です。
- 原因:
- API接続・認証エラー
- 原因: LLMプロバイダー（OpenAIやAnthropicなど）のAPIキー設定ミスや、指定したモデル名が誤っている（または廃止されている）ケースです。
- 対策: 環境変数が正しく読み込まれているか確認してください。また、利用しようとしているモデル（例: ClaudeやChatGPTの特定バージョン）が、使用中のライブラリバージョンでサポートされているか、正しいモデルIDを指定しているかを確認する必要があります。
まとめ：保守可能なAIアプリケーションを目指して
今回は、LlamaIndexのQuery Pipeline APIを用いて、複雑なRAGロジックを整理・実装する方法を解説しました。
従来のChainやQueryEngineによる実装と比較して、Query Pipelineには以下のメリットがあります。
- 可読性の向上: 処理フローがDAG（有向非巡回グラフ）として可視化され、ブラックボックス化を防げます。
- 拡張性の確保: 新しい検索手法や後処理ロジックを、既存のパイプラインに「プラグイン」感覚で追加可能です。
- デバッグの容易さ: 各モジュールの入出力を個別に検証でき、問題の切り分けがスムーズになります。
AI技術、特にLLMアプリケーションの開発環境は極めて変化が速いです。例えば、Google Vertex AI SDKの生成AIモジュールから新しいGoogle Gen AI SDKへの移行や、LangChainコアライブラリにおけるセキュリティ強化（CVE-2025-68664対応など）といった重要な変更が頻繁に行われています。こうした外部環境の変化に柔軟に対応できるアーキテクチャを採用することが、システムの寿命を延ばす鍵となります。
Query Pipeline活用のベストプラクティス
- 小さく作って組み合わせる: 巨大なパイプラインを一つ作るのではなく、機能ごとに小さなパイプラインを作り、それらをモジュールとして組み合わせる設計を推奨します。まずは動く最小単位を作るアプローチです。
- カスタムコンポーネントの活用: 独自のビジネスロジックは積極的にカスタムコンポーネント化し、再利用可能な資産として蓄積しましょう。
- 依存関係の局所化: モデルプロバイダーのSDK（OpenAIやGoogle等）に依存する部分は、特定のコンポーネント内に閉じ込めることで、将来的なSDKの仕様変更や廃止（Deprecation）が生じた際も、修正範囲を最小限に抑えることができます。
次のステップ：エージェント化への道
Query Pipelineで構築した堅牢なRAGシステムは、より高度な「自律型エージェント」への足掛かりとなります。
ChatGPTやGeminiなど、推論能力が強化されたモデルが登場し、AIが自律的にタスクを計画・実行する能力が向上しています。また、LangGraphのようなエージェントオーケストレーション機能も進化し、グラフ構造での制御フロー管理が標準になりつつあります。
Query Pipelineで整理された検索・推論ロジックは、これらのエージェントフレームワークにおける「ツール」や「スキル」としてそのまま転用可能です。まずは現状のRAGロジックをPipelineで整理し、ビジネス価値を生み出す保守性の高い基盤を築くことから始めてみてはいかがでしょうか。皆さんのプロジェクトでも、ぜひこのアプローチを試して、その効果を実感してみてください。
コメント