ワークフローとは？

ワークフローは、イベント駆動型でステップベースのアプローチであり、アプリケーションの実行フローを制御・管理する手法です。この方式では、アプリケーションが独立したが互いに関連する複数の部分に分割され、それぞれの部分を「ステップ」と呼びます。各ステップは特定のイベントによって起動され、実行中に他のイベントをトリガーし、さらに次のステップを起動することがあります。

これらのステップとイベントを適切に設計・組み合わせることで、複雑な処理フローを構築し、ビジネスロジックをカプセル化できます。

入門編

まず始めに、Python環境がインストールされていることを確認してください。本記事ではPython 3.10を使用していますが、プロジェクト要件に合致するバージョンであれば他のバージョンでも問題ありません。次に、新しいPythonプロジェクトを作成し、段階的に学習していきます。依存関係が他のプロジェクトと競合しないように、仮想環境を使用します。

仮想環境のセットアップ

• プロジェクトディレクトリでターミナルを開き、以下のコマンドで仮想環境を作成：
  python -m venv venv
• Windowsでは、以下のコマンドで仮想環境を有効化：
  .\\venv\\Scripts\\activate
• 依存パッケージリストの保存：
  pip freeze > requirements.txt
• 仮想環境から抜ける：
  deactivate

VSCode設定

開発にはVSCodeを使用していますが、他のIDEでも同様に設定可能です。仮想環境を有効にしても、デフォルトでは実行時にシステムのPythonが使用されるため、正しく実行されるように設定が必要です。

1. Ctrl + Shift + P (Windows/Linux) または Cmd + Shift + P (Mac) を押してコマンドパネルを開く
2. Python: Select Interpreter を選択
3. 表示された一覧から仮想環境内のPythonインタープリターを選択

Hello World ワークフロー

最初の準備として、以下の2つのライブラリをインストールします：

pip install llama-index-core
pip install llama-index-utils-workflow

以下のコードを貼り付けて実行すると、基本的なワークフローが動作します。この例では「Hello, World!」を出力します：

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step,
)

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result="Hello, world!")

async def main():
    w = MyWorkflow(timeout=10, verbose=False)
    result = await w.run()
    print(result)

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

このコードでは、@step アノテーションを持つメソッドを定義し、それらを実行することでワークフローの流れを制御します。重要なポイントとして、StartEvent から始まり、StopEvent で終了する必要があります。これによりワークフローの整合性が保たれます。途中でステップがスキップされるとエラーが発生します。

複数イベントの扱い

イベントの流れをより詳しく見てみましょう。以下の例では、カスタムイベントを2つ定義し、順番に実行させています：

from llama_index.core.workflow import (
    StartEvent,
    StopEvent,
    Workflow,
    step,
    Event,
)

class FirstEvent(Event):
    first_output: str

class SecondEvent(Event):
    second_output: str

次にワークフロークラスでそれらを利用します：

class MyWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> FirstEvent:
        print(ev.first_input)
        return FirstEvent(first_output="First step complete.")

    @step
    async def step_two(self, ev: FirstEvent) -> SecondEvent:
        print(ev.first_output)
        return SecondEvent(second_output="Second step complete.")

    @step
    async def step_three(self, ev: SecondEvent) -> StopEvent:
        print(ev.second_output)
        return StopEvent(result="Workflow complete.")

async def main():
    w = MyWorkflow(timeout=10, verbose=False)
    result = await w.run(first_input="Start the workflow.")
    print(result)

この場合、ステップ2でStopEventを返すと、ステップ3が実行されず、エラーが発生します。ワークフローの構成は順序通りに進まなければならないため、意図しない終了は問題を引き起こします。

ワークフローの可視化

ワークフローの構造を可視化する機能はシンプルですが、呼び出しフローの理解に役立ちます。以下のように使用します：

from llama_index.utils.workflow import draw_all_possible_flows

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
    draw_all_possible_flows(MyWorkflow, filename="multi_step_workflow.html")

実行後、HTMLファイルが生成され、ワークフロー全体の構造を視覚的に確認できます。各ステップ実行時にはメソッド名とイベント名が出力され、最後に「done」が表示されます。

まとめ

本記事ではワークフローの基本概念と実装方法について紹介しました。イベント駆動型の仕組みにより、アプリケーションの実行フローを柔軟に管理できるようになりました。仮想環境の構築から基本的なワークフローの作成まで、Pythonによるマルチステップ処理の流れを体験しました。

今後の学習では、より高度なワークフロー構築方法について探求していく予定です。AI技術の応用においても、こうしたワークフローの活用が重要になるでしょう。