AI ワークフローエンジン実践：スクリプトからビジュアルオーケストレーションへ — AI開発チュートリアル

一、なぜワークフローエンジンが必要なのか：スクリプトのジレンマから始まる

私が初めてPythonスクリプトでDeepSeek APIを呼び出したとき、コードは数十行だけで、スムーズに動作しました。しかし、ビジネスが複雑になるにつれて（例えば、マルチターン対話、ナレッジベース検索、結果検証、エラーリトライが必要になる）、スクリプトは制御不能になり始めました。すべてのif-else分岐、すべてのtry-exceptがコードを保守困難にし、並列実行や可視化モニタリングは言うまでもありません。多くの開発者が同様の経験をしていると思います。スクリプトはローカルでは問題なく動作しますが、本番環境にデプロイすると、多様な入力や突発的なAPIエラーに直面して脆弱になります。

ワークフローエンジンの核心的な価値は、「プロセス」を「コード」から切り離すことにあります。タスク間の依存関係、分岐、結合を宣言的に定義でき、エンジンがスケジューリング、状態管理、フォールトトレランスを担当します。これは、手書きSQLからORMへの移行、素の関数からマイクロサービスオーケストレーションへの移行に似ています。AIアプリケーションにとって、ワークフローエンジンは特に重要です。なぜなら、大規模言語モデルAPIの呼び出しには通常、ネットワーク遅延、コスト管理、結果の不確実性が伴い、これらはすべて細かいプロセス管理を必要とするからです。

この記事では、実践的な観点から、単純なDeepSeek呼び出しスクリプトから、イベント駆動型でDAGベースのワークフローエンジンへの進化を段階的に示します。その過程で遭遇した落とし穴を共有し、実行可能なコード例を提供します。

二、出発点：素朴なDeepSeek呼び出しスクリプト

まず、最も基本的なスクリプトから始めましょう。製品説明を入力し、DeepSeekにマーケティングコピーを生成させ、キーワードを抽出するという要件があるとします。PythonでDeepSeek APIを直接呼び出すコードは次のとおりです。

import requests
import json

def call_deepseek(prompt, api_key="your-deepseek-api-key"):
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    payload = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.7
    }
    response = requests.post("https://api.deepseek.com/chat/completions", headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["choices"][0]["message"]["content"]

# ビジネスロジック
description = "ポータブルスマートスピーカー、音声アシスタント対応、内蔵バッテリー、連続再生12時間。"

# マーケティングコピーを生成
prompt1 = f"以下の製品の魅力的なマーケティングコピーを書いてください：{description}"
copy = call_deepseek(prompt1)

# キーワードを抽出
prompt2 = f"以下のテキストから3〜5個のキーワードを抽出してください：{copy}"
keywords = call_deepseek(prompt2)

print("マーケティングコピー:", copy)
print("キーワード:", keywords)

このスクリプトには明らかに2つの欠点があります。1つ目は、2回のAPI呼び出しが直列であることです。キーワード抽出がコピーに依存しないなら並列化できますが、ここでは依存しているため順次実行するしかありません。2つ目は、エラーハンドリングがないことです。ネットワークの変動やAPIのレート制限が発生すると、スクリプト全体がクラッシュします。もちろん、try-exceptやリトライを追加することもできますが、すべてのタスクに書く必要があり、コードはすぐに冗長になります。

三、ジレンマの悪化：マルチタスクオーケストレーションと状態管理

実際のビジネスでは、タスクは2つ以上になることがよくあります。例えば、製品説明の感情分析、コピー生成、キーワード抽出、英語への翻訳、さらにはコンテンツのコンプライアンスチェックが必要になるかもしれません。これらのタスクに依存関係はあるでしょうか？感情分析とキーワード抽出は独立しており、並列化できます。コピー生成は説明に依存し、翻訳はコピーに依存します。手動でスクリプトを書く場合、マルチスレッドや非同期を使うかもしれませんが、スレッド間のデータ受け渡し、結果の集約、例外処理がコードを複雑にします。

さらに重要なのは、各タスクの実行状態や所要時間を視覚的に監視できないことです。あるとき、本番環境でAPIのタイムアウトが断続的に発生しましたが、どのタスクが原因か全く分からず、推測するしかありませんでした。これが、よりエレガントな解決策を求めるきっかけとなりました。

四、ワークフローへの移行：タスク抽象化とグラフモデル

ワークフローエンジンの核心的な考え方は、各ステップを「ノード」として抽象化し、ノード間の依存関係を「エッジ」で表すことです。プロセス全体は有向非巡回グラフ（DAG）です。各ノードは関数、API呼び出し、さらにはサブワークフローにすることができます。エンジンはグラフを走査し、トポロジカル順に実行し、コンテキストオブジェクトを介してデータを渡します。

私は最初、PythonでシンプルなDAG実行器を実装し、ノード定義、依存関係宣言、結果出力をサポートしました。以下は簡略化した実装です。

from dataclasses import dataclass
from typing import Callable, Any, Dict
import asyncio

@dataclass
class WorkflowNode:
    name: str
    func: Callable
    depends_on: list

class Workflow:
    def __init__(self):
        self.nodes = {}
        self.results = {}

    def add_node(self, name, func, depends_on=None):
        self.nodes[name] = WorkflowNode(name, func, depends_on or [])

    async def execute(self):
        # 簡略化したトポロジカルソート。循環なし、順序が有効と仮定
        for node in self.nodes.values():
            # 依存関係の完了を待つ
            for dep in node.depends_on:
                while dep not in self.results:
                    await asyncio.sleep(0.1)  # 単純なポーリング
            # ノードを実行
            inputs = {dep: self.results[dep] for dep in node.depends_on}
            self.results[node.name] = await node.func(**inputs)
        return self.results

この実装は粗末ですが機能します。asyncioを使用して非同期にし、依存関係をポーリングで待ちます。実際のエンジニアリングでは、Airflow、Prefect、Temporalなどのより成熟したワークフローフレームワークを使用します。これらは十分なスケジューリング、リトライ、監視機能を提供します。しかし、AIワークフローでは、動的分岐（LLMの出力に基づいて後続フローを決定する）、人間参加型（手動レビューが必要）など、特別なサポートが必要になることがよくあります。

五、エンジニアリング実践：イベント駆動型AIワークフロー

本番環境では、最終的にPrefectをエンジンとして選択しました。Pythonベースでカスタマイズが容易で、非同期とイベントトリガーをネイティブにサポートしているからです。私のアーキテクチャは、各AI呼び出しをPrefectタスクとしてカプセル化し、タスク間でパラメータを渡すというものです。例えば、generate_copyというタスクを定義し、DeepSeek APIを呼び出します。extract_keywordsタスクはコピーを入力として受け取ります。

しかし、Prefectのデフォルトのスケジューリングはフローポーリングであり、応答速度が十分ではありません。そこで、イベント駆動に切り替えました。メッセージキュー（Redis Streamsなど）を介して新しいタスクをエンジンにプッシュし、エンジンが対応するフローをトリガーします。これにより、各ユーザーリクエストは独立したフローインスタンスとなり、互いに影響しません。同時に、フローのチェックポイント（状態、結果）をデータベースに保存し、UIで表示できるようにしました。

ここで重要なエンジニアリングの落とし穴があります。APIの冪等性とリトライ戦略です。DeepSeek APIは時々429（レート制限）や5xxを返すため、ワークフローで指数バックオフリトライを実装する必要があります。しかし、リトライは重複実行を引き起こす可能性があり、タスクに副作用（メール送信など）がある場合は冪等性を実装する必要があります。私の方法は、各タスクにグローバルに一意のIDを割り当て、実行結果を記録し、リトライ前に成功しているかどうかをチェックすることです。

六、可視化の価値：プロセスを透明で制御可能にする

スクリプトからビジュアルオーケストレーションへの移行で最大の利点は透明性です。Prefect UIや自作のフロントエンドを通じて、各フローの実行状態、所要時間、入出力をリアルタイムで確認でき、失敗したノードを手動で再実行することもできます。これはAI生成の品質問題のデバッグに特に重要です。

例えば、あるユーザーが不適切なコピーを報告したとします。スクリプト時代には、フロー全体を再実行するしかなく、プロンプトの問題なのかモデルの温度なのかを特定できませんでした。ワークフローがあれば、そのノードの具体的な入力とパラメータを確認し、問題を再現し、プロンプトや温度を調整できます。この能力はAIアプリケーション開発において非常に貴重です。LLMの出力は非決定的であり、問題を特定するために可観測性が必要だからです。

さらに、ビジュアルオーケストレーションはチームコラボレーションを促進します。私の同僚（プログラミングの深いバックグラウンドがない人）もDAGエディタを使用してワークフローロジックを変更できます。例えば、ノードの順序を調整したり、新しい処理ステップを追加したりできます。これにより、AIアプリケーションの敷居が大幅に下がります。

七、実践的なケース分析：完全なAIワークフロー

次に、実際のケースを共有します。私たちはクライアント向けに「スマートカスタマーサービスのチケット分析」システムを構築しました。フローは次のとおりです。

イベントリスニング：新しいチケットを受信。
ユーザー意図分類（DeepSeek分類器）。
感情分析（DeepSeek感情モデル）。
ナレッジベースマッチング：分類に基づいてベクトルデータベースを照会。
返信ドラフト生成（DeepSeek生成器）。
手動レビュー（イベント駆動で一時停止）。
返信送信。

このフローでは、意図分類と感情分析は並列化できます。ナレッジベースマッチングは分類結果に依存し、ドラフト生成はマッチングと感情に依存します。各ステップをPrefectタスクとして定義し、Redis Streamsを使用してフローインスタンスをトリガーします。主要なコードは次のとおりです（簡略化）。

from prefect import flow, task, get_run_logger
from prefect.tasks import exponential_backoff

@task(retries=3, retry_delay_seconds=exponential_backoff(backoff_factor=2))
def sandbox_analysis(desc: str):
    # DeepSeek感情分析を呼び出す
    ...

@task
async def kb_match(category: str):
    # ベクトルデータベース照会
    ...

@flow
async def process_ticket(ticket_id: str):
    logger = get_run_logger()
    ticket = fetch_ticket(ticket_id)
    cat_task = classify_async.submit(ticket.desc)
    senti_task = sentiment_async.submit(ticket.desc)
    cat, senti = await cat_task.result(), await senti_task.result()
    kb_results = await kb_match.submit(cat).result()
    draft = await generate_draft.submit(ticket.desc, cat, senti, kb_results).result()
    logger.info(f"Draft ready for {ticket_id}")
    # 手動レビューのために一時停止
    await wait_for_review(ticket_id, draft)
    send_reply(ticket_id, draft)

このフローはワークフローエンジンで明確なDAG表現を持ち、各ノードにはログと可観測性があります。

八、エンジニアリングの落とし穴と解決策のまとめ

スクリプトからワークフローへの移行中に、多くの課題に直面しました。ここでいくつか要点をまとめます。読者が回避できるように。

依存関係の欠如：一部のサードパーティライブラリ（prefectなど）はPythonバージョンと互換性がない場合があります。仮想環境を使用し、バージョンを固定することをお勧めします。
非同期実行の罠：Prefectの一部のタスクはデフォルトで同期です。同期タスクで非同期関数を呼び出すとイベントループをブロックします。asyncio.run()を使用するか、明示的に非同期タスクとして宣言する必要があります。
データのシリアライゼーション：ノード間で渡すデータはシリアライズ可能である必要があります。DeepSeek APIが返すJSONは安全ですが、カスタムオブジェクトの場合は辞書に変換するか、クラウドストレージを使用する必要があります。
モニタリングとアラート：ワークフローエンジンはUIを提供しますが、Prometheus + Grafanaと統合して、タスクのレイテンシ、成功率を記録し、アラートを設定することをお勧めします。そうしないと、フローが詰まったときに最後に知ることになるかもしれません。
コスト管理：AIワークフローでは、トークン消費が主要なコストです。ノードレベルでトークン使用量を記録し、ワークフローの設定を介してモデルやサンプリングパラメータを動的に調整し、高コストのタスクが無駄にリトライされないようにすることをお勧めします。

九、まとめと展望

単純なスクリプトからビジュアルオーケストレーションへの進化は、技術スタックのアップグレードだけでなく、思考様式の転換でもあります。ワークフローエンジンにより、AIアプリケーションを生産ラインとして捉え、各ステップを制御可能、テスト可能、最適化可能にできます。実践では、特にAI呼び出しの経験があるチームには、早い段階でワークフロー思考を導入することを強くお勧めします。そのエンジニアリング上の利点は計り知れません。

将来、AIワークフローエンジンはよりインテリジェントになるでしょう。例えば、プロンプトを適応的に調整したり、類似した結果を自動的にキャッシュしたり、例外時に動的にダウングレード（より安価なモデルに切り替えるなど）したりする機能です。DeepSeekエコシステムも豊富なAPIとモデルを提供しており、ワークフローでこれらの機能を組み合わせて強力なシステムを構築できます。この記事が皆さんの参考になれば幸いです。コメントで実践的な経験を共有してください。