LlamaIndex Workflows：プロダクション品質のEvent-driven AI Agentを構築する – ITFROMZERO

Table of Contents

考え方を変える：固定的なPipelineから柔軟なEvent-drivenへ

基本的なRAGを実装する際、多くの方はSequential Pipeline（データが直線的に流れるモデル）に慣れているでしょう。通常のフローは「クエリを受け取る → ドキュメントを検索する → 回答を生成する」という形です。シンプルなタスクならこれで十分機能します。しかし、AgentがエラーをチェックしたりステップXを要件を満たすまで繰り返す必要がある場合、従来のPipelineはif-elseの「ぐちゃぐちゃ」なコードへとあっという間に変貌してしまいます。

私自身、LangChainの初期バージョンでルーティングロジックの管理だけに丸一週間費やしたことがあります。当時のコードはロジックが膨らむにつれて保守が非常に困難で、壊れやすいものでした。LlamaIndex Workflowsはこの問題を根本から解決します。「チェーン」から「イベント」へと発想を転換することで、Agentを直線的なフローに縛り付ける代わりに「イベントXが発生したらステップYを実行する」とシンプルに定義できます。

AI Agentの構築アプローチ比較

なぜWorkflowsが価値あるものなのか、以下の比較をご覧ください。

1. Linear Chains（線形チェーン）

実際：前進のみの製造ラインのようなもので、逆戻りはできません。
制限：コードを複雑にせずに前のステップへ戻ったり、複雑な分岐処理をしたりすることができません。

2. State Machines（DAG – LangGraph）

実際：状態（state）の管理に優れていますが、セットアップが非常に煩雑です。
制限：NodeとEdgeの定義により、システムが大きくなるとフロー図がクモの巣のようになります。

3. LlamaIndex Workflows（Event-driven）

実際：各ステップは完全に独立しており、イベント（events）を通じて通信します。
メリット：async/awaitとデコレーターのおかげでコードが非常に「Pythonic」です。既存のフロー全体を壊す心配なく機能の追加・削除ができます。

Workflowsはいつ使うべきか？

実際の本番環境での経験から、Workflowsが最も力を発揮する3つのシナリオを挙げます。

自己修正（Self-Correction）：Agentが書いたコードのユニットテストを自動実行し、エラーがあれば自動修正します（ループ処理）。
並列処理（Parallelism）：3〜4つの異なるソースから同時にデータを取得し、情報を素早く照合します。
人間の確認待ち（Human-in-the-loop）：Agentが上司にコンテンツ確認のメールを送り、承認が得られるまで次のステップを待機します。

詳細な実装ガイド

まず、完全なサポートを得るためにllama-index v0.10.20以降を使用していることを確認してください。

pip install llama-index llama-index-core

ステップ1：イベント（Events）の定義

Eventはデータを運ぶ手段です。Pydanticを使ってデータ構造を明示的に定義します。

from llama_index.core.workflow import Event

class SearchEvent(Event):
    query: str

class RefineEvent(Event):
    initial_answer: str
    context: str

ステップ2：Workflow構造の構築

@stepタグが付いた各関数は、Agent内の小さなmicro-serviceとして機能します。

from llama_index.core.workflow import Workflow, StartEvent, StopEvent, step
from llama_index.llms.openai import OpenAI

class ResearchAgent(Workflow):
    llm = OpenAI(model="gpt-4o")

    @step
    async def search_step(self, ev: StartEvent) -> SearchEvent:
        user_query = ev.get("query")
        print(f"データを検索中: {user_query}")
        return SearchEvent(query=user_query)

    @step
    async def process_step(self, ev: SearchEvent) -> StopEvent:
        # LLMからの結果をシミュレート
        result = f"{ev.query}の分析データ：AI市場は年率30%で成長しています。"
        return StopEvent(result=result)

ステップ3：実行

Workflowの実行は非常にシンプルです。run()関数を呼び出して結果を待つだけです。

async def main():
    agent = ResearchAgent(timeout=60, verbose=True)
    result = await agent.run(query="AIマーケット2025")
    print(f"結果: {result}")

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

高度なテクニック：自己反省メカニズム（Reflection）

本番環境でよく使うパターンの一つが、AgentにそのOutputを自己検査させることです。回答が基準を満たしていなければ、RefineEventを自動発行して再処理を要求します。

@step
async def critic_step(self, ev: ProcessEvent) -> RefineEvent | StopEvent:
    score = check_quality(ev.answer) 
    if score > 0.8:
        return StopEvent(result=ev.answer)
    return RefineEvent(feedback="情報がやや薄い。具体的な数値データが必要です。")

このテクニックをコード作成支援システムに適用したところ、初回実行の成功率が60%から85%以上に向上しました。

実戦で得たベストプラクティス

Context（ctx）の活用：Eventでデータを渡す以外に、ctxを使ってUserIDやSessionIDなどの共有変数を保存しましょう。
フローの可視化：workflow.draw("flow.html")を使いましょう。フローが10ステップを超えると、目で追ってデバッグしたくなくなります。
無限ループの防止：必ずtimeoutを設定しましょう。二つのEventが相互に呼び合い続けてAPIアカウントが大変なことになるのを防ぎます。
命名規則：Event1、Event2のような名前は避けましょう。ValidationFailedEventのように命名することで、ログを見ただけでAgentが直面している問題がすぐわかります。

まとめ

LlamaIndex Workflowsは単なる機能ではなく、プロフェッショナルなAIを構築するための新しい考え方です。ロジックを独立したステップに分離することで、システムの柔軟性と保守性が大幅に向上します。長々としたコードチェーンに行き詰まりを感じているなら、ぜひEvent-drivenアプローチに切り替えてみてください。コードのメンテナンスが格段に楽になるはずです！

Eventのルーティングで困っていたり、より複雑なシナリオのサンプルコードが必要な方は、コメントでお知らせください。サポートします！