Antigravity でエージェントを本番に出してから半年が経ち、最初に積み上げた技術的負債は「失敗トレースが読めない」という問題でした。最初の頃は、console.log を散らかしてエージェント実行のステップを記録していました。1 日数十回の実行までは「ログを上から目で追う」で何とかなっていましたが、月数千回のオーダーになった途端、エラーが起きても「どのステップで何が起きたのか」が読めなくなりました。
可観測性ツールを導入して半年運用してきて、いまは「失敗トレースを 3 行読めば原因が分かる」状態に持っていけています。インシデント検知から原因特定までの平均時間(MTTR)も、導入前の 47 分から 11 分まで縮まりました。以下では、Antigravity エージェントに対するトレース設計の考え方と、私が実際にリポジトリで運用しているスパン構造・属性設計・失敗タグ付け・ダッシュボード・リトライポリシーまで、手を動かせる粒度でまとめます。
個人開発者として 2014 年からアプリ開発を続け、累計 5,000 万ダウンロードに到達するなかで、最初の壁紙アプリを App Store に出したとき、本番でのクラッシュレポートをまともに読める形に整えるまでに 3 ヶ月かかったことを覚えています。AdMob のレポートと Crashlytics のスタックトレースを毎晩突き合わせて、エラー分類を 1 つずつ手で増やしていった作業に近いものでした。今回エージェントの可観測性を整備した過程は、その経験をアーティスト・実装者として AI 時代の文脈に翻訳する作業に近いものでした。
なぜエージェントは「失敗の理由が分からない」状態に陥るのか
私が観測した中で、原因はおおよそ次の 3 つに集約されます。
ひとつ目は、ステップ境界が曖昧なケースです。エージェントは「ツール呼び出し → LLM 推論 → 状態更新」のサイクルを回しますが、この境界をログに記録していないと、エラーが「どのサイクルの、どのステップで」起きたのかが追えません。タイムスタンプ付きの平文ログだけ眺めても、サイクル間の構造が見えません。
ふたつ目は、入出力が黒箱になっているケースです。LLM 呼び出しの中で何が起きたかは記録するのに、その前後で渡したコンテキスト・受け取った生レスポンスが残っていない、という状態です。エラーは「LLM が変な返事をした」で終わってしまい、再現できません。
みっつ目は、失敗の分類が無いケースです。エラーログは出ているが「これがリトライすべきものか、ユーザーに通知すべきものか、コードのバグか」を後で判断できません。アラート疲れや、リトライポリシーの破綻に直結します。
可観測性の設計とは、この 3 つを構造化して残す仕組みを最初に作っておくことです。
トレースの基本単位 — スパン構造を決める
私は OpenTelemetry の概念に揃えて、エージェント実行を次の階層構造でスパン化しています。これは Antigravity 固有ではなく汎用的な構造ですが、エージェント特有の「サイクル」を 1 階層挟むのが大事です。
- agent.run (ルートスパン: 1 回のエージェント実行全体)
- agent.cycle (推論・実行サイクル 1 回ぶん)
- agent.plan (LLM による次アクション決定)
- agent.tool_call (ツール実行 — 個別ツールごとに別スパン)
- agent.observe (実行結果を観測してコンテキストに反映)
- agent.cycle
- ...
このうち、agent.cycle を独立したスパンにするのが要点です。エージェントが 8 サイクル回って 5 サイクル目で失敗した場合、サイクル単位のスパンが無いと「5 サイクル目でツールが落ちた」という事実を取り出すのに人間がログを目で追う必要が出ます。
実装は、エージェント実行ループの中でサイクルを開始するタイミングで tracer.start_as_current_span("agent.cycle") を呼ぶだけです。Python なら次のような形になります。
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
tracer = trace.get_tracer("antigravity.agent")
def run_agent(task):
with tracer.start_as_current_span("agent.run") as run_span:
run_span.set_attribute("task.id", task.id)
run_span.set_attribute("task.kind", task.kind)
run_span.set_attribute("agent.policy", task.policy_name)
run_span.set_attribute("agent.tools_available", list(task.tools))
for cycle_idx in range(MAX_CYCLES):
with tracer.start_as_current_span("agent.cycle") as cycle_span:
cycle_span.set_attribute("cycle.index", cycle_idx)
try:
action = plan_next_action(task)
if action is None:
cycle_span.set_attribute("cycle.terminated", True)
run_span.set_attribute("result.status", "success")
return success(task)
result = execute_tool(action)
observe(task, result)
cycle_span.set_attribute("cycle.action_type", action.kind)
except AgentError as e:
cycle_span.set_status(Status(StatusCode.ERROR, str(e)))
cycle_span.set_attribute("result.failure_class", e.failure_class)
run_span.set_attribute("result.status", "failure")
run_span.set_attribute("result.failure_class", e.failure_class)
raise
TypeScript(Node.js)なら、@opentelemetry/api を使って同じ構造をこう書きます。
import { trace, SpanStatusCode } from "@opentelemetry/api";
const tracer = trace.getTracer("antigravity.agent");
export async function runAgent(task: Task): Promise<Result> {
return await tracer.startActiveSpan("agent.run", async (runSpan) => {
runSpan.setAttribute("task.id", task.id);
runSpan.setAttribute("task.kind", task.kind);
runSpan.setAttribute("agent.policy", task.policyName);
try {
for (let cycleIdx = 0; cycleIdx < MAX_CYCLES; cycleIdx++) {
const done = await tracer.startActiveSpan(
"agent.cycle",
async (cycleSpan) => {
cycleSpan.setAttribute("cycle.index", cycleIdx);
const action = await planNextAction(task);
if (action === null) {
cycleSpan.setAttribute("cycle.terminated", true);
return true;
}
cycleSpan.setAttribute("cycle.action_type", action.kind);
const result = await executeTool(action);
observe(task, result);
cycleSpan.end();
return false;
}
);
if (done) {
runSpan.setAttribute("result.status", "success");
return success(task);
}
}
} catch (e) {
runSpan.setStatus({ code: SpanStatusCode.ERROR, message: String(e) });
runSpan.setAttribute("result.failure_class", classify(e));
throw e;
} finally {
runSpan.end();
}
});
}
注意点として、agent.cycle には必ず cycle.index を属性として乗せておきます。後で「3 サイクル目の挙動を見たい」というクエリが必ず必要になるためです。
属性設計 — 何を残し、何を残さないか
スパンの属性は「あとから検索できるか」「保管コストが許容範囲か」のバランスで決めます。私の運用では、次のような分け方にしています。
ルートスパン(agent.run)に乗せる属性:
task.id / task.kind: タスクの一意識別と分類
agent.policy: 適用したエージェントポリシーの名前とバージョン
agent.tools_available: そのタスクで利用可能なツール名の配列
result.status: 成功 / 失敗 / 中断
result.failure_class: 失敗時の分類タグ(後述)
cost.input_tokens / cost.output_tokens: タスク全体での消費トークン
cost.usd_estimate: 推定コスト(USD)
サイクルスパン(agent.cycle)に乗せる属性:
cycle.index: サイクル番号
cycle.action_type: そのサイクルが選んだアクションの種類
cycle.token_count: そのサイクル内で消費した LLM トークン数
cycle.duration_ms: サイクル全体の所要時間
cycle.terminated: 正常終了フラグ
ツールスパン(agent.tool_call)に乗せる属性:
tool.name: ツール名
tool.input_size: 入力サイズ(バイト or 要素数)
tool.output_size: 出力サイズ
tool.error_kind: エラー時の分類
tool.retry_count: そのコール内のリトライ回数
ここで「乗せない」と決めているのが、生の入力・出力テキストです。トレースに入れると保管コストが跳ね上がる上、PII が混ざる事故が起きます。生の入出力は、トレース ID を紐づけた別ストレージ(私の場合は Cloudflare R2 のオブジェクトストア)に保存し、トレースには「保存先のキー」だけ載せています。
def record_io(span, payload, kind):
# 生 I/O は別ストアに置いてキーだけ載せる
key = f"{span.context.trace_id:032x}/{span.context.span_id:016x}/{kind}"
r2_client.put_object(Bucket="agent-io", Key=key, Body=payload)
span.set_attribute(f"io.{kind}.storage_key", key)
span.set_attribute(f"io.{kind}.size_bytes", len(payload))
この設計だと、デバッグしたいトレースを Grafana で見つけたあと、io.input.storage_key を使って 1 クエリで生入力を取り出せます。属性に長文を詰め込むより、検索 → 詳細遷移の 2 ステップに分けたほうが運用全体のコストが下がります。R2 の保管コストは 1GB あたり月 0.015 USD で、1 ヶ月で 6GB 程度のトレース生 I/O を保管しても 0.10 USD 未満に収まっています。
失敗の分類タグ — result.failure_class
エージェントの失敗を「全部 error として記録する」のは可観測性の敗北です。私は次の分類を result.failure_class 属性として付けています。
tool_unavailable: 外部 API が落ちている / 認証切れ
tool_misuse: ツールに不正な入力を渡した(プロンプトの設計ミス)
model_refusal: LLM が安全方針で拒否した
model_hallucination: LLM が存在しないツール / 関数を呼ぼうとした
cycle_budget_exceeded: 最大サイクル数を超過した
token_budget_exceeded: トークン予算超過
validation_failed: ツール出力のスキーマ検証に失敗した
unknown_runtime_error: 上記に該当しない実行時エラー
この分類は、運用しながら少しずつ育てていくものです。最初から完璧を目指すと書き始められないので、私は最初の 1 ヶ月で出てきたエラーを上から眺めて、頻度の高い 5 種類だけ分類しました。残りはすべて unknown_runtime_error に入れて、月一で見直しています。
分類タグが整っていると、ダッシュボードで「tool_unavailable と validation_failed だけ可視化する」のような切り出しができます。これがリトライポリシーや、ユーザー通知文言の設計に直結します。
failure_class とリトライポリシーの対応表
failure_class を整備した最大の利益は、「どの失敗を、どう扱うか」のポリシーを単一のテーブルにまとめられることです。私のリポジトリには以下の対応表をコードで持っています。
| failure_class | リトライ | バックオフ | 通知 | サポート連絡 |
| tool_unavailable | あり(最大3回) | 指数 (1s, 2s, 4s) | 失敗継続時のみ | 24h で 10 件超で起票 |
| tool_misuse | なし | — | あり(即時) | なし(プロンプト修正) |
| model_refusal | なし | — | あり(即時) | なし |
| model_hallucination | あり(1回のみ) | 即時 | あり | 週次レビュー |
| cycle_budget_exceeded | なし | — | あり | なし(予算見直し) |
| token_budget_exceeded | なし | — | あり | なし(予算見直し) |
| validation_failed | あり(最大2回) | 即時 | あり(2回失敗時) | 週次レビュー |
| unknown_runtime_error | あり(最大1回) | 5s | あり(即時) | 即時起票 |
実装は Python の dataclass で次のように書いています。
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class RetryPolicy:
max_retries: int
backoff_strategy: str # "exponential" | "immediate" | "fixed"
notify_user: bool
ticket_threshold: Optional[int] = None
RETRY_POLICIES = {
"tool_unavailable": RetryPolicy(3, "exponential", False, ticket_threshold=10),
"tool_misuse": RetryPolicy(0, "immediate", True),
"model_refusal": RetryPolicy(0, "immediate", True),
"model_hallucination": RetryPolicy(1, "immediate", True),
"cycle_budget_exceeded": RetryPolicy(0, "immediate", True),
"token_budget_exceeded": RetryPolicy(0, "immediate", True),
"validation_failed": RetryPolicy(2, "immediate", True),
"unknown_runtime_error": RetryPolicy(1, "fixed", True),
}
このテーブルは「設計判断を1箇所に集める」のが目的です。新しい failure_class を足すときも、ここを更新するだけで全体の挙動が変わります。リトライロジックがあちこちに散ると、運用が壊れるのは時間の問題です。
ダッシュボード設計 — 「3 行で原因が見える」運用
トレースを Grafana や Honeycomb のようなツールに流し込んだあと、私は次の 3 つのビューを最初に作っています。これだけで、本番の障害対応時間が体感で半分以下になりました(具体的には MTTR 47 分 → 11 分)。
ビュー 1: 直近 24 時間の失敗トレース一覧
result.status = failure で絞り込んだスパン一覧を、result.failure_class でグループ化します。「いま何が一番起きているか」が一目で分かります。Antigravity の更新で挙動が変わったとき、ここで model_hallucination が急に増えるのを観測できれば、ユーザー報告より先に気づけます。
Grafana の LogQL なら次のクエリでこのビューが組めます(Loki + OpenTelemetry Logs を使っている場合)。
sum by (failure_class) (
count_over_time(
{service="antigravity-agent"}
| json
| result_status="failure"
[1h]
)
)
ビュー 2: サイクル数分布のヒストグラム
成功した agent.run の cycle.index の最大値を集計したヒストグラムです。「通常は 3 サイクルで終わるタスクが、最近 6 サイクル使うようになっている」という変化が見えると、プロンプト設計の劣化を検知できます。LLM のモデル更新後に分布がずれることが多いので、新モデル切り替え後 1 週間はここを毎日眺めるようにしています。
Honeycomb なら次のクエリです。
WHERE name = "agent.run" AND result.status = "success"
VISUALIZE HEATMAP(cycle.index_max)
GROUP BY task.kind
TIME 24h
ビュー 3: ツール別エラー率の推移
tool.name ごとに tool.error_kind が null でない比率を時系列で出します。特定のツール(外部 API 連携など)が静かに劣化しているのを早く検知できます。私の場合、ある決済プロバイダの API が週次でわずかに劣化していくのを、ここで気づいてサポートに連絡したことが 2 回あります。
sum by (tool_name) (rate(agent_tool_call_total{error_kind!=""}[1h]))
/
sum by (tool_name) (rate(agent_tool_call_total[1h]))
ビュー 1 〜 3 は OSS の Grafana テンプレートのような形で配布されているわけではないので、自分のスキーマに合わせて自作します。ただ、属性設計が上の通り整っていれば、Grafana のクエリエディタで PromQL / LogQL を書ける範囲で 1〜2 時間で組めます。
コスト可視化 — 「どのタスクが財布を食っているか」を 1 グラフに
エージェントは「気がつくとコストが跳ねている」分野なので、トレースにコスト属性を乗せておくと運用判断が早くなります。私は cost.input_tokens / cost.output_tokens / cost.usd_estimate の 3 属性を agent.run に乗せて、次のクエリで「タスク種別別のコスト推移」を見ています。
sum by (task_kind) (
rate(agent_run_cost_usd[1h])
)
この可視化を導入してから 1 週間で、特定のタスク種別(社内文書要約系)が全体コストの 62% を占めていることが見えました。コンテキスト圧縮(古い文書からの引用を要約する前処理)を入れることで、その種別のコストを 38% 削減できました。
「気づく → 改善する」のループを 1 週間で回せるかどうかは、計測の解像度で決まります。
PII と監査ログの扱い
エージェントが扱うデータには、ユーザーの個人情報が混ざることがあります。私はトレース属性に PII が含まれる可能性のあるフィールド(メールアドレス、決済情報、ユーザー本文等)を絶対に書かないルールにしています。代わりに次のフローを取っています。
- PII を含む可能性のあるテキストは、
SecureBlob 型でラップして専用ストアに保存
- トレースには
pii.storage_key と pii.kind だけ載せる
- デバッグ時にトレース ID から PII ストアへのアクセスは別の権限が必要
これは GDPR / 個人情報保護法対応として最小限の構えですが、運用上は「うっかりログに混ざる事故」を構造的に防げます。実装はラッパー型を 1 つ作って通すだけなので、半日で入れられます。
@dataclass
class SecureBlob:
payload: bytes
kind: str # "email" | "phone" | "user_text"
def __repr__(self):
# str() / repr() でログに漏れることを防ぐ
return f"<SecureBlob kind={self.kind} size={len(self.payload)}>"
def store(self, span) -> str:
key = f"pii/{secrets.token_hex(16)}"
secure_store.put(key, self.payload)
span.set_attribute("pii.storage_key", key)
span.set_attribute("pii.kind", self.kind)
return key
運用 6 ヶ月で見えてくるもの — 私の実例
トレース設計を入れて 6 ヶ月運用した結果、見えるようになったことを 4 つだけ書き残しておきます。
第一に、失敗の 6 割は tool_unavailable で、そのほとんどが特定の外部 API に集中していることが見えました。これが見えたことで、その API の呼び出し前にヘルスチェック → フォールバック手順を入れる、という設計改善に繋がりました。それまでは「なんかよく失敗するなあ」という体感だけで、原因を特定できていませんでした。
第二に、model_hallucination がモデル更新の翌週に倍増するパターンが見えました。これが見えるとモデル更新時の再評価ステップを儀式として組み込むようになり、本番投入前にプロンプトを更新する判断ができるようになりました。
第三に、特定のタスク種別でだけサイクル数が異常に多くなることが見えました。これはプロンプトの初期コンテキストが不十分で、エージェントが情報収集に余分なサイクルを使っていたためです。コンテキスト改修で平均サイクル数が 4.2 → 2.8 まで下がり、トークン消費も 35% 削減できました。
第四に、MTTR が 47 分から 11 分に縮まったことが、長期で見ると一番の収穫です。インシデント対応 1 回あたり 30 分強の節約は、月 20 件のインシデントで月 10 時間以上の節約になります。可観測性は「最初の 1 時間の投資」が一番割に合う領域だと感じています。
推奨される導入順序
これから可観測性を導入する方向けに、私が「もう一度ゼロからやるならこの順番」と思う 4 ステップを残しておきます。
- agent.cycle スパンだけ足す(30 分) — 既存のエージェント実行ループに
tracer.start_as_current_span("agent.cycle") を 1 行入れる。これだけでサイクル境界が見えるようになり、運用の見え方が一変します
- failure_class を 5 種類だけ定義する(1 時間) — 上で挙げた 5 つの分類のうち、運用しているエージェントで頻度の高い 5 つを選び、
result.failure_class 属性として乗せる
- ダッシュボード 1 を作る(1 時間) — 失敗トレース一覧をビューにします。それだけでインシデント対応が変わります
- コスト属性を乗せる(30 分) —
cost.usd_estimate を agent.run に乗せ、タスク種別別のコスト推移を見える化する
合計 3 時間の投資で、「失敗の理由が一発で分かる」状態の半分以上が手に入ります。残りの半分は、運用しながら 1 ヶ月かけて磨いていく領域です。
可観測性は「始めるまでが重い」分野ですが、最初にスパン構造と分類タグを 1 時間決めて、その後で 1 サイクルずつ実装していけば、運用 1 ヶ月後には「失敗の理由が一発で分かる」状態に近づけます。次のエージェント実装では、まず agent.cycle スパンを 1 つ足すところから始めてみてください。同じ課題に取り組んでいる方の参考になれば幸いです。