Gemini と Temporal を使用した永続的な AI エージェント

このチュートリアルでは、推論に Gemini API を使用し、永続化に Temporal を使用する ReAct スタイルのエージェントループを構築する手順について説明します。このチュートリアルの完全なソースコードは GitHubで入手できます。

エージェントは、天気予報の検索や IP アドレスのジオロケーションなどのツールを呼び出すことができ、応答に必要な十分な情報が得られるまでループします。

一般的なエージェントデモとの違いは耐久性 です。すべての LLM 呼び出し、すべてのツール呼び出し、エージェントループのすべてのステップは Temporal によって永続化されます。プロセスがクラッシュした場合、ネットワークが切断された場合、API がタイムアウトした場合、Temporal は自動的に再試行し、最後に完了したステップから再開します。会話履歴が失われることも、ツール呼び出しが誤って繰り返されることもありません。

アーキテクチャ

このアーキテクチャは次の 3 つの部分で構成されます。

ワークフロー: 実行ロジックをオーケストレートするエージェントループ。
アクティビティ: Temporal が永続化する個々の作業単位（LLM 呼び出し、ツール呼び出し）。
ワーカー: ワークフローとアクティビティを実行するプロセス。

この例では、これら 3 つの要素をすべて 1 つのファイル（durable_agent_worker.py）に配置します。実際の環境では、さまざまなデプロイとスケーラビリティのメリットを得るために、これらを分離します。エージェントにプロンプトを提供するコードは、2 つ目のファイル（start_workflow.py）に配置します。

前提条件

このガイドを完了するには、次のものが必要です。

Gemini API キー。Google AI Studio で無料で作成できます。
Python バージョン 3.10 以降。
ローカル開発用サーバーを実行するための Temporal CLI。

設定

始める前に、 Temporal 開発用サーバーがローカルで実行されていることを確認してください。

temporal server start-dev

次に、必要な依存関係をインストールします。

pip install temporalio google-genai httpx pydantic python-dotenv

Gemini API キーを使用して、プロジェクトディレクトリに .env ファイルを作成します。API キーは Google AI Studioから取得できます。

echo "GOOGLE_API_KEY=your-api-key-here" > .env

実装

このチュートリアルの残りの部分では、durable_agent_worker.py を上から下まで順に説明し、エージェントを段階的に構築していきます。ファイルを作成して、手順に沿って進めてください。

インポートとサンドボックスの設定

最初に、事前に定義する必要があるインポートから始めます。workflow.unsafe.imports_passed_through() ブロックは、特定のモジュールを制限なく通過させるように Temporal のワークフローサンドボックスに指示します。これは、いくつかのライブラリ（特に urllib.request.Request をサブクラス化する httpx）が、サンドボックスでブロックされるパターンを使用しているため必要です。

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import pydantic_core  # noqa: F401
    import annotated_types  # noqa: F401

    import httpx
    from pydantic import BaseModel, Field
    from google import genai
    from google.genai import types

システム指示

次に、エージェントの個性を定義します。システムの指示は、モデルの動作方法を指定します。このエージェントは、ツールが必要ない場合は俳句で応答するように指示されています。

SYSTEM_INSTRUCTIONS = """
You are a helpful agent that can use tools to help the user.
You will be given an input from the user and a list of tools to use.
You may or may not need to use the tools to satisfy the user ask.
If no tools are needed, respond in haikus.
"""

ツール定義

次に、エージェントが使用できるツールを定義します。各ツールは、説明的なドキュメント文字列を持つ非同期関数です。パラメータを受け取るツールは、単一の引数として Pydantic モデルを使用します。これは Temporal のベストプラクティスであり、時間の経過とともにオプションのフィールドを追加してもアクティビティの署名を安定させることができます。

import json

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

class GetWeatherAlertsRequest(BaseModel):
    """Request model for getting weather alerts."""

    state: str = Field(description="Two-letter US state code (e.g. CA, NY)")

async def get_weather_alerts(request: GetWeatherAlertsRequest) -> str:
    """Get weather alerts for a US state.

    Args:
        request: The request object containing:
            - state: Two-letter US state code (e.g. CA, NY)
    """
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    url = f"{NWS_API_BASE}/alerts/active/area/{request.state}"

    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers, timeout=5.0)
        response.raise_for_status()
        return json.dumps(response.json())