⚙ AIツール/2026-06-17上級

Antigravity の LLM アプリは、ダッシュボードが緑のまま請求と品質がずれていく — 観測の計装メモ

LLM アプリの監視は、合計のコストとレイテンシだけ見ていると静かなずれを見逃します。機能・テナント・プロンプト版で属性付けし、品質劣化とコスト急増を早期に捕まえる計装の設計メモです。

llmops² observability¹³ opentelemetry² cost quality² antigravity³⁶⁷ production⁵⁵

✦ プレミアム記事

Antigravity で組んだエージェントを本番に出して数週間、Grafana のダッシュボードはずっと緑でした。合計コストはなだらか、P95 レイテンシも閾値内、エラーレートはほぼゼロ。それでも月末の請求は見積もりの 1.6 倍で、ある機能の回答だけが目に見えて雑になっていました。

合計値は、平均の中に問題を溶かしてしまいます。個人開発で複数の機能を一つの API キーに相乗りさせていると、この「溶けてしまう」性質が特に効いてきます。ここでは、合計の裏で進む二種類のずれ — 請求のずれと品質のずれ — を観測で捕まえるための計装を、実際に組み直した順に書き残します。

合計を見ても、誰がコストを食べているかは分からない

最初に作った監視は、モデル別のトークン数とコストを足し上げるだけのものでした。これは「いくら使ったか」には答えますが、「どこで増えたか」には答えません。本番では、コストは機能ごと・テナントごとに大きく偏ります。要約機能が全体の 6 割を食べていたり、特定の一社が平均の 20 倍を消費していたりします。合計のグラフはそれを平らに均してしまいます。

そこで計装の単位を、モデルではなく「機能 × テナント × プロンプト版」に変えました。OpenTelemetry のスパンとメトリクスに、この三つを必ず属性として乗せます。属性さえ揃っていれば、後から PromQL でどの切り口にも展開できます。

# llm_telemetry.py
import time
import anthropic
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
 
_tp = TracerProvider()
_tp.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317")))
trace.set_tracer_provider(_tp)
metrics.set_meter_provider(MeterProvider(metric_readers=[
    PeriodicExportingMetricReader(
        OTLPMetricExporter(endpoint="http://otel-collector:4317"),
        export_interval_millis=30_000,
    )
]))
tracer = trace.get_tracer("llm-app")
meter = metrics.get_meter("llm-app")
 
tokens = meter.create_counter("llm.tokens", description="tokens by direction")
cost = meter.create_counter("llm.cost_usd", unit="USD", description="API cost in USD")
latency = meter.create_histogram("llm.latency_ms", unit="ms", description="end-to-end latency")
errors = meter.create_counter("llm.errors", description="error count by type")
 
# 価格は $/1M tokens。モデル更新のたびに必ず見直す（後述）
MODEL_PRICES = {
    "claude-sonnet-4-6": {"in": 3.0, "out": 15.0},
    "claude-haiku-4-5-20251001": {"in": 0.25, "out": 1.25},
    "claude-opus-4-8": {"in": 5.0, "out": 25.0},
}
 
class TelemetryClient:
    """機能・テナント・プロンプト版で属性付けする計装ラッパー"""
 
    def __init__(self):
        self._client = anthropic.Anthropic()
 
    def call(self, *, model, messages, feature, tenant, prompt_version,
             max_tokens=2048, system=None, role="product"):
        # role="product" は本番応答、role="eval" は品質評価用（費用を分離する鍵）
        attrs = {
            "feature": feature, "tenant": tenant,
            "prompt_version": prompt_version, "model": model, "role": role,
        }
        with tracer.start_as_current_span("llm.call") as span:
            for k, v in attrs.items():
                span.set_attribute(k, v)
            t0 = time.monotonic()
            try:
                kwargs = {"model": model, "max_tokens": max_tokens, "messages": messages}
                if system:
                    kwargs["system"] = system
                res = self._client.messages.create(**kwargs)
                ms = (time.monotonic() - t0) * 1000
                tin, tout = res.usage.input_tokens, res.usage.output_tokens
                price = MODEL_PRICES.get(model, {"in": 3.0, "out": 15.0})
                usd = tin / 1e6 * price["in"] + tout / 1e6 * price["out"]
 
                tokens.add(tin, {**attrs, "direction": "in"})
                tokens.add(tout, {**attrs, "direction": "out"})
                cost.add(usd, attrs)
                latency.record(ms, attrs)
                span.set_attribute("llm.cost_usd", usd)
                span.set_attribute("llm.latency_ms", ms)
                return {"text": res.content[0].text, "cost_usd": usd,
                        "latency_ms": ms, "in": tin, "out": tout}
            except Exception as e:
                errors.add(1, {**attrs, "error_type": type(e).__name__})
                span.record_exception(e)
                span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
                raise

ここで prompt_version を属性に含めているのは、後の品質追跡で効いてきます。プロンプトを変えた瞬間からコストと品質がどう動いたかを、版の境目で切れるようにしておきたいからです。バージョン文字列はプロンプトテンプレートのハッシュでも、手で振る chat-v7 のような短い識別子でも構いません。大事なのは「いつ何を変えたか」がメトリクスの軸に残ることです。

機能別・テナント別にコストを展開するクエリは、属性が揃っていれば素直に書けます。

# 機能ごとの直近1時間のコスト（どの機能が食べているか）
sum(increase(llm_cost_usd_total{role="product"}[1h])) by (feature)
 
# テナント別の上位消費（特定の一社の暴走を見つける）
topk(5, sum(increase(llm_cost_usd_total[24h])) by (tenant))

評価モデルの費用が本番費用に紛れ込む

品質を自動評価するために LLM-as-a-Judge を回すと、評価それ自体が API を叩きます。最初これを区別していなかったので、本番コストのグラフに評価分が上乗せされ、「本番が高い」のか「評価を回しすぎている」のかが分からなくなりました。role="eval" を属性で分けたのはこのためです。評価は安いモデルで回し、PromQL では role="product" で本番費用だけを見ます。

評価は全リクエストにかけるとそれだけで費用が膨らむので、巡回サンプリングにしました。毎分すべてを採点するのではなく、機能ごとに一定割合だけを抜き取って評価します。これで評価費用を本番費用の数パーセントに抑えつつ、傾向の変化は十分に拾えます。

# quality_sampler.py
import json, random
from dataclasses import dataclass
 
@dataclass
class Score:
    relevance: float
    grounded: float        # 参照情報に基づいているか（RAG の場合）
    overall: float
    note: str
 
class QualitySampler:
    """巡回サンプリングで品質を採点し、プロンプト版ごとに集計する"""
 
    def __init__(self, telemetry, sample_rate=0.05, eval_model="claude-haiku-4-5-20251001"):
        self.t = telemetry
        self.rate = sample_rate
        self.eval_model = eval_model
 
    def maybe_score(self, *, question, answer, context, feature, tenant, prompt_version):
        if random.random() > self.rate:
            return None  # サンプル対象外
        ctx = f"\n\n参照情報:\n{context}" if context else ""
        prompt = (
            "次の回答を JSON のみで採点してください。"
            "各項目は 0.0〜1.0。\n"
            f"質問: {question}{ctx}\n\n回答: {answer}\n\n"
            '{"relevance": 適切さ, "grounded": 参照情報への忠実さ, '
            '"overall": 総合, "note": "一文の所見"}'
        )
        res = self.t.call(
            model=self.eval_model,
            messages=[{"role": "user", "content": prompt}],
            feature=feature, tenant=tenant, prompt_version=prompt_version,
            role="eval", max_tokens=300,
        )
        data = json.loads(res["text"])
        score = Score(**data)
        # メトリクスにも品質を流し、版ごとに追えるようにする
        from opentelemetry import metrics
        gauge = metrics.get_meter("llm-app").create_histogram("llm.quality_overall")
        gauge.record(score.overall, {"feature": feature, "prompt_version": prompt_version})
        return score

採点結果を prompt_version で集計すると、品質劣化が「いつから」始まったかが版の境目に現れます。これが、次の一番厄介なずれを捕まえる土台になります。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦機能・テナント・プロンプト版で属性付けし、合計値の裏に隠れたコスト偏りを切り出す計装パターン

✦既定モデルが勝手に上がる時代に、品質劣化を数値で捕まえる巡回サンプリング評価の設計

✦評価モデルの費用が本番費用に紛れ込む罠と、コスト台帳でテナント単位の急増を止める実装

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

既定モデルが勝手に上がる時代の、品質の静かな劣化

2026 年に入ってから、プラットフォーム側が既定モデルを黙って引き上げる動きが目立ちます。Gemini API では検索やエージェントの既定モデルが新版に差し替わり、Antigravity でもバージョンごとに裏で使われるモデルが動きます。便利な反面、こちらのプロンプトを一切変えていないのに、ある日から出力の傾向が変わることがあります。多くの場合は改善ですが、こちらの細かい指示の効き方が変わって、特定機能だけ精度が落ちることがあります。

この劣化はエラーにならないので、エラーレートのアラートには引っかかりません。レイテンシも大きくは動きません。捕まえられるのは品質メトリクスだけです。そこで、採点済みスコアの移動平均を版ごと・機能ごとに追い、直前の窓と比べて一定以上下がったら警告を出すようにしました。

# quality_drift.py
from collections import deque, defaultdict
import statistics
 
class QualityDrift:
    """機能ごとに品質スコアの移動窓を持ち、劣化を検知する"""
 
    def __init__(self, window=200, drop_threshold=0.08):
        self.window = window
        self.drop = drop_threshold
        self.recent = defaultdict(lambda: deque(maxlen=window))
        self.baseline = {}  # feature -> 直近で安定していた平均
 
    def observe(self, feature, overall):
        self.recent[feature].append(overall)
 
    def check(self, feature):
        buf = self.recent[feature]
        if len(buf) < self.window:
            return None  # まだ判断材料が足りない
        half = self.window // 2
        prev = statistics.mean(list(buf)[:half])
        now = statistics.mean(list(buf)[half:])
        if prev - now >= self.drop:
            return {
                "type": "QUALITY_DRIFT",
                "feature": feature,
                "from": round(prev, 3),
                "to": round(now, 3),
                "message": f"{feature} の品質が {prev:.2f}→{now:.2f} に低下。"
                           "モデル既定変更・プロンプト改変・入力分布のずれを疑う",
            }
        return None

警告が出たら、まず確認するのは三点です。モデルの既定が変わっていないか（プラットフォームのリリースノート）、自分がプロンプトを触っていないか（prompt_version の境目）、入力の分布がずれていないか（新しいテナントが想定外の使い方を始めていないか）です。原因の切り分けには、品質低下の窓と prompt_version のメトリクスを重ねるのが一番速い、というのが実感です。版が変わっていないのにスコアだけ落ちていれば、こちらの外側で何かが動いた、と読めます。

私が本番で踏んだのはまさにこの形でした。プロンプトの版は同じまま、要約機能のスコアだけが二週間かけてじわじわ下がっていました。原因は裏の既定モデルの差し替えで、こちらの「箇条書きにしない」という指示の効きが弱まっていたのです。版を固定して明示的にモデルを指定し直したら、スコアは元に戻りました。品質メトリクスを版で切っていなければ、月末の請求と読者の体感が悪くなるまで気づけなかったはずです。

コスト台帳で、テナント単位の急増を止める

合計のコストアラートは、テナント単位の暴走には鈍感です。一社だけが平均の数十倍を消費していても、全体としてはまだ閾値内、ということが起きます。そこで、テナントごとの消費を台帳に記録し、各テナントの平常域からの逸脱で個別に警告を出すようにしました。閾値は全体平均ではなく、そのテナント自身の過去の分布を基準にします。

# cost_ledger.py
from collections import defaultdict
from datetime import datetime, timedelta
import statistics
 
class CostLedger:
    """テナントごとの消費を記録し、自身の平常域からの逸脱を検知する"""
 
    def __init__(self, window_hours=24, sigma=3.0):
        self.window = timedelta(hours=window_hours)
        self.sigma = sigma
        self.entries = []  # (ts, tenant, usd)
        self.hourly = defaultdict(lambda: defaultdict(float))  # tenant -> hour -> usd
 
    def record(self, tenant, usd):
        now = datetime.now()
        self.entries.append((now, tenant, usd))
        self.hourly[tenant][now.replace(minute=0, second=0, microsecond=0)] += usd
        cutoff = now - self.window * 3
        self.entries = [(t, n, c) for (t, n, c) in self.entries if t > cutoff]
 
    def check(self, tenant):
        hours = sorted(self.hourly[tenant].items())
        if len(hours) < 12:
            return None  # 基準を作るには履歴が短い
        past = [v for _, v in hours[:-1]]
        current = hours[-1][1]
        mean, sd = statistics.mean(past), statistics.pstdev(past)
        if sd > 0 and current > mean + self.sigma * sd:
            return {
                "type": "TENANT_COST_SPIKE",
                "tenant": tenant,
                "current_usd": round(current, 4),
                "baseline_usd": round(mean, 4),
                "message": f"{tenant} の時間あたり消費が平常域を逸脱（{current:.3f} vs 平均 {mean:.3f}）",
            }
        return None

テナント単位で見られるようになると、対応の選択肢が増えます。バグでループしているなら止めればよいし、正当な大量利用なら従量課金の話につなげられます。いずれにせよ、月末の請求で初めて気づくよりはずっと早く動けます。Antigravity の Managed Agents のようにスケジュール起動でエージェントを回す構成では、無人の時間帯にこそ暴走が起きやすいので、この台帳は特に効きます。