【イベント駆動】LlamaIndexの次世代機能「Workflows」入門！LangGraphと比較する頑健なAIエージェント構築手順

LLM（大規模言語モデル）を活用したアプリケーション開発において、単一のプロンプト処理から、自律的に思考し行動する「AIエージェント」の構築へとトレンドがシフトしています。その中で、RAG（検索拡張生成）フレームワークとしてデファクトスタンダードの地位を築いている LlamaIndex から、次世代のエージェント構築機能である「Workflows（ワークフロー）」がリリースされました。

本記事では、LlamaIndex Workflowsの基本概念から、競合となるLangGraphとの設計思想の違い、具体的なPythonコードを用いた実装手順、開発時に陥りやすいトラブルシューティングまでを徹底的に解説します。

1. LlamaIndex Workflowsとは？イベント駆動型がもたらす新パラダイム

従来のLlamaIndexでは、処理の流れを一本道で定義する QueryEngine や、ノードを連結する IngestionPipeline などが主軸でした。しかし、複雑な条件分岐や並行処理、人間による承認（Human-in-the-loop）が必要な高度なエージェントを構築する場合、従来の線形な構造では記述が極めて困難でした。

LlamaIndex Workflowsは、これらの課題を「イベント駆動型（Event-driven）」のアプローチで解決します。全体のフローを静的なグラフとして定義するのではなく、特定の「イベント」の発生をトリガーとして、各「ステップ」が自律的に実行される柔軟なアーキテクチャを採用しています。

これにより、以下のような高度なユースケースを直感的かつ堅牢に実装できるようになりました。

LLMの出力結果に応じた動的なルーティング（条件分岐）
複数の並行タスクがすべて完了した後に次の処理へ進むジョイン（Join）処理
長時間実行されるプロセスの中断と、人間からのフィードバックによる再開

開発ソースは LlamaIndex GitHubリポジトリで公開されており、オープンソースコミュニティによって急速に機能拡張が進んでいます。

2. LangGraph vs LlamaIndex Workflows：設計思想の比較

状態管理型のAIエージェントフレームワークといえば、LangChainファミリーの「LangGraph」が有名です。ここで、両者の違いを表で比較してみましょう。

比較項目	LangGraph	LlamaIndex Workflows
基本アーキテクチャ	状態遷移グラフ（DAGベース）	イベント駆動（Pub/Sub風）
状態（State）の定義	スキーマを事前に厳密定義	コンテキストオブジェクトによる柔軟な共有
ステップ間の結合度	ノードとエッジの定義による強結合	イベントの送受信による疎結合
RAGとの親和性	LangChainエコシステムに依存	LlamaIndexの豊富なインデックス群と直結

LangGraphの詳細は LangGraph超入門の記事で詳しく解説していますが、LangGraphが「状態（State）を中心に据え、明示的な遷移エッジで制御する」のに対し、LlamaIndex Workflowsは「任意のイベントをトリガーに、実行可能なステップが動的に発火する」という違いがあります。

Workflowsは各ステップが疎結合に保たれるため、プログラムのモジュール性が非常に高く、ステップの追加や変更が容易であるという開発上の大きなメリットがあります。

3. Workflowsを構成する4つのコア概念

LlamaIndex Workflowsを理解する上で、重要な4つの構成要素があります。

Workflow (ワークフロー) システム全体のコンテナとなるクラスです。Workflow クラスを継承して独自クラスを作成します。
Step (ステップ) 実際の処理を行う関数です。@step() デコレータを付与することで、ワークフローの一部として登録されます。
Event (イベント) ステップ間で受け渡されるデータコンテナです。Pydanticをベースとした Event クラスを継承して作成します。
Context (コンテキスト) ワークフローの実行状態や、グローバルな変数をステップ間で保持・共有するためのオブジェクト（Context）です。

4. 実装ハンズオン：条件分岐を持つAIエージェントの構築

それでは、実際にPythonを使ってシンプルなWorkflowを実装してみましょう。今回は「ユーザーの質問内容を分析し、一般的な質問か、専門的な問い合わせ（テクニカル質問）かを自動的に分類して、異なるLLMプロンプトで応答するエージェント」を構築します。

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

まずはLlamaIndexのコアパッケージと、今回使用するOpenAIのLLM統合パッケージをインストールします。

pip install llama-index llama-index-core llama-index-llms-openai

4.2 Pythonによる実装コード

以下のコードは、入力されたクエリをLLMが判別し、それぞれ適切な処理ステップへとイベントを飛ばすルーティングの例です。

import asyncio
import os
from llama_index.core.workflow import (
    Workflow,
    Step,
    StartEvent,
    StopEvent,
    Event
)
from llama_index.llms.openai import OpenAI

# OpenAI APIキーの設定（環境変数から読み込むことを推奨）
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# 1. 独自イベントの定義
class GeneralQueryEvent(Event):
    query: str

class TechnicalQueryEvent(Event):
    query: str

# 2. ワークフローの構築
class SupportAgentWorkflow(Workflow):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.llm = OpenAI(model="gpt-4o-mini")

    @step()
    async def classify_input(self, ev: StartEvent) -> GeneralQueryEvent | TechnicalQueryEvent:
        """ユーザー入力を判定し、対応するイベントをディスパッチする"""
        user_query = ev.get("query")
        
        prompt = f"""
        ユーザーの質問を分析し、一般的な質問（日常会話、単純な挨拶、手続きなど）か、
        技術的な質問（プログラミング、システムエラー、仕様など）かを判定してください。
        回答は必ず 'general' または 'technical' のどちらか1単語のみで出力してください。
        
        質問: {user_query}
        """
        
        response = await self.llm.acomplete(prompt)
        category = response.text.strip().lower()
        
        if "technical" in category:
            print(f"[INFO] テクニカルな質問と判定されました: {user_query}")
            return TechnicalQueryEvent(query=user_query)
        else:
            print(f"[INFO] 一般的な質問と判定されました: {user_query}")
            return GeneralQueryEvent(query=user_query)

    @step()
    async def handle_general_query(self, ev: GeneralQueryEvent) -> StopEvent:
        """一般的な質問に対してフレンドリーに回答する"""
        prompt = f"あなたは親切なアシスタントです。以下の質問に優しく答えてください:\n{ev.query}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=response.text)

    @step()
    async def handle_technical_query(self, ev: TechnicalQueryEvent) -> StopEvent:
        """技術的な質問に対して専門的な視点で詳細に解説する"""
        prompt = f"あなたは優秀なシニアエンジニアです。以下の技術的課題に専門用語を用いて詳細に回答してください:\n{ev.query}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=response.text)

# 3. ワークフローの実行関数
async def main():
    workflow = SupportAgentWorkflow(timeout=30)
    
    # テスト1: テクニカルな質問
    tech_query = "Pythonのasyncioを使って並行処理を実装するときのデッドロック対策を教えて"
    print(f"\n--- 実行クエリ: {tech_query} ---")
    tech_result = await workflow.run(query=tech_query)
    print(f"回答:\n{tech_result}")
    
    # テスト2: 一般的な質問
    general_query = "こんにちは！今日の天気が良いので、おすすめの散歩の過ごし方を教えて。"
    print(f"\n--- 実行クエリ: {general_query} ---")
    general_result = await workflow.run(query=general_query)
    print(f"回答:\n{general_result}")

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

コードの解説

@step() で各メソッドを自動的にステップ登録しています。ステップがどのイベントを引数にとり、どのイベントを返すかで制御の流れが決まります。
StartEvent から始まり、最終的に StopEvent(result=...) が返されるとワークフローは終了し、その result が呼び出し元に返却されます。
async（非同期処理）をネイティブサポートしているため、LLM APIを効率的に待機させることが可能です。

5. 開発で陥りがちなエラーとトラブルシューティング

LlamaIndex Workflowsの実装において、開発者がよく直面する代表的なエラーとその原因・解決方法を解説します。

5.1 `WorkflowValidationError` またはイベントの無限ループ問題

【現象】 プログラムを実行した際、処理が途中でフリーズしてタイムアウトしたり、「適切な次のステップが見つかりません」といった検証エラーが発生する。

【原因】 ステップが定義した戻り値（Event）と、次に受け取るべきステップの引数（Event）が正しくマッピングされていない、もしくは循環参照が起きていることが原因です。

【対策】

@step() を持つ各関数の「引数の型ヒント」と「戻り値の型ヒント」が、定義した Event クラスと一致しているか厳密にチェックしてください。
型安全をより担保したい場合は、Pydanticベースのイベントクラスでフィールド検証を行うと良いでしょう。型検証のテクニックは Pydanticを用いた堅牢な開発ガイドも参考になります。

5.2 非同期処理（`asyncio`）のコンテキストエラー

【現象】 RuntimeError: no running event loop や、Jupyter Notebookなどで実行した際に非同期ループが重複しているというエラーが発生する。

【原因】 LlamaIndex Workflowsはネイティブで非同期（Async）設計になっており、ステップ内での await 処理を前提としています。同期的（Sync）なコードの中に組み込もうとしたり、Jupyter環境などの特殊なイベントループ上で不適切な呼び出し方をするとエラーになります。