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

このチュートリアルでは、推論に Gemini API を使用し、Temporal を永続性に使用する、耐久性のある AI エージェントを構築する手順について説明します。Temporal の組み込み Gemini SDK 統合を使用します。

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

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

アーキテクチャ

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

• ワークフロー: 単一の generate_content 呼び出し。Gemini SDK の自動関数呼び出し（AFC）ループはワークフロー内で実行され、Temporal はそのすべてのステップを永続化します。
• アクティビティ: Temporal が永続化する個々の作業単位。Gemini API 呼び出しは自動的にアクティビティになります。
• ワーカー: ワークフローとアクティビティを実行するプロセス。API キーが存在する唯一の場所です。

この例では、これら 3 つの要素をすべて 1 つのファイル（durable_agent_worker.py）に配置します。実際の環境では、さまざまなデプロイとスケーラビリティのメリットを得るために、これらを分離します。Temporal CLI を使用してエージェントにプロンプトを指定するため、クライアント コードを記述する必要はありません。

前提条件

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

• Gemini API キー。 Google AI Studio で無料で作成できます。
• Python バージョン 3.10 以降。
• uv は依存関係 管理用です。
• ローカル 開発用サーバーの実行とワークフローの開始に使用する Temporal CLI。

設定

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

temporal server start-dev

次に、プロジェクトを作成して必要な依存関係をインストールします。

uv init durable-gemini-agent
cd durable-gemini-agent
uv add "temporalio[google-genai]" httpx python-dotenv

uv は仮想環境を作成して管理するため、このチュートリアルの以降の Python コマンドはすべて uv run を介して実行されます。

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 のワークフロー サンドボックスに httpx を無制限に通過させるように指示します。httpx をインポートすると class _CookieCompatRequest(urllib.request.Request) が実行され、サンドボックスは stdlib クラスのサブクラス化をブロックします。

ツールは httpx を使用し、activity_as_tool() は、Gemini がシグネチャからスキーマを派生できるように、ワークフローでこれらのツール関数をインポートする必要があります。そのため、ファイルを分割しても httpx はサンドボックスに到達します。ツールを独自のモジュールに移動しても回避できません。

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import httpx

ここで google.genai を一覧表示する必要はありません。後で構成する Temporal プラグインは、pydantic_core と annotated_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.
"""

ツール定義

次に、エージェントが使用できるツールを定義します。各ツールは通常の Temporal アクティビティです。型アノテーション付きのパラメータと説明的な docstring を持つ、@activity.defn で装飾された非同期関数です。Gemini はそのシグネチャと docstring から関数宣言を構築するため、Args セクションで各パラメータを記述します。

import json

from temporalio import activity

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

@activity.defn
async def get_weather_alerts(state: str) -> str:
    """Get weather alerts for a US state.

    Args:
        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/{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())

次に、IP アドレスのジオロケーションのツールを定義します。

@activity.defn
async def get_ip_address() -> str:
    """Get the public IP address of the current machine."""
    async with httpx.AsyncClient() as client:
        response = await client.get("https://icanhazip.com")
        response.raise_for_status()
        return response.text.strip()

@activity.defn
async def get_location_info(ipaddress: str) -> str:
    """Get the location information for an IP address including city, state, and country.

    Args:
        ipaddress: An IP address to look up
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://ip-api.com/json/{ipaddress}")
        response.raise_for_status()
        result = response.json()
        return f"{result['city']}, {result['regionName']}, {result['country']}"

これがツールレイヤ全体です。ツール レジストリ、FunctionDeclaration 構造、ディスパッチ テーブルはありません。次のセクションでは、これらのアクティビティを activity_as_tool() でラップします。これにより、各パラメータがアクティビティに位置的に渡されます。パラメータが 0 個、1 個、または複数のツールはすべて機能します。

エージェントのワークフロー

これで、エージェントの構築を完了するためのすべての要素が揃いました。AgentWorkflow クラスは 1 回の generate_content 呼び出しを行います。TemporalAsyncClient は、すべての API 呼び出しが Temporal アクティビティとして実行されるドロップイン AsyncClient であり、 activity_as_tool() は各アクティビティを Gemini ツールに変換します。

モデルがツールをリクエストすると、ワークフロー内で実行されている SDK の AFC ループは、workflow.execute_activity を介してツールをディスパッチし、会話に結果を追加して、モデルを再度呼び出します。このループがエージェントであり、各ステップは Temporal のイベント履歴に記録されたアクティビティであるため、耐久性があります。

from datetime import timedelta

from google.genai import types
from temporalio.contrib.google_genai import TemporalAsyncClient, activity_as_tool
from temporalio.workflow import ActivityConfig

TOOL_CONFIG = ActivityConfig(start_to_close_timeout=timedelta(seconds=30))

@workflow.defn
class AgentWorkflow:
    """Agent workflow that uses Gemini for LLM calls and executes tools."""

    @workflow.run
    async def run(self, prompt: str) -> str:
        client = TemporalAsyncClient()

        response = await client.models.generate_content(
            model="gemini-3.8-flash",
            contents=prompt,
            config=types.GenerateContentConfig(
                system_instruction=SYSTEM_INSTRUCTIONS,
                tools=[
                    activity_as_tool(get_weather_alerts, activity_config=TOOL_CONFIG),
                    activity_as_tool(get_ip_address, activity_config=TOOL_CONFIG),
                    activity_as_tool(get_location_info, activity_config=TOOL_CONFIG),
                ],
            ),
        )

        # Leave this in place. You will un-comment it during a durability
        # test later on.
        # await workflow.sleep(timedelta(seconds=10))

        return response.text or ""

注意点

• ワークフロー内 で TemporalAsyncClient を構築します。認証情報は含まれていません。API 呼び出しをアクティビティ呼び出しに変換する方法のみを認識します。
• activity_config は start_to_close_timeout または schedule_to_close_timeout を設定する必要があります。Temporal にはタイムアウトが必要ですが、ツール アクティビティのデフォルトはありません。
• Gemini API アクティビティのデフォルトは 60 秒の start_to_close_timeout です。 モデル呼び出しに時間がかかる場合は、TemporalAsyncClient(activity_config=...) でオーバーライドします。

エージェントは完全に耐久性があります。ワーカーが数回ターンした後にクラッシュした場合、Temporal は、すでに実行された LLM 呼び出しやツール呼び出しを再呼び出しすることなく、中断したところから正確に再開します。

再試行数

再試行は Temporal が所有しているため、Gemini SDK 独自の再試行ループを有効にしないでください。代わりに、アクティビティ構成で retry_policy を使用して再試行動作を設定します。

from temporalio.common import RetryPolicy

TOOL_CONFIG = ActivityConfig(
    start_to_close_timeout=timedelta(seconds=30),
    retry_policy=RetryPolicy(maximum_attempts=3),
)

API の失敗も分類されます。一時的なステータス（408、429、5xx）は再試行可能であるため、アクティビティの再試行ポリシーが適用されます。その他のステータス（不正なリクエストに対する 400 など）は再試行できないため、解決しないエラーで試行回数を使い果たすのではなく、ワークフローはすぐに失敗します。

この分類を拡張できます。統合では、各 API の失敗が ApplicationError として表示されます。そのタイプは Gemini 例外クラス名（4xx の場合は ClientError、5xx の場合は ServerError）です。そのため、non_retryable_error_types に名前をリストすると、一時的なセットから除外されます。たとえば、Gemini 側の停止の再試行を停止し、最初の 5xx でワークフローを失敗させるには、TemporalAsyncClient を使用して Gemini API アクティビティにポリシーを適用します。

from temporalio.common import RetryPolicy

client = TemporalAsyncClient(
    activity_config=ActivityConfig(
        start_to_close_timeout=timedelta(seconds=60),
        retry_policy=RetryPolicy(
            maximum_attempts=5,
            non_retryable_error_types=["ServerError"],
        ),
    ),
)

ワーカーの起動

最後に、すべてを接続します。Temporal ワーカーは Temporal サービスに接続し、ワークフローとアクティビティのタスクのスケジューラとして機能します。

ここで、API キーを使用して実際の genai.Client が作成されます。 GoogleGenAIPlugin はそのクライアントを受け取り、Gemini API アクティビティを登録し、Pydantic データ コンバータをインストールして、ワークフロー サンドボックスを構成します。

import asyncio
import os

from dotenv import load_dotenv
from google import genai
from temporalio.client import Client
from temporalio.contrib.google_genai import GoogleGenAIPlugin
from temporalio.envconfig import ClientConfig
from temporalio.worker import Worker

async def main():
    gemini = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    plugin = GoogleGenAIPlugin(gemini)

    config = ClientConfig.load_client_connect_config()
    config.setdefault("target_host", "localhost:7233")
    client = await Client.connect(**config, plugins=[plugin])

    worker = Worker(
        client,
        task_queue="gemini-agent",
        workflows=[
            AgentWorkflow,
        ],
        activities=[
            get_weather_alerts,
            get_ip_address,
            get_location_info,
        ],
    )
    await worker.run()

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

このプラグインは、通常必要な 3 つのボイラープレートを削除します。

• data_converter=pydantic_data_converter はありません。プラグインは Pydantic ペイロード コンバータを自動的にインストールします。
• activity_executor=ThreadPoolExecutor はありません。すべてのアクティビティは非同期です。
• activities リストに Gemini アクティビティはありません。プラグインが登録します。独自のツールのみを登録します。

エージェントを実行する

これでエージェントは完成です。クライアントを記述する必要はありません。Temporal CLI でワークフローを開始できます。

まだ行っていない場合は、Temporal 開発用サーバーを起動します。

temporal server start-dev

新しいターミナル ウィンドウで、エージェント ワーカーを起動します。

uv run durable_agent_worker.py

3 つ目のターミナル ウィンドウで、エージェントにクエリを送信します。

temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"are there any weather alerts for where I am?"'

タスクキューに注目してください。これはワーカーがポーリングするキューと同じです。ワークフローを開始すると、ユーザー プロンプトを保持するワークフロー タスクがそのキューにディスパッチされ、エージェントが開始されます。ワークフローが完了するまで execute はブロックされ、結果が出力されます。待機したくない場合は、明示的な --workflow-id を指定して temporal workflow start を使用し、後で temporal workflow result -w your-workflow-id で結果を収集します。--workflow-id を省略すると、Temporal によってワークフロー ID が生成されます。

--input は JSON を受け取るため、裸の文字列プロンプトにはシェル引用符内に独自の引用符が必要です。CLI には Gemini API キーもデータ コンバータ構成も必要ありません。ワークフローの引数と戻り値はどちらもプレーン文字列であり、デフォルトの JSON ペイロード コンバータが処理します。

http://localhost:8233/namespaces/default/workflows で Temporal UI を開き、エージェント ループが展開されるのを確認します。gemini_api_client_async_request アクティビティ（モデルターンごとに 1 つ）が、ツール呼び出しごとに 1 つのアクティビティとインターリーブされ、それぞれに tool_call の概要が表示されます。そのインターリーブは AFC ループであり、耐久性と可観測性が向上しています。

いくつかのプロンプトを試して、エージェントが推論してツールを呼び出すことを確認します。各コマンドは、新しい --input を使用する以外は上記と同じです。

temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"are there any weather alerts for New York?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"where am I?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"what is my ip address?"'
temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
    --input '"tell me a joke"'

最後のプロンプトではツールは必要ないため、エージェントは SYSTEM_INSTRUCTIONS に基づいて俳句で応答します。

耐久性をテストする

Temporal を基盤に構築することで、エージェントは障害からシームレスに復旧できます。これは、2 つの異なるテストで確認できます。

ネットワーク停止をシミュレートする

このテストでは、パソコンのインターネット接続を一時的に無効にし、ワークフローを送信して、Temporal が自動的に再試行するのを確認してから、ネットワークを復元して復旧を確認します。

1. パソコンをインターネットから切断します（Wi-Fi をオフにするなど）。

2. ワークフローを送信します。

   temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
       --input '"tell me a joke"'

3. Temporal UI（http://localhost:8233）を確認します。Gemini API アクティビティが失敗し、Temporal がバックグラウンドで再試行を自動的に管理していることがわかります。

4. インターネットに再接続します。

5. 次の自動再試行で Gemini API に正常に到達し、ターミナルに最終結果が出力されます。

ワーカーのクラッシュから復旧する

このテストでは、実行中にワーカーを強制終了して再起動します。Temporal はワークフロー履歴（イベント ソーシング）を再生し、最後に完了したアクティビティから再開します。すでに完了した LLM 呼び出しとツール呼び出しは繰り返されません。

1. ワーカーを強制終了する時間を確保するため、durable_agent_worker.py を開き、AgentWorkflow.run の耐久性のあるタイマーのコメントを解除します。

   await workflow.sleep(timedelta(seconds=10))
   

   workflow.sleep はローカル タイマーではなく、Temporal タイマーです。履歴に記録され、再起動後も存続するため、このテストは信頼できます。

2. ワーカーを再起動します。

   uv run durable_agent_worker.py

3. 複数のツールをトリガーするクエリを送信します。

   temporal workflow execute --type AgentWorkflow --task-queue gemini-agent \
       --input '"are there any weather alerts where I am?"'

4. ツール呼び出しが完了してタイマーが実行されたら、ワーカー プロセスを強制終了します（ワーカー ターミナルで Ctrl-C、バックグラウンドで実行している場合は kill %1）。

5. ワーカーを再起動します。

   uv run durable_agent_worker.py

Temporal はワークフロー履歴を再生します。すでに完了した LLM 呼び出しとツール呼び出しは再実行されません。結果は履歴（イベントログ）から即座に再生され、タイマーが再開され、ワークフローが正常に完了します。

さらに便利なサービス

この統合は、このチュートリアルで説明する以上の機能をサポートしています。詳細については、 プラグインのドキュメント をご覧ください。

• ストリーミング。通常どおり generate_content_stream を使用します。外部コンシューマー（チャット UI）がワークフローの実行中にチャンクをリアルタイムで確認できるようにするには、TemporalAsyncClient(streaming_topic=...) を設定し、ワークフローで WorkflowStream をホストします。
• MCP。GoogleGenAIPlugin(mcp_servers={...}) を使用して、クライアントサイドの MCP サーバーをワーカーに登録し、ワークフローで TemporalMcpClientSession を使用して名前で参照します。ツールの検出と呼び出しは、プールされたワーカー側の接続に対してアクティビティとして実行されます。
• Vertex AI。ワーカー側の genai.Client とワークフロー側の TemporalAsyncClient の両方に vertexai=True を渡し、ワークフロー側で project と location を明示的に設定して、再生が決定論的になるようにします。

その他のリソース

• Temporal のドキュメント
• Temporal Python SDK
• Temporal Gemini SDK 統合
• Google GenAI SDK