エージェント状態機械とフォールトトレラントオーケストレーション：信頼性の高いエージェントの構築

はじめに：なぜエージェントにステートマシンが必要なのか

複雑なエージェントを構築する際、私たちはしばしば誤った考えに陥ります。それは、エージェントをステートレスなブラックボックスと見なし、毎回ゼロから始めることです。しかし、実際のビジネスでは、エージェントは多段階の推論、ツール呼び出し、ユーザー確認などのインタラクションを必要とすることがよくあります。外部呼び出しの失敗（APIタイムアウトやツールエラーなど）が一度でも発生すると、会話チェーン全体が崩壊する可能性があります。ステートマシンは、このための構造化されたソリューションを提供します。エージェントのライフサイクルを有限の状態（idle、thinking、tool_call、respondingなど）に分割し、状態間の正当な遷移を定義します。状態を明示的に管理することで、エージェントの各ステップの動作を正確に制御できるだけでなく、異常発生時に安全な状態へロールバックしたり、エラー情報をユーザーにフィードバックしたりすることができます。

この記事では、DeepSeek APIの実践経験に基づき、エージェント向けステートマシンの設計方法を深く掘り下げ、フォールトトレラントなオーケストレーション戦略と組み合わせて、不安定なネットワークやツール障害に直面しても安定して動作するエージェントを構築します。原理から始めて、ステートマシンを備えたエージェントフレームワークを段階的に実装し、エンジニアリングで遭遇した落とし穴とその解決策を共有します。基本的なエージェント開発にすでに精通しているなら、このチュートリアルはシステムの堅牢性と保守性を向上させるのに役立ちます。

ステートマシンの核心要素：状態、イベント、遷移

ステートマシンは、状態、イベント、遷移の3つの要素で構成されます。エージェントの文脈では、状態はidle（ユーザー入力を待機）、processing（処理中）、tool_call（ツール結果を待機）、error（エラー発生）などです。イベントは状態の変化を引き起こします。例えば、user_message、api_response、tool_timeoutなどです。遷移は、現在の状態で特定のイベントを受信したときに、エージェントが実行すべきアクションと次に遷移すべき状態を定義します。

ステートマシンを設計する際、遷移ルールをビジネスロジックにハードコードするのではなく、設定ファイルで定義することを強くお勧めします。その利点は、可読性が高く、拡張が容易で、視覚化も簡単なことです。以下は、Pythonの辞書に基づくステートマシン定義の例で、各状態がイベントに対してどのように応答するかを明確に示しています。

STATE_MACHINE = {
    "idle": {
        "user_message": {"action": "handle_user", "next": "processing"},
        "error": {"action": "notify_user", "next": "idle"}
    },
    "processing": {
        "llm_done": {"action": "generate_reply", "next": "responding"},
        "llm_error": {"action": "retry_llm", "next": "processing"},
        "tool_required": {"action": "call_tool", "next": "tool_call"}
    },
    "tool_call": {
        "tool_success": {"action": "notify_llm", "next": "processing"},
        "tool_error": {"action": "handle_tool_error", "next": "error"},
        "tool_timeout": {"action": "abort_tool", "next": "error"}
    },
    "responding": {
        "message_sent": {"action": "reset", "next": "idle"},
        "message_failed": {"action": "retry_send", "next": "responding"}
    },
    "error": {
        "user_retry": {"action": "reset", "next": "idle"}
    }
}

このステートマシンは簡略化されたモデルですが、フォールトトレランスを状態遷移に組み込む方法を示しています。例えば、processing状態でLLM呼び出しが失敗した場合、retry_llmアクションをトリガーし、同じ状態に留まることで、直接エラー状態に入るのではなく、再試行の機会を提供します。tool_call状態では、タイムアウトはエラーとして扱われ、error状態に遷移し、ユーザーの決定を待ちます。

DeepSeek APIの統合と状態管理

DeepSeekはOpenAI互換のAPIを提供しており、base_urlはhttps://api.deepseek.com、モデルはdeepseek-chatです。実際のコーディングでは、ステートマシンにDeepSeek呼び出しを統合する必要があります。一般的な落とし穴は、APIキーの漏洩や設定ミスによる呼び出し失敗です。そのため、ステートマシンの初期化時にAPI設定を検証し、明確なエラーメッセージを提供する必要があります。

以下のコードは、ステートマシンのprocessing状態でDeepSeek APIを呼び出す方法を示しています。タイムアウトと再試行をより細かく制御するために、requestsライブラリを直接使用します。

import requests
import json

def call_deepseek(messages, max_retries=2, timeout=30):
    headers = {
        "Authorization": "Bearer your-deepseek-api-key",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "deepseek-chat",
        "messages": messages,
        "temperature": 0.7
    }
    response = None
    for attempt in range(max_retries):
        try:
            response = requests.post(
                "https://api.deepseek.com/chat/completions",
                headers=headers,
                json=payload,
                timeout=timeout
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.Timeout as e:
            print(f"Attempt {attempt+1} timed out: {e}")
        except requests.exceptions.ConnectionError as e:
            print(f"Attempt {attempt+1} connection error: {e}")
        except requests.exceptio
ns.HTTPError as e:
            code = response.status_code if response else None
            if code in (429, 500, 502, 503):
                print(f"Attempt {attempt+1} HTTP {code}, retrying...")
            else:
                raise e
    # 最終失敗、エラーメッセージを返す
    return {"error": "DeepSeek API 呼び出し失敗", "detail": str(e)}
re>
注意点として、シンプルなリトライ機構を実装し、タイムアウトや接続エラーをキャッチしています。ステートマシンでは、戻り値に error キーが含まれている場合、llm_error イベントをトリガーし、リトライまたはエラー処理ロジックに入るべきです。複数回リトライしても失敗した場合は、ユーザーに通知し、error 状態に入ります。
フォールトトレラントなオーケストレーション：リトライ、タイムアウト、およびデグラデーション戦略
フォールトトレラントなオーケストレーションは、単にAPI呼び出しにリトライを追加するだけでなく、さまざまな障害に対処するための戦略を体系的に設計する必要があります。外部依存関係の安定性に基づいて、3つのレベルに分類できます。レベル1は一時的な障害（ネットワークの揺らぎなど）で、通常はリトライで解決できます。レベル2は部分的な障害（特定のモデルが利用できないなど）で、バックアップモデルやバックアップAPIにデグラデーションすることを検討できます。レベル3は長期的な障害（APIキーの無効化など）で、手動介入が必要です。
私のエンジニアリング実践では、ステートマシンの動作決定を導くための障害分類表を作成しました。以下の表に、一般的な障害タイプと推奨される戦略を示します。
障害タイプ
例
戦略
状態遷移
一時的なタイムアウト
LLM呼び出しのタイムアウト
指数バックオフで2回リトライ
processing → processing
レート制限（429）
リクエストが多すぎる
待機後にリトライ、または低速キューにデグラデーション
processing → processing
モデルが利用不可
404を返す
deepseek-chat-0712などの別のモデルにデグラデーション
processing → processing
APIキーが無効
401 Unauthorized
リトライを停止し、ユーザーに設定確認を通知
processing → error
ツール障害
サードパーティAPIがエラーを返す
1回リトライし、失敗した場合は部分的な結果を提供
tool_call → tool_call
重要な原則は、リトライは無限ループではないということです。各状態には最大リトライ回数があり、それを超えた場合はエラー状態に遷移しなければなりません。そうしないと、リソースの浪費や無限ループを引き起こします。さらに、リトライ時に指数バックオフを使用すると、サービスへの負荷を軽減できます。例えば、最初は1秒待ち、次は2秒、その次は4秒待ちます。
エンジニアリングの落とし穴：ステートマシンにおけるコンテキストの喪失と回復
ステートマシンを実装する際に直面した最大の落とし穴は、コンテキストの喪失です。エージェントがマルチターンの会話を処理する場合、会話履歴（メッセージ）をステートマシンのコンテキストに保存する必要がありますが、異常な遷移やサービス再起動時に永続化されていないと、すべてのコンテキストが失われ、ユーザーは問題を再説明しなければならなくなります。
解決策は、ステートマシンとコンテキストを一緒にデータベースやファイルにシリアライズすることです。例えば、状態オブジェクトをJSONに変換し、現在の状態、未送信のメッセージ履歴、必要な一時変数を含めます。各遷移後に永続化します。これにより、プロセスがクラッシュしても、前回の状態から再開できます。以下は状態の永続化の例です。
def save_state(session_id, state_machine, context):
    state_snapshot = {
        "session_id": session_id,
        "state": context.state,
        "messages": context.messages,
        "data": context.data
    }
    with open(f"sessions/{session_id}.json", "w") as f:
        json.dump(state_snapshot, f)

def load_state(session_id):
    try:
        with open(f"sessions/{session_id}.json", "r") as f:
            snapshot = json.load(f)
        # コンテキストを復元
        ctx = Context()
        ctx.state = snapshot["state"]
        ctx.messages = snapshot["messages"]
        ctx.data = snapshot["data"]
        return ctx
    except FileNotFoundError:
        return None
さらに、ステートマシンでは、イベントトリガーに並行性の問題が発生する可能性があります。例えば、ユーザーが複数のメッセージをすばやく送信する場合です。ロックを追加するか、一意のsession_idを使用して分離する必要があります。分散環境を使用する場合は、Redisなどの分散ロックを推奨します。私のプロジェクトでは、単純にファイルロックを使用しましたが、本番環境ではより堅牢なソリューションを採用することをお勧めします。
実際のケース分析：ツール呼び出し失敗後のデグラデーション
実際のケースとして、天気予報エージェントを構築する際に、サードパーティの天気APIに依存していました。ある時、そのAPIが500エラーを数時間返しました。私たちのステートマシンはtool_call状態でエラーをキャッチし、まず2回リトライしましたが、両方失敗しました。その後、デグラデーション戦略をトリガーしました：バックアップの天気データソース（静的キャッシュや別の無料APIなど）を使用しますが、バックアップも失敗した場合は、「天気サービスは一時的に利用できません」とユーザーに伝え、後でリトライできるように会話コンテキストを保持します。
このプロセスでは、ステートマシンはtool_call → tool_call（リトライ）→ tool_call（デグラデーション）→ tool_call（成功）という遷移を経験しました。重要なのは、デグラデーション中にユーザーのリクエストコンテキストを失わず、エラー情報をログ分析用に記録したことです。最終的にユーザーはクエリを完了できましたが、体験はやや低下しましたが、システム全体はクラッシュしませんでした。
デグラデーションの信頼性を確保するために、各外部呼び出しに対して独立したデグラデーション関数を設定し、それらの戻り値の構造の一貫性をテストしました。ステートマシンでは、イベントパラメータを介してデグラデーションフラグを渡すことで、ビジネスロジックが現在どのデータソースを使用しているかを明確に把握できます。
パフォーマンス最適化：ステートマシンのタイムアウトと並行性制御
高並行性のシナリオでは、ステートマシンのパフォーマンス最適化が重要です。一般的な問題は、タイムアウトなしの待機です。例えば、LLM呼び出しにタイムアウトを設定しないと、スレッドが永久にブロックされます。API呼び出しには適切なタイムアウトを設定し、ステートマシンの単一フロー全体にも合計タイムアウト制限（例：10秒）を設定し、超過した場合は強制的にエラー状態に遷移させる必要があります。
並行性制御に関しては、Pythonのasyncioを使用して複数のセッションを非同期に処理できます。各セッションのステートマシンは独立して実行できますが、データベース接続プールなどの共有リソースへのアクセスに注意する必要があります。私の実装では、シングルスレッドのイベントループを採用し、各セッションに1つのタスクを割り当て、ロックの複雑さを回避しましたが、前提として外部呼び出しは非ブロッキング（非同期ライブラリを使用するか、スレッドプールで実行）である必要があります。
以下は、concurrent.futuresを使用してステートマシンの処理時間を制限する簡単なタイムアウト制御の例です。
from concurrent.futures import ThreadPoolExecutor, TimeoutError

def run_state_machine_with_timeout(session_id, input, timeout_seconds=15):
    executor 
= ThreadPoolExecutor(max_workers=1)
    future = executor.submit(process_event, session_id, input)
    try:
        result = future.result(timeout=timeout_seconds)
        return result
    except TimeoutError:
        # 強制的にエラー状態に切り替える
        context = load_state(session_id)
        if context:
            context.state = "error"
            save_state(session_id, context)
        return {"error": "処理タイムアウト"}