0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kai_kou(甲斐 甲)

LlamaIndex Workflowsで並行実行、実測すると本当に半分の時間だった

0
Posted at

はじめに

エージェントの処理を「調査ステップ」と「下書きステップ」のように分けて書くとき、両方とも独立して実行できるなら同時に走らせたいと考えるはずです。ところが多くのエージェントフレームワークは、ステップを定義する API はあっても「本当に並行で動くのか」までは明示しません。ドキュメントのサンプルコードは大抵ステップが1本道で、並行処理の挙動は読者が自分で確かめるしかありません。

LlamaIndex から独立した Python パッケージ llama-index-workflows は、イベント駆動でステップを繋ぐ軽量フレームワークです。今回はこのパッケージを実際にインストールし、複数ステップを同時に走らせた場合の壁時計時間と、ステップ内で例外を起こした場合の挙動を実測で確認しました。

この記事でわかること

  • llama-index-workflows の Event / Step / Context という3つの基本要素と最小実装
  • 2つの1秒処理を並行ステップとして実行し、実測した合計時間
  • ctx.collect_events() で複数イベントの到着を待ち合わせる際の内部的な呼び出し回数
  • ステップ内で例外が起きたとき、フレームワーク側の例外にラップされるかどうか

対象読者

  • Python でエージェントのステップ処理を自作している方
  • LangGraph や Google ADK 以外の軽量な選択肢を探している方
  • イベント駆動ワークフローの並行実行を実際の数値で確認したい方

前提環境

  • Python: 3.11.15
  • llama-index-workflows: 2.22.2（pip show llama-index-workflows で確認）
  • 実行環境: Claude Cloud スケジュール実行（本記事執筆パイプライン自身のセッション。python3 -m venv で隔離した検証用環境を作成し実行）

TL;DR

  • llama-index-workflows は pip install llama-index-workflows で単独導入できます。llama-index-core にも同梱されていますが、単独パッケージ版は依存が最小限で済みます
  • Event（Pydantic モデル）・Step（@step デコレータ）・Context（状態管理）の3要素だけでワークフローを組めます
  • ctx.send_event() で2つのイベントを発火し ctx.collect_events() で待ち合わせると、受信側のステップは イベントごとに一度ずつ呼ばれ、揃うまでは None を返して待機する実装になっていました
  • 1秒 sleep する処理を2本並行実行させたところ、実測の壁時計時間は 1.008秒（逐次なら約2秒のはず）で、実際に並行実行されていることを確認しました
  • ステップ内で ValueError を送出すると、フレームワーク独自の例外にラップされず 生の ValueError がそのまま 呼び出し元に伝播しました

LlamaIndex Workflows の3要素

llama-index-workflows のワークフローは、Pydantic の Event を継承したデータ構造をステップ間で受け渡す形で組みます。公式ドキュメントが示す最小構成は次の3つだけです。

  • Event: Event を継承した Pydantic モデル。ステップ間でやり取りするデータの型
  • Step: @step デコレータを付けた非同期メソッド。特定の Event を受け取り、別の Event を返します
  • Context: ステップ間で状態を共有したり、複数イベントの到着を待ち合わせたりするための実行コンテキスト

インストールは次のコマンド一つで完了します。

pip install llama-index-workflows

pip show llama-index-workflows
# Name: llama-index-workflows
# Version: 2.22.2

llama-index-core を使っている場合は llama_index.core.workflow として同じ機能にアクセスできますが、今回は依存を絞るために単独パッケージを使いました。

検証1: 複数イベントを待ち合わせる collect_events の呼び出し回数

まず、1つのステップから2種類のイベントを発火し、別のステップで両方が揃うのを待つパターンを試しました。

from workflows import Workflow, step, Context
from workflows.events import StartEvent, StopEvent, Event


class ResearchDoneEvent(Event):
    findings: list[str]


class DraftDoneEvent(Event):
    draft: str


class ParallelWorkflow(Workflow):
    @step
    async def start(self, ctx: Context, ev: StartEvent) -> ResearchDoneEvent | DraftDoneEvent:
        ctx.send_event(ResearchDoneEvent(findings=["fact-A", "fact-B"]))
        ctx.send_event(DraftDoneEvent(draft="draft-v1"))
        return None

    @step
    async def combine(
        self, ctx: Context, ev: ResearchDoneEvent | DraftDoneEvent
    ) -> StopEvent | None:
        data = ctx.collect_events(ev, [ResearchDoneEvent, DraftDoneEvent])
        if data is None:
            return None
        research, draft = data
        return StopEvent(result={"findings": research.findings, "draft": draft.draft})

verbose=True で実行すると、内部の tick ログが出力されます。

[tick] add: StartEvent()
[start:0] started from StartEvent
[start:0] complete with no result
[tick] add: ResearchDoneEvent(findings=['fact-A', 'fact-B'])
[combine:0] started from ResearchDoneEvent
[combine:0] complete with no result
[tick] add: DraftDoneEvent(draft='draft-v1')
[combine:0] started from DraftDoneEvent
[result] StopEvent(result={'findings': ['fact-A', 'fact-B'], 'draft': 'draft-v1'})
[combine:0] complete with StopEvent
RESULT: {'findings': ['fact-A', 'fact-B'], 'draft': 'draft-v1'}

ここで分かるのは、combine ステップは1回だけ呼ばれるのではなく、イベントが届くたびに毎回起動している という点です。1回目（ResearchDoneEvent のみ到着）は collect_events が None を返すため何もせず終了し、2回目（DraftDoneEvent も到着）で初めて両方が揃い StopEvent を返します。公式ドキュメントのサンプルは1つのイベントを受けて1つの結果を返す単純な例が中心で、複数イベントを待ち合わせる際にステップが複数回呼ばれる挙動までは明示されていませんでした。この実装を前提にすると、collect_events より前に副作用（DB書き込みなど）を置くと、揃うまでの間に複数回実行されてしまう点に注意が必要です。

検証2: 並行実行は実際に速くなるのか

次に、fan_out → task_a / task_b（各1秒 sleep）→ join という構成にして、壁時計時間を計測しました。

import time
import asyncio
from workflows import Workflow, step, Context
from workflows.events import StartEvent, StopEvent, Event


class TaskAEvent(Event):
    pass


class TaskBEvent(Event):
    pass


class DoneAEvent(Event):
    elapsed: float


class DoneBEvent(Event):
    elapsed: float


class TimingWorkflow(Workflow):
    @step
    async def fan_out(self, ctx: Context, ev: StartEvent) -> TaskAEvent | TaskBEvent:
        ctx.send_event(TaskAEvent())
        ctx.send_event(TaskBEvent())
        return None

    @step
    async def task_a(self, ctx: Context, ev: TaskAEvent) -> DoneAEvent:
        t0 = time.monotonic()
        await asyncio.sleep(1.0)
        return DoneAEvent(elapsed=time.monotonic() - t0)

    @step
    async def task_b(self, ctx: Context, ev: TaskBEvent) -> DoneBEvent:
        t0 = time.monotonic()
        await asyncio.sleep(1.0)
        return DoneBEvent(elapsed=time.monotonic() - t0)

    @step
    async def join(self, ctx: Context, ev: DoneAEvent | DoneBEvent) -> StopEvent | None:
        data = ctx.collect_events(ev, [DoneAEvent, DoneBEvent])
        if data is None:
            return None
        a, b = data
        return StopEvent(result={"a_elapsed": a.elapsed, "b_elapsed": b.elapsed})

実行結果は次のとおりです。

RESULT: {'a_elapsed': 1.0016058539999904, 'b_elapsed': 1.0017060250000327}
WALL_TOTAL_SEC: 1.008

task_a と task_b はそれぞれ内部で約1.0016秒・1.0017秒かかっていますが、ワークフロー全体の壁時計時間は 1.008秒でした。2つのステップを逐次実行していれば合計は2秒に近づくはずなので、@step で定義した非同期ステップが asyncio のイベントループ上で実際に並行実行されていることを実測で確認できました。ドキュメントの「async-first」という説明は、コードの見た目が非同期というだけでなく、実行時間にもそのまま反映される設計だと分かります。

検証3: ステップ内の例外はどう伝播するか

最後に、ステップ内で意図的に例外を発生させ、呼び出し元にどう届くかを確認しました。

from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent
from workflows.errors import WorkflowRuntimeError


class FlakyWorkflow(Workflow):
    @step
    async def boom(self, ev: StartEvent) -> StopEvent:
        raise ValueError("simulated failure in step")


wf = FlakyWorkflow(timeout=5)
try:
    await wf.run()
except Exception as e:
    print(f"CAUGHT: {type(e).__name__}: {e}")
    print(f"IS_WORKFLOW_RUNTIME_ERROR: {isinstance(e, WorkflowRuntimeError)}")

CAUGHT: ValueError: simulated failure in step
IS_WORKFLOW_RUNTIME_ERROR: False

ステップ内で送出した ValueError は、フレームワーク側の WorkflowRuntimeError にラップされず、そのままの型で呼び出し元に伝播しました。エラーハンドリングを書く際は、ワークフロー共通の例外クラスだけを except で捕まえるのではなく、各ステップが投げうる個別の例外も想定して書く必要があります。パッケージには retry_policy や catch_error といったモジュールも用意されているため、リトライや個別のエラー処理を入れたい場合はそちらを使う構成になっているとみられます。

著者視点の発見ポイント

公式のPyPIパッケージ説明文とGitHubリポジトリでは「event-driven, async-first, step-based」という3つの形容詞で設計思想が説明されていますが、実際にコードを書いて動かすまでは、この3つの言葉がどこまで実装に反映されているのか判断がつきませんでした。今回の検証で、「async-first」は単なるキーワードではなく、asyncio.sleep を使った2つのステップの壁時計時間が実測でほぼ1秒に収まるという具体的な数値に表れていることを確認できました。同時に、複数イベントを待ち合わせる collect_events がステップを複数回呼び出す実装になっている点と、ステップ内の例外がラップされずに素通りする点は、公式ドキュメントのミニマルなサンプルコードだけを読んでいては気づきにくい部分です。並行実行の速度だけを見て採用を決めるのではなく、こうした細部の挙動を実際に動かして確認してから設計に組み込むほうが、後から手戻りが少なくなります。

まとめ

  • llama-index-workflows は pip install 一発で導入でき、Event / Step / Context の3要素で組める軽量なワークフローフレームワークです
  • 並行ステップの壁時計時間は実測で約1.008秒となり、async-first という設計思想が実際の実行速度に反映されていることを確認しました
  • collect_events による複数イベントの待ち合わせはステップが複数回呼ばれる実装のため、副作用の配置には注意が必要です
  • ステップ内の例外は独自クラスにラップされず生のまま伝播するため、エラーハンドリングは個別の例外型を想定して書く必要があります

次のステップとして、resource によるリソース注入や retry_policy を使ったリトライ処理も実測で確認します。

参考リンク

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?