定期実行のジョブが二十数本まで増えたあたりで、私は自分のコードの「入り口」がいつの間にか四種類に分かれていることに気づきました。記事の下調べは generateContent を直接叩き、夜間の一括処理は Batch、App Store レビューの要約は自前で組んだエージェントループ、画像まわりは別のヘルパー。どれも当時は最短だった選び方なのですが、半年ぶりに一本のジョブを直そうとして、まずどの入り口を使っていたかを思い出すところから始める羽目になりました。
6月30日に Interactions API が一般提供となり、Gemini のモデルとエージェントを扱う主要な入り口がここへ寄せられました。Managed Agents、バックグラウンド実行、Gemini Omni も同じ入り口の下に並びます。これは派手な新機能というより、地味に、しかし長く効く種類の更新だと感じています。呼び出し口が一本化されるということは、半年後の自分が「どこから呼んでいたか」を思い出さなくてよくなる、ということだからです。
この記事は、新規に一本のスクリプトを書く話ではありません。すでに動いている散らばった呼び出し口を、止めずに、壊さずに、一つの正面玄関へ畳んでいくための移行設計です。動くアダプタ層のコードと、移行の順序、そして移行中だからこそ起きる事故の避け方までを扱います。
入り口が散らばると、何が高くつくのか
呼び出し口が増えること自体は、最初は痛みを生みません。痛むのは半年後です。具体的には三つの形で表に出てきます。
一つ目は、計装の重複です。トークン消費の記録、失敗時のリトライ、タイムアウトの扱いを、入り口ごとに少しずつ違う形で書いてしまう。私の場合、リトライ回数の上限が generateContent 経路では3回、Batch 経路では設定し忘れて無制限、という不揃いを後から見つけました。コストの異常に気づくのが遅れる典型です。
二つ目は、モデルの差し替えが一度で終わらないことです。gemini-flash-latest が 3.5 Flash の実体になったとき、私は四つの入り口を別々に直す必要がありました。一か所直すたびに、本当に全部直したか不安になる。これは数の問題ではなく、変更の影響範囲が見えないことの問題です。
三つ目は、新しい運用形態に乗り換えにくいことです。バックグラウンド実行で「投げて、終わったら受け取る」形にしたくても、入り口が散らばっていると、どの経路から書き換えればよいかの当たりがつきません。
一本化の本質的な利点は、これら三つが「一か所を直せば済む」状態に変わることです。Interactions API はその受け皿になりますが、いきなり全部を載せ替える必要はありません。間に薄い層を一枚挟むのが、私の見つけた最も安全なやり方でした。
正面玄関は API ではなく、自分のアダプタ層に置く
ここが、この記事で最もお伝えしたい判断です。一本化の正面玄関を Interactions API そのものに直接置くのではなく、自分が所有する薄いアダプタ層に置きます。
理由は単純で、API の細部は今後も変わるからです。6月6日にレガシーの outputs スキーマが削除されたように、スキーマや引数は廃止期限とともに動きます。アプリ側のコードが API の細部を直接握っていると、その変更のたびに全ジョブを触ることになります。間にアダプタを一枚挟んでおけば、変わるのはアダプタの内側だけです。
アダプタが提供するのは、たった一つの入り口です。「何をしてほしいか」を渡すと、「結果」が返る。その内側で Interactions API を呼びます。
# llm_gateway.py — 自分が所有する唯一の正面玄関
import os
import time
import uuid
import logging
from dataclasses import dataclass, field
from typing import Any
from google import genai # 実体の呼び出しはこの内側だけに閉じ込める
log = logging.getLogger("llm_gateway")
_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
# モデル選択を一か所に集約する。差し替えはここだけを直す。
MODEL_BY_TIER = {
"fast": "gemini-flash-latest", # 下調べ・前処理・分類
"deep": "gemini-3-pro", # 推論が要る本処理
}
@dataclass
class Request:
task: str # 何をしてほしいか（プロンプト本体）
tier: str = "fast" # fast / deep
idempotency_key: str = field(default_factory=lambda: uuid.uuid4().hex)
background: bool = False # 長時間処理は投げて後で受け取る
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class Result:
text: str
model: str
usage: dict[str, int]
idempotency_key: str
def run(req: Request, *, max_retries: int = 3) -> Result:
"""全ジョブが通る唯一の入り口。計装・リトライ・モデル選択をここに集約する。"""
model = MODEL_BY_TIER[req.tier]
started = time.monotonic()
last_err: Exception | None = None
for attempt in range(1, max_retries + 1):
try:
# ↓ ここが API 依存の唯一の点。細部が変わってもこの関数の外には漏らさない。
resp = _client.interactions.create(
model=model,
input=req.task,
# 同じ idempotency_key の再送は重複課金・重複実行を防ぐ
idempotency_key=req.idempotency_key,
background=req.background,
)
usage = {
"input": resp.usage.input_tokens,
"output": resp.usage.output_tokens,
}
_record(req, model, usage, time.monotonic() - started, attempt)
return Result(
text=resp.output_text,
model=model,
usage=usage,
idempotency_key=req.idempotency_key,
)
except Exception as err: # 実運用では型を絞る
last_err = err
wait = min(2 ** attempt, 30)
log.warning("run failed (attempt %d/%d): %s — retry in %ss",
attempt, max_retries, err, wait)
time.sleep(wait)
_record(req, model, {}, time.monotonic() - started, max_retries, failed=True)
raise RuntimeError(f"llm_gateway.run exhausted retries") from last_err
def _record(req, model, usage, elapsed, attempts, *, failed=False):
# 計装も一か所だけ。コスト集計・遅延監視はこのログを読めば済む。
log.info("llm_call key=%s model=%s tier=%s in=%s out=%s elapsed=%.2f attempts=%d failed=%s job=%s",
req.idempotency_key, model, req.tier,
usage.get("input"), usage.get("output"),
elapsed, attempts, failed, req.metadata.get("job", "-"))注意していただきたいのは、_client.interactions.create(...) の引数名は提供時点のドキュメントで確認すべき箇所だということです。GA で引数は安定方向に向かいますが、ここを直接アプリに散らさないという設計そのものが、その不確実性に対する保険になります。アダプタの外側のコードは run(Request(...)) しか知りません。
このアダプタを置いた瞬間に、先ほどの三つの痛みが消えます。計装は _record の一か所。モデル差し替えは MODEL_BY_TIER の一か所。リトライ上限は run の引数一か所。どれも、もう探し回らなくてよくなります。