独自ツールを足した直後のエージェントは、たいてい気持ちよく動きます。問題が出るのはその数日後、同じツールが想定より多く呼ばれ始めたときです。私自身、個人開発で使っている自動化エージェントに社内 API 連携ツールを一本追加したところ、最初の週は完璧で、翌週に「同じ請求が二重に作られている」という形で初めて綻びに気づきました。原因はツールの実装そのものではなく、エージェントがタイムアウトを失敗と解釈して、すでに成功している処理をもう一度呼んでいたことでした。
独自ツールの設計記事はスキーマや権限境界の話で終わりがちですが、運用で本当に効いてくるのは「同じツールが二度呼ばれても壊れない」という再実行耐性のほうです。ここでは、その観点から効いた実装と判断基準だけを残します。
二重副作用はどこから来るのか
エージェントがツールを二度呼ぶ経路は、思っているより多くあります。代表的なのは次のような場面です。
ツールは実際には成功したのに、レスポンスがタイムアウトで返り、エージェントが「失敗した」と判断して再試行する。
長い作業の途中でコンテキストが圧縮され、実行済みの呼び出しをエージェントが忘れて、もう一度組み立てる。
並列エージェント構成で、複数のワーカーが同じタスクを拾ってしまう。
これらに共通するのは、ツール側から見ると「ほぼ同じ引数の呼び出しが短時間に二度来る」という現象として現れることです。本番運用で初めて表面化するこの種の事故は、エージェント側のプロンプトをいくら直しても根絶できません。防御はツール側の、しかも引数の側で打つのが確実です。
冪等キー — 同じ操作を二度実行しない
書き込み系・破壊系のツールには、冪等キーを必須にすることを強く推奨します。冪等キーは「この操作は一度きり実行されるべき」という単位を表す文字列で、呼び出し側（エージェント）が生成し、ツール側が保存します。同じキーで二度目の呼び出しが来たら、再実行せず初回の結果をそのまま返します。
import hashlib, json, time
class IdempotencyStore : """冪等キーと初回結果を TTL 付きで保持する。 本番では Redis などの共有ストアにする（並列ワーカー間で共有するため）。""" def __init__ (self, ttl_seconds: int = 86400 ): self ._store: dict[ str , tuple[ float , dict ]] = {} self ._ttl = ttl_seconds
def get (self, key: str ) -> dict | None : entry = self ._store.get(key) if not entry: return None ts, result = entry if time.time() - ts > self ._ttl: self ._store.pop(key, None ) return None return result
def put (self, key: str , result: dict ) -> None : self ._store[key] = (time.time(), result)
def create_invoice (args: dict , idem: IdempotencyStore) -> dict : key = args.get( "idempotency_key" ) if not key: return { "ok" : False , "error" : { "code" : "MISSING_IDEMPOTENCY_KEY" , "message" : "書き込み操作には idempotency_key が必須です。" , "retryable" : False }} cached = idem.get(key) if cached is not None : # 二度目以降は副作用を起こさず初回結果を返す return { ** cached, "replayed" : True } result = { "ok" : True , "data" : _do_create_invoice(args)} idem.put(key, result) return result ここで大事なのは、冪等キーを「呼び出し側が生成する」ことです。ツール側で引数のハッシュからキーを作る方法もありますが、それだと「意図的に二度作りたい正当な再呼び出し」と「事故の再実行」を区別できません。エージェントが操作の単位ごとに一意なキーを発行し、再試行のときは同じキーを使い回す、という約束にすると、両者がきれいに分かれます。
スキーマ側では、idempotency_key を書き込み系ツールの必須引数にしておきます。エージェントには「この操作は一度きり。再試行するなら同じキーを使ってください」という説明を description に添えます。
CREATE_INVOICE_SCHEMA = { "name" : "create_invoice" , "description" : "請求を作成します。冪等です。再試行時は同じ idempotency_key を使ってください。" , "parameters" : { "type" : "object" , "properties" : { "customer_id" : { "type" : "string" }, "amount_cents" : { "type" : "integer" , "minimum" : 1 }, "idempotency_key" : { "type" : "string" , "description" : "この請求作成を一意に識別する文字列。再試行では同じ値を使う。" , }, }, "required" : [ "customer_id" , "amount_cents" , "idempotency_key" ], }, } TTL の長さは操作の性質で決めます。請求や送金のように「同じ操作を翌日にもう一度やったらまず事故」というものは 24 時間以上、逆に通知送信のように短時間の重複だけ防げばよいものは数分で十分です。私は破壊的なものほど長く取るようにしています。
エラー契約 — エージェントが次の一手を選べる形で返す
再実行耐性の話とエラーハンドリングは地続きです。なぜなら、エージェントが「失敗した」と誤認して再試行するかどうかは、ツールが返すエラーの形に左右されるからです。スタックトレースをそのまま返すと、エージェントはそれが再試行可能なのか致命的なのかを判断できず、とりあえずもう一度呼ぶという最悪の選択をしがちです。
ツールのエラーは、エージェントが分岐できる構造で返します。最低限、エラーコード・人間可読のメッセージ・再試行可否の三点を固定フィールドにします。
def get_order_status (order_id: str ) -> dict : try : return { "ok" : True , "data" : adapter.get_status(order_id)} except ValueError as e: return { "ok" : False , "error" : { "code" : "INVALID_INPUT" , "message" : str (e), "retryable" : False }} except TransientAPIError: return { "ok" : False , "error" : { "code" : "UPSTREAM_UNAVAILABLE" , "message" : "注文 API が一時的に応答していません。" , "retryable" : True , "retry_after_seconds" : 30 }} except NotFoundError: return { "ok" : False , "error" : { "code" : "NOT_FOUND" , "message" : f "注文 { order_id } は存在しません。" , "retryable" : False }} retryable: False を返す経路を意図的に増やすのがコツです。入力不正・存在しないリソース・権限不足は、何度呼んでも結果が変わりません。これらを明確に「再試行不可」と伝えるだけで、エージェントの無駄な再呼び出しが目に見えて減ります。逆に retryable: True を返すのは、依存先の一時障害のように「時間を置けば直る見込みがあるもの」に限定します。retry_after_seconds を添えておくと、エージェントが即座に叩き直す事故も防げます。
エラーコードの一覧は、ツール群をまたいで共通にします。下の表は、私が実際に使っているコードと、エージェントに期待する挙動の対応です。
エラーコード 意味 retryable エージェントの期待挙動
INVALID_INPUT引数が不正 false 引数を直してから呼ぶ。同じ引数で再試行しない NOT_FOUND対象が存在しない false 別の経路で対象を特定する PERMISSION_DENIED権限不足 false 呼ばずに人間へ委ねる UPSTREAM_UNAVAILABLE依存先の一時障害 true retry_after_seconds 後に再試行RATE_LIMITED呼び出し過多 true 猶予秒後に再試行。並列を絞る CONFLICT状態が競合 false 最新状態を取り直してから判断
この対応表を description やシステムプロンプトにそのまま載せておくと、エージェントのリトライ挙動が安定します。コード体系を共通化しておくと、新しいツールを足すたびに挙動を一から教え直す必要がなくなる点も実務では大きいです。
ヘルスゲート — 落ちている依存先に問い合わせ続けない
独自ツールは外部 API や DB に依存することが多く、その依存先が落ちている間にエージェントが問い合わせ続けると、待ち時間とトークンを無駄に消費します。依存先の健康状態を短くキャッシュし、不健全なら即座に UPSTREAM_UNAVAILABLE を返すゲートを挟みます。
class ToolHealth : def __init__ (self, check_interval: int = 30 ): self ._last: dict[ str , float ] = {} self ._ok: dict[ str , bool ] = {} self ._interval = check_interval
def is_healthy (self, tool: str , checker) -> bool : now = time.monotonic() if now - self ._last.get(tool, 0 ) < self ._interval: return self ._ok.get(tool, True ) # キャッシュを返す try : ok = checker() except Exception : ok = False self ._last[tool] = now self ._ok[tool] = ok return ok 実チェックは 30 秒に一度だけ走らせ、間はキャッシュを返す、という素朴な実装で十分に効きます。ヘルスチェックが落ちているツールは、エージェントから呼ばれた瞬間に再試行可能なエラーとして弾けるので、サーキットブレーカーの軽量版として機能します。ここで返すエラーも retryable: True にしておくと、依存先が復帰したあとにエージェントが自然と再開できます。
呼び出しログがないと、事故に後から気づく
冒頭の二重請求に私が気づけたのは、ツール呼び出しを構造化ログに残していたからです。本番運用では、すべての呼び出しを最低限この六項目で残しておくと、エラーの回避と原因究明が一気に楽になります。
import json, logging
def log_tool_call (session_id, tool, args, result, elapsed_ms): logging.info(json.dumps({ "session_id" : session_id, "tool" : tool, "args" : _mask_secrets(args), # 機微情報はマスク "ok" : result.get( "ok" ), "error_code" : (result.get( "error" ) or {}).get( "code" ), "replayed" : result.get( "replayed" , False ), # 冪等再生かどうか "elapsed_ms" : elapsed_ms, }, ensure_ascii = False )) replayed を残しておくのが地味に効きます。冪等キーで二度目の実行を弾いた回数が見えるので、「エージェントがどれだけ再試行で同じ操作を叩いているか」が定量的に分かります。この数字が増えているツールは、タイムアウト設定かエラー契約のどちらかが甘い、という診断につながります。私はこのログを週に一度眺めて、再生率の高いツールから順に手当てするようにしています。
ツールを足す前にもう一段問う
最後に、再実行耐性とは別の角度の運用知見をひとつ。独自ツールは増やすほどエージェントの選択精度を下げます。似た名前のツールが並ぶと、エージェントがどれを呼ぶべきか迷い、間違ったツールを選ぶ確率が上がるからです。
私はツールを足すかどうかの判断に、設計の良し悪しとは別の基準を加えています。標準ツールの組み合わせで解けないか、既存ツールに引数を一本足すだけで済まないか、そして「このツールが二度呼ばれても安全か」をその場で答えられるか。三つめに即答できないなら、まだ足すべきではない、というのが運用で固まった感覚です。
読み取り専用で決定的な結果を返すツールから始め、書き込み系は冪等キーとエラー契約を揃えてから段階的に足す。このペースを守るだけで、「エージェントを使うと楽になる」という体験を壊さずに機能を広げられます。独自ツールの価値は、増やした数ではなく、増やしても壊れない設計に宿ると考えています。