大規模なドキュメント合成や長期間にわたる調査など、複数のステップからなるエージェント ワークフローをデプロイする場合、リアルタイム モデルでバックグラウンド エージェントを実行すると、インフラストラクチャに不要な負荷がかかり、リソースの枯渇（429）エラーが発生する可能性があります。
Gemini Enterprise Agent Platform には、遅延を許容できるワークロード専用に設計された、スループット最適化スケジューラである遅延階層が用意されています。スケジューラは、長時間実行される自律型タスクをライブチャット クエリと同じように緊急度の高いものとして扱うのではなく、複雑なマルチステップ エージェント ワークフローをオフピーク時にキューに入れて、成功率と全体的なスループットを高めることを目指します。
遅延階層を使用してリクエストを送信すると、API はタスクを非同期で受け入れ、すぐにインタラクション ID を返します。遅延階層には次の特徴があります。
割引料金: 標準リクエストと比較してモデル推論の料金が 50% 割引されるため、本番環境でのエージェント費用を管理できます。詳細は、 料金をご覧ください。
スループットの向上: 遅延階層では、負荷の高い非同期ワークロードをオフピーク時に移動することで、429 エラー（モデル容量 の制約）とレート制限を軽減し、リアルタイムの本番環境のニーズに対応できるように標準階層の割り当てを解放します。
完了タイムアウト: 遅延階層では、タスクの 95% を 24 時間以内に完了することを目指しています。この期間内にタスクが完了しない場合、タスクは期限切れとなり、
failed状態に移行します。キューで費やされる実際の時間は、現在のリージョン クラスタの容量と需要によって異なります。
ユースケース
遅延階層は、次の例のように、数時間のターンアラウンド時間を許容できるユースケースに適しています。
金融: 株式と市場の調査（毎日または毎週）。
法とコンプライアンス: 複数のドキュメントにわたる規制、M &A のデュー デリジェンス。
戦略: 継続的な競合他社のインテリジェンスとトレンドの統合。
セキュリティ: コードベースの脆弱性スキャンと修正。
サポートされているエージェント
Deep Research Agent の自律型エージェント スケジュールを構成できます。
遅延タスクを作成する
次の例は、Deep Research
タスク
を遅延階層で client.interactions.create() を使用して開始する方法を示しています。
import time
from google import genai
client = genai.Client(
enterprise=True,
project="PROJECT_ID",
location="global",
)
PROMPT = "Analyze the latest market trends in renewable energy storage."
DEEP_RESEARCH_AGENT = "deep-research-preview-04-2026"
interaction = client.interactions.create(
input=PROMPT,
agent=DEEP_RESEARCH_AGENT, # Agent identifier
service_tier="deferred", # Run on deferred tier for off-peak scheduling
background=True, # Return immediately instead of waiting for the answer
store=True, # Persist interaction state to poll or stream later
stream=False, # `stream` must be set to False during task creation
)
print(f"Interaction ID: {interaction.id}")
print(f"Status: {interaction.status}")
print(f"Service tier: {interaction.service_tier}")
このメソッドは、status="in_progress" と
service_tier="deferred" を返してすぐに終了します。
タスクの進行状況をモニタリング
オフピーク時の容量を待機している間、またはアクティブに実行されている間、インタラクションの status は in_progress のままになります。エージェントが計画、検索、分析の手順を実行すると、新しいアイテムが steps リストに追加されます。
インタラクションを定期的にポーリングするか、更新をストリーミングすることで、タスクのステータスをプログラムで追跡できます。
ポーリング
インタラクションがターミナル状態（completed、failed、cancelled）のいずれかに達するまで、インタラクションを定期的に（15 ～ 30 秒ごとなど）ポーリングします。
TERMINAL_STATES = ("completed", "failed", "cancelled")
POLL_INTERVAL_SECONDS = 15
TIMEOUT_MINUTES = 60
started = time.time()
deadline = started + TIMEOUT_MINUTES * 60
while True:
current = client.interactions.get(interaction.id)
elapsed = int(time.time() - started)
steps = getattr(current, "steps", None) or []
print(f"[{elapsed:>4}s] status={current.status} steps={len(steps)}")
if current.status in TERMINAL_STATES:
break
if time.time() >= deadline:
raise TimeoutError(
f"Still {current.status} after {TIMEOUT_MINUTES} min. The interaction "
"continues running server-side; re-run the check to resume polling."
)
time.sleep(POLL_INTERVAL_SECONDS)
print(f"\nFinished in {int(time.time() - started)}s with status={current.status}.")
ストリーミング
background=True と store=True の横に stream=True を設定すると、インタラクションが in_progress ステータスになったら、更新をリアルタイムでストリーミングできます。ストリームは、中間的な思考、テキストの差分、ステータスの更新などのイベントが発生すると、それらをプッシュします。
タスクが in_progress の間に接続が切断された場合は、client.interactions.get() を使用して stream=True でストリームに再接続し、最後に受信したイベント ID を last_event_id に渡すことができます。last_event_id を省略すると、API はすべてのイベントを最初から再生します。
INTERACTION_ID = interaction.id # from the create step
MAX_RECONNECTS = 5
STREAM_TIMEOUT = 300 # seconds
print(
f"streaming interaction: {INTERACTION_ID} (status={interaction.status})\n"
)
def render(event):
"""Prints one SSE event. Returns True once the interaction has finished."""
if event.event_type == "step.delta":
delta = event.delta
if delta.type == "text":
print(delta.text, end="", flush=True)
elif delta.type == "thought_summary":
summary = (getattr(delta.content, "text", "") or "").strip()
if summary:
print(f"\n[thinking] {summary[:200]}", flush=True)
elif delta.type.endswith("_call"):
queries = getattr(getattr(delta, "arguments", None), "queries", None)
print(
f"\n[{delta.type}] {', '.join(queries) if queries else ''}",
flush=True,
)
elif event.event_type == "interaction.status_update":
print(f"[status] {event.status}", flush=True)
elif event.event_type == "interaction.completed":
print(f"\n\n[status] {event.interaction.status}", flush=True)
return True
elif event.event_type == "error":
print(f"\n[error] {event.error.message}", flush=True)
return True
return False
last_event_id = None
finished = False
for attempt in range(MAX_RECONNECTS):
try:
# stream=True turns the GET into a live subscription. last_event_id=None on
# the first pass, so the server starts from the beginning of the run.
for event in client.interactions.get(
INTERACTION_ID,
stream=True,
last_event_id=last_event_id,
timeout=STREAM_TIMEOUT,
):
last_event_id = event.event_id or last_event_id
finished = render(event) or finished
except Exception as e: # pylint: disable=broad-except
# A dropped connection loses nothing: the run continues server-side and the
# next iteration reattaches from last_event_id.
print(f"\n[stream dropped: {type(e).__name__}] reattaching...", flush=True)
if finished:
break
# The server also closes the stream when the run ends, without an error.
if (
client.interactions.get(INTERACTION_ID, timeout=STREAM_TIMEOUT).status
!= "in_progress"
):
break
else:
print(f"\n[gave up after {MAX_RECONNECTS} reconnects]")
print(f"\n\nStreamed interaction: {INTERACTION_ID}")
タスクをキャンセルする
タスクのステータスが queued、in_progress、requires_action の場合は、タスクをキャンセルできます。タスクをキャンセルすると、ステータスが cancelled に移行します。
タスクをキャンセルするには、client.interactions.cancel() を使用します。
client.interactions.cancel(INTERACTION_ID)
最終出力とトークン使用量を取得する
インタラクションが completed 状態になると、完全なトランスクリプトが steps リストに表示されます。最終的な回答は、出力が生成された最後のステップのテキスト コンテンツです。
インタラクションは保存されるため（store=True）、任意のセッションのインタラクション ID を使用して、いつでも結果を取得できます。
def get_final_text(completed_interaction):
"""Returns the text of the last step that produced output."""
for step in reversed(getattr(completed_interaction, "steps", None) or []):
text = "".join(
part.text for part in (getattr(step, "content", None) or [])
if getattr(part, "text", None)
)
if text:
return text
return ""
final = client.interactions.get(interaction.id)
print(f"Status: {final.status}\n")
print(get_final_text(final) or "(No text output)")
if final.usage:
print(
f"\nToken usage:\n"
f" Input tokens: {final.usage.total_input_tokens}\n"
f" Output tokens: {final.usage.total_output_tokens}\n"
f" Total tokens: {final.usage.total_tokens}"
)