Claude API の本番観測性を OpenTelemetry GenAI 規約で再設計する — モデル移行とコスト監査に耐える設計メモ

// src/lib/otel-metrics.ts
import { metrics } from "@opentelemetry/api";
 
const meter = metrics.getMeter("dolice.claude-api", "1.0.0");
 
export const claudeLatency = meter.createHistogram(
  "gen_ai.client.operation.duration",
  {
    description: "End-to-end latency of Claude API calls",
    unit: "s",
    advice: {
      explicitBucketBoundaries: [
        0.1, 0.25, 0.5, 1, 2, 4, 8, 16, 32, 64,
      ],
    },
  },
);
 
export const claudeTokens = meter.createHistogram(
  "gen_ai.client.token.usage",
  {
    description: "Token usage per request",
    unit: "{token}",
    advice: {
      explicitBucketBoundaries: [
        16, 64, 256, 1024, 4096, 16384, 65536, 200000,
      ],
    },
  },
);
 
export const claudeCostJpy = meter.createCounter("dolice.claude.cost_jpy", {
  description: "Estimated JPY cost per call (rolled up from token counts)",
});
 
export const claudeErrors = meter.createCounter(
  "gen_ai.client.error.count",
  { description: "Error count grouped by error type" },
);

function recordUsage(model: string, res: Anthropic.Message, latencyMs: number) {
  const attrs = {
    "gen_ai.system": "anthropic",
    "gen_ai.request.model": model,
    "gen_ai.response.model": res.model,
    "dolice.tenant_id": currentTenantId(),
    "dolice.surface": currentSurface(),
  };
  claudeLatency.record(latencyMs / 1000, attrs);
  claudeTokens.record(res.usage.input_tokens, {
    ...attrs,
    "gen_ai.token.type": "input",
  });
  claudeTokens.record(res.usage.output_tokens, {
    ...attrs,
    "gen_ai.token.type": "output",
  });
  const jpy =
    res.usage.input_tokens * PRICING[model].inputJpy +
    res.usage.output_tokens * PRICING[model].outputJpy +
    (res.usage.cache_creation_input_tokens ?? 0) *
      PRICING[model].cacheCreationJpy +
    (res.usage.cache_read_input_tokens ?? 0) *
      PRICING[model].cacheReadJpy;
  claudeCostJpy.add(jpy, attrs);
}

-- ClickHouse / TimescaleDB 共通: テナント別の日次コスト集計
SELECT
  toDate(timestamp) AS day,
  attributes['dolice.tenant_id'] AS tenant,
  attributes['gen_ai.response.model'] AS model,
  sum(value) AS jpy_total
FROM otel_metric_sum
WHERE name = 'dolice.claude.cost_jpy'
  AND timestamp >= now() - INTERVAL 30 DAY
GROUP BY day, tenant, model
ORDER BY day DESC, jpy_total DESC;

# scripts/replay_shadow_traffic.py
"""24 時間分のリクエストを別モデルにリプレイしてレスポンス品質を比較するスクリプト。
 
Usage:
  python replay_shadow_traffic.py \
      --from claude-sonnet-4-6 \
      --to claude-opus-4-6 \
      --hours 24 \
      --sample 0.02
"""
import argparse
import asyncio
import json
import os
import statistics
from datetime import datetime, timedelta
from anthropic import AsyncAnthropic
import clickhouse_connect
 
client = AsyncAnthropic()
 
async def replay_one(row, target_model: str) -> dict:
    payload = json.loads(row["request_payload"])
    started = datetime.now()
    res = await client.messages.create(
        model=target_model,
        system=payload.get("system"),
        messages=payload["messages"],
        max_tokens=payload["max_tokens"],
    )
    elapsed = (datetime.now() - started).total_seconds()
    return {
        "original_id": row["response_id"],
        "original_model": row["model"],
        "shadow_model": target_model,
        "original_output_tokens": row["output_tokens"],
        "shadow_output_tokens": res.usage.output_tokens,
        "shadow_latency": elapsed,
        "shadow_text": res.content[0].text[:2000],
    }
 
async def main(args):
    ch = clickhouse_connect.get_client(host=os.environ["CH_HOST"])
    since = datetime.now() - timedelta(hours=args.hours)
    rows = ch.query(
        """
        SELECT response_id, model, request_payload, output_tokens
        FROM claude_request_replay
        WHERE timestamp >= %(since)s
          AND model = %(from_model)s
          AND rand() / pow(2, 32) < %(sample)s
        """,
        parameters={
            "since": since,
            "from_model": args.from_model,
            "sample": args.sample,
        },
    ).named_results()
 
    sem = asyncio.Semaphore(4)  # 同時 4 本まで
    async def bounded(row):
        async with sem:
            try:
                return await replay_one(row, args.to)
            except Exception as e:
                return {"error": str(e), "original_id": row["response_id"]}
    results = await asyncio.gather(*(bounded(r) for r in rows))
 
    ok = [r for r in results if "error" not in r]
    latencies = [r["shadow_latency"] for r in ok]
    ratio_tokens = [
        r["shadow_output_tokens"] / max(1, r["original_output_tokens"])
        for r in ok
    ]
    print(f"Replayed: {len(ok)} / {len(results)} (errors {len(results) - len(ok)})")
    print(f"Shadow latency p50/p95: {statistics.median(latencies):.2f}s / "
          f"{statistics.quantiles(latencies, n=20)[-1]:.2f}s")
    print(f"Output token ratio (shadow / original) median: "
          f"{statistics.median(ratio_tokens):.2f}")
    ch.insert("claude_shadow_results", ok)
 
if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--from", dest="from_model", required=True)
    p.add_argument("--to", required=True)
    p.add_argument("--hours", type=int, default=24)
    p.add_argument("--sample", type=float, default=0.02)
    asyncio.run(main(p.parse_args()))