LlamaIndex Workflows実装ガイド【2026年最新】

「LangGraphとCrewAIは一通り触ったけど、もっと薄いレイヤーでイベント駆動の制御フローだけ組みたい」——AIエージェントの設計を相談されるとき、こういう声を聞くことが増えてきました。RAGの構築で使われることが多いLlamaIndexですが、実はエージェントの制御フローだけを担う軽量な独立パッケージ「LlamaIndex Workflows」を持っています。

LlamaIndex Workflowsは、2025年6月に本体のllama_indexパッケージから独立し、それ以降も高頻度でアップデートが続いている、非同期・イベント駆動のオーケストレーションフレームワークです。RAGの一機能ではなく、AIエージェント・ドキュメント処理パイプライン・マルチエージェントシステムまで、汎用的な制御フローを型安全に書けるツールとして育っています。

この記事では、公式ドキュメントの仕様に基づいて、最小構成のセットアップから、並列実行・Human-in-the-Loop・型付きState管理という3つの実践パターンまでコード付きで整理します。LangGraphやCrewAIとの位置づけの違いも、確認できた事実の範囲でまとめました。

そもそもLlamaIndex Workflowsとは何か

LlamaIndex Workflowsは、@stepデコレータで定義した関数（ステップ）どうしを、PydanticベースのEventクラスでつなぐことで、複雑な制御フローを組み立てるフレームワークです。各ステップは「どのEventを受け取り、どのEventを返すか」を型ヒントとして書くだけで、フレームワーク側がその型シグネチャからルーティングを自動推論します。

項目	内容
パッケージ名	llama-index-workflows（Python）/ @llamaindex/workflow-core（TypeScript）
独立パッケージ化	2025年6月、llama_index本体から分離。LlamaIndexエコシステムの外でも単体で使える汎用フレームワークとして再設計された
設計思想	async-first・イベント駆動。ルーティング、並列処理、ループ、複数ステップ間の状態管理を1つの仕組みで扱う
想定用途	AIエージェント、ドキュメント処理パイプライン、マルチモーダルアプリ、リサーチアシスタント、マルチエージェントシステム

独立パッケージ化から現在まで、PyPI上だけで50以上のバージョンが継続的にリリースされており、2026年6月末時点の最新版は2.22系まで進んでいます。RAG専用ツールという印象を持たれがちですが、制御フロー部分だけを切り出して他のLLMクライアントと組み合わせて使うこともできます。

セットアップと最小構成（5分で動かす）

まずはインストールです。LlamaIndex本体は不要で、Workflowsパッケージ単体で動きます。今回はOpenAIのLLM統合もあわせて使うため、それぞれインストールします。

# 動作環境: Python 3.10+
pip install llama-index-workflows llama-index-llms-openai

# 認証（OpenAIのAPIキー）
export OPENAI_API_KEY=your-api-key

最小構成として、問い合わせへの一次回答を作り、その回答をレビューして返す2ステップのワークフローを書いてみます。StartEventとStopEventは開始と終了を表す特別なEventで、自分で定義する必要はありません。

# 動作環境: Python 3.10+, llama-index-workflows, llama-index-llms-openai
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent
from llama_index.llms.openai import OpenAI


class DraftEvent(Event):
    draft: str


class SupportReplyFlow(Workflow):
    llm = OpenAI(model="gpt-4.1")

    @step
    async def draft_reply(self, ev: StartEvent) -> DraftEvent:
        prompt = f"次の問い合わせに丁寧な一次回答を書いてください: {ev.inquiry}"
        response = await self.llm.acomplete(prompt)
        return DraftEvent(draft=str(response))

    @step
    async def review_reply(self, ev: DraftEvent) -> StopEvent:
        prompt = f"次の回答文をトーン・正確性の観点でレビューし、改善版を返してください:n{ev.draft}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))


flow = SupportReplyFlow(timeout=60, verbose=False)
result = await flow.run(inquiry="返品したいのですが、送料はどちらが負担ですか？")
print(str(result))

draft_replyがDraftEventを返すと、その型を受け取るreview_replyが自動的に呼び出されます。ステップを増やすときも、間に新しいEventクラスを挟むだけで直列のパイプラインを継ぎ足せます。

まず試したい実践パターン3選

1. 並列実行（Fan-out / Fan-in）

独立したサブタスクを並列化したい場合は、1つのステップから複数のEventを返し、@step(num_workers=N)で並列度を指定します。ここでは5件のタスクを最大5並列で処理し、全件そろってから結果をまとめる例です。

# 動作環境: Python 3.10+, llama-index-workflows
import asyncio
import random
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class Task(Event):
    n: int


class Done(Event):
    n: int


class ParallelResearchFlow(Workflow):
    @step
    async def fan_out(self, ev: StartEvent) -> list[Task]:
        return [Task(n=i) for i in range(5)]

    @step(num_workers=5)
    async def work(self, ev: Task) -> Done:
        # 実際にはここでURL取得やLLM呼び出しなどのI/Oを行う
        await asyncio.sleep(random.uniform(0, 1))
        return Done(n=ev.n)

    @step
    async def join(self, events: list[Done]) -> StopEvent:
        return StopEvent(result=sorted(e.n for e in events))

workステップの戻り値の型がlist[Done]を受け取るjoinステップにまとめて渡されるため、5件のタスクを待ち合わせてから次に進む「ファンイン」が自然に書けます。動的に生成したイベント数を待ち合わせたい場合は、Context.send_eventで個別にイベントを送り、Context.collect_eventsで受け取り件数がそろうまで待つ書き方も用意されています。

2. Human-in-the-Loop（承認フロー）

請求書送信や外部への一括メール配信など、AIの出力をそのまま実行せず人間の承認を挟みたい場面では、InputRequiredEventとHumanResponseEventを使います。ワークフローはこのイベントを送出した時点で一時停止し、応答が来るまで待機します。

# 動作環境: Python 3.10+, llama-index-workflows
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent, InputRequiredEvent, HumanResponseEvent


class ApprovalFlow(Workflow):
    @step
    async def ask_approval(self, ev: StartEvent) -> InputRequiredEvent:
        return InputRequiredEvent(prefix=f"この請求書を送信していいですか？ (yes/no): {ev.summary}")

    @step
    async def send_or_cancel(self, ev: HumanResponseEvent) -> StopEvent:
        if ev.response.strip().lower() == "yes":
            # ここで実際の送信処理を呼び出す
            return StopEvent(result="送信しました")
        return StopEvent(result="承認されなかったため送信をキャンセルしました")


workflow = ApprovalFlow()
handler = workflow.run(summary="A社向け 108,000円")
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        response = input(event.prefix)
        await handler.send_event(HumanResponseEvent(response=response))
final_result = await handler
print(final_result)

公式ドキュメントは、1ステップ内で完結させるctx.wait_for_event()よりも、この例のように「送出用のステップ」と「応答を受け取るステップ」を分ける書き方を推奨しています。理由は、ワークフローが失敗時に途中から再実行（リプレイ）される際、1ステップ完結型だと再実行の扱いが複雑になるためです。承認待ちが長時間に及ぶ場合は、ContextをシリアライズしてDBやファイルに保存し、別プロセスから再開する使い方も可能です。

3. 型安全なState管理

複数ステップにまたがる状態（会話ターン数、収集済みデータなど）は、PydanticモデルをContext[モデル名]として型付けすることで、State管理そのものをバリデーション付きにできます。

# 動作環境: Python 3.10+, llama-index-workflows
from pydantic import BaseModel, Field
from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class MyEvent(Event):
    msg: list[str]


class RunState(BaseModel):
    num_runs: int = Field(default=0)


class RepeatFlow(Workflow):
    @step
    async def start(self, ctx: Context[RunState], ev: StartEvent) -> MyEvent:
        async with ctx.store.edit_state() as state:
            state.num_runs += 1
        return MyEvent(msg=[ev.input_msg] * state.num_runs)

    @step
    async def process(self, ctx: Context[RunState], ev: MyEvent) -> StopEvent:
        data_length = len("".join(ev.msg))
        return StopEvent(result=f"{len(ev.msg)}回処理、文字数合計{data_length}")

ポイントは、RunStateの各フィールドに必ずデフォルト値を設定しておくことです。デフォルトがあることで、Contextオブジェクトが初回アクセス時に状態を自動初期化できます。状態の書き換えはasync with ctx.store.edit_state()のブロック内でまとめて行うことで、並列実行時の競合を避ける設計になっています。

制御フローは”3つのアプローチ”で考える

アプローチ	向いている場面	使うAPI	難易度
直列パイプライン	ドラフト作成→レビューのような順番が固定の処理	@stepの型ヒントだけでルーティング	低
並列（Fan-out/Fan-in）	複数ソースの並行リサーチ、独立したチェックの一括実行	num_workers、send_event、collect_events	中
対話型（Human-in-the-Loop）	送金・送信など、実行前に人間の承認が必要な処理	InputRequiredEvent／HumanResponseEvent	中〜高

LangGraph・CrewAIと何が違うのか

Python向けのエージェントオーケストレーションには複数の選択肢があり、それぞれ設計思想が異なります。LlamaIndex Workflowsを検討する際に押さえておきたい違いを整理しました。

項目	LlamaIndex Workflows	LangGraph	CrewAI
設計の軸	Eventでつなぐ軽量なステップ関数	状態を持つグラフ（ノード・エッジ）	役割ベースのマルチエージェント「クルー」
独立パッケージ化・大型マイルストーン	2025年6月に1.0として独立、以降も継続的にリリース	2025年10月22日にv1.0が正式GA	2026年6月、v1.14でメモリ・ナレッジ・RAGのバックエンドをプラガブル化
型付き状態管理	Context[PydanticModel]で状態全体を型付け	State定義（TypedDict/Pydanticスキーマ）をグラフに紐付け	タスクの構造化出力をPydanticで指定
観測性	llama-index-instrumentation経由でOpenTelemetry・Arize Phoenixに対応	LangSmithとの統合が中心	実行イベント（finish_reason等）をLLMイベントとして公開

3つとも「どれか1つが優れている」という単純な話ではなく、すでにLlamaIndexでRAGパイプラインを組んでいるなら地続きで導入しやすく、状態遷移が複雑な長時間ワークフローを厳密に管理したいならLangGraphの状態マシン的な設計が向いている、役割分担が明確な複数エージェントの協調にはCrewAIが書きやすい、という住み分けです。すでにAIエージェントの実装全体をどう設計するか迷っている場合は、AIエージェントFW5強徹底比較｜選定指針もあわせて参考にしてください。

【要注意】よくある失敗パターンと回避策

失敗1：並列ステップのnum_workersを指定し忘れる

❌ @stepだけで並列処理のつもりで書いてしまう
⭕ 並列実行させたいステップには必ず@step(num_workers=N)を明示する。指定を忘れると、Eventが順番に1件ずつ処理され、意図せず直列実行になる

失敗2：Stateモデルのフィールドにデフォルト値を設定しない

❌ RunStateのフィールドをnum_runs: intのように必須項目にしてしまう
⭕ Contextは初回アクセス時に状態を自動初期化するため、フィールドには必ずField(default=...)を設定しておく。忘れるとバリデーションエラーで実行が止まる

失敗3：Human-in-the-Loopを1ステップのwait_for_event()だけで組む

❌ 待機処理を1つのステップ内に詰め込む
⭕ 公式推奨は、イベントを送出するステップと応答を受け取るステップを分けるパターン。ワークフローが途中失敗から再実行される際、1ステップ完結型は再実行時の扱いが複雑になりやすい

失敗4：ステップの戻り値の型ヒントを省略する

❌ 戻り値の型を書かず、辞書やタプルをそのまま返す
⭕ Workflowsはステップの型ヒントを見てルーティングを自動推論する仕組みのため、型ヒントを省略すると次のステップに正しくつながらない。各ステップの入出力は必ずEventのサブクラスで明示する

よくある質問

Q1. LlamaIndex Workflowsを使うのに、LlamaIndex本体（RAG機能）は必要ですか？

不要です。Workflowsはllama_index本体から独立したパッケージとして提供されており、pip install llama-index-workflowsだけで単体インストールできます。RAG機能を使わず、制御フローの仕組みだけを流用することもできます。

Q2. TypeScriptでも使えますか？

使えます。npm i @llamaindex/workflow-coreでインストールでき、Python版と同様にイベント駆動のステップ定義が可能です。

Q3. 実行状況を監視・可視化する方法はありますか？

llama-index-instrumentationを追加インストールすると、OpenTelemetryやArize Phoenixとの連携によるオブザーバビリティが有効になります。本番運用では、ステップの失敗・リトライ・所要時間をこうした外部ツールで監視することが推奨されています。

Q4. LangGraphやCrewAIと比べて、どちらを最初に学ぶべきですか？

すでに複雑な状態遷移を伴う長時間ワークフローの設計に慣れているならLangGraph、役割分担された複数エージェントの協調を素早く組みたいならCrewAIが書きやすい場面が多いです。LlamaIndex WorkflowsはEventと型ヒントだけで完結する分、学習コストが低く、既存のPythonの非同期処理に近い感覚で書けるのが特徴です。

参考・出典

• Announcing Workflows 1.0: a lightweight framework for agentic systems — LlamaIndex公式ブログ（参照日: 2026-07-09）
• Workflows Introduction — LlamaIndex Developer Documentation（参照日: 2026-07-09）
• Concurrent execution of workflows — LlamaIndex Developer Documentation（参照日: 2026-07-09）
• Human in the Loop — LlamaIndex Developer Documentation（参照日: 2026-07-09）
• llama-index-workflows — PyPI（参照日: 2026-07-09）
• LangGraph 1.0 is now generally available — LangChain Changelog（2025年10月22日公開、参照日: 2026-07-09）

まとめ：今日から始める3つのアクション

1. 今日やること：pip install llama-index-workflowsで、2ステップの直列ワークフローを1つ動かしてみる
2. 今週中：現状LangGraphやCrewAIで組んでいる処理のうち、独立して並列化できるサブタスクを洗い出し、num_workers付きのステップに切り出せるか検討する
3. 今月中：送金・送信など「実行前に人が確認すべき処理」を洗い出し、Human-in-the-Loopパターンで承認ステップを組み込む

あわせて読みたい:

• Human-in-the-Loop完全ガイド｜エージェント承認設計【2026】 — 承認フロー設計をフレームワーク非依存でさらに深掘り
• AIエージェントFW5強徹底比較｜選定指針 — LangGraph・CrewAIを含む主要フレームワークの選び方

---

著者：佐藤傑（さとう・すぐる）
株式会社Uravation代表取締役。X（@SuguruKun_ai）フォロワー10万人超。100社以上の企業向けAI研修・導入支援。著書累計2.5万部突破。