プロダクションAIパイプライン構築: 1万回以上の実行から学んだ教訓

去年の10月、金曜日の夕方にデプロイを走らせた。小さな変更のつもりだった——プロンプトの一部を最適化して、コスト削減を狙った修正。翌朝起きたら、Slackに30件のアラートが来ていた。

5人チームで作っているドキュメント分類パイプラインが、夜中の2時頃から静かに壊れていた。エラーレートが通常の0.3%から18%に跳ね上がっていたのに、誰も気づかなかった。観測可能性がほぼゼロだったから。

あの経験が、今回書くことのほぼ全部を教えてくれた。

キューの背圧が見えないと、静かに詰まる

うちのパイプラインは最初、シンプルな構成だった。ユーザーがドキュメントをアップロード → Celeryタスクがキューに積まれる → ワーカーがLLMを叩く → 結果をDBに保存。以上。

1日あたり数百リクエストのうちは普通に動いた。でも1日数千になったとき、問題が出始めた。LLMのAPIレートリミットに当たると、タスクが失敗してリトライキューに積み直される。そのリトライがまたレートリミットに当たる。気づいたら、キューが膨らみ続けて、正常なリクエストも一緒に詰まっていた。

最初のうちは「リトライすれば大丈夫」という考えが甘かった。指数バックオフは入れていたけど、jitterを入れていなかった。全ワーカーが同時に同じタイミングでリトライするから、バースト的にAPIを叩き続ける——いわゆるThunderingHerdの問題で、自分でやらかすまでちゃんと理解できていなかった。

修正はシンプルで、でも効果は大きかった:

import random

def exponential_backoff_with_jitter(
    attempt: int,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> float:
    """
    Full jitterを使ったバックオフ。
    AWS Architectureブログ推奨の実装に近い形。
    attempt=0から始める想定。
    """
    delay = min(base_delay * (2 ** attempt), max_delay)
    # [0, delay] の範囲でランダムに待つことで、ワーカーの同期を崩す
    return random.uniform(0, delay)


# Celery 5.3系のタスク定義例
@celery_app.task(
    bind=True,
    max_retries=5,
    autoretry_for=(RateLimitError, APITimeoutError),
)
def classify_document(self, doc_id: str) -> dict:
    try:
        result = call_llm(doc_id)
        return result
    except RateLimitError as exc:
        # retry_jitter=Trueだけだと高負荷時に不十分だった。自前実装のほうが安定した
        wait_time = exponential_backoff_with_jitter(self.request.retries)
        raise self.retry(exc=exc, countdown=wait_time)

Celery 5.3系にはretry_jitter=Trueという組み込みオプションがあるんだけど、実際に試したら挙動が思ったより控えめで、高負荷時には自前実装のほうが安定した。マイレージ・メイ・ベリー、というやつ。

実際のところ: キューの深さ、ワーカーあたりのスループット、リトライ率の3つを最低限モニタリングしないと、詰まっていることに気づくのが遅くなる。感覚値ではなく数字で見る習慣をつけるだけで、かなり変わる。

トークン最適化: 短くすればいいというわけじゃなかった

コスト削減の話をすると、みんな「プロンプトを短くしろ」で終わりにしがち。確かにそれは正しいんだけど、僕が金曜にやらかしたのがまさにこれで——プロンプトのcontext部分を50%削減したら、分類精度が83%から61%に落ちた。

計測していたからわかった話で、もし精度指標を追っていなかったら気づかなかったと思う。

実際に効果があったのは、単純に短くすることよりも構造を変えることだった。うちのケースでは:

• Few-shotサンプルを20個から7個に減らしたが、質の高い例だけ残した → 精度ほぼ変わらず、トークン数30%削減
• システムプロンプトの重複記述を整理した → 地味だけど15%くらい削減
• 出力フォーマットをJSONからYAMLに変えてみた → これは失敗。パース時のエラーが増えた。結局JSON戻した

Right, so — GPT-4oとclaude-sonnet-4-6を同じタスクで比較したとき、特定の分類カテゴリだとSonnetのほうが明らかに苦手なパターンがあった。モデルによって得意・不得意があるのは当たり前だけど、実際に自分のデータで計測するまでは「どっちも同じくらいでしょ」と思っていた。1万回通してみると、そういう差がはっきり出てくる。

コスト的には、claude-haiku-4-5を使えばさらに安くなるんだけど、うちのタスクでは精度が許容範囲を下回った。8割のケースはHaikuで十分なのかもしれない。でも残り2割がビジネス的に重要なドキュメントだったりするので、今のところSonnetで統一している。

観測可能性: 後から入れようとすると本当に地獄

正直に言うと、最初は「とにかく動くものを作ってから、後でモニタリングを入れよう」という考えで進めた。よくある話だと思う。

これが一番の間違いだった。

後からLogging・Tracing・Metricsを入れようとすると、コードのあちこちに手を入れる必要が出てきて、バグを仕込むリスクも上がる。設計段階から組み込むほうが、トータルのコストははるかに低い。

僕がいま使っているスタックは:

• LangSmith: LLMの入出力トレーシング専用。プロンプトのバージョン管理も地味に便利
• Prometheus + Grafana: ビジネスメトリクス（分類精度、スループット、コスト/リクエスト）
• Sentry: 例外トラッキング。コンテキスト長超過や不正な出力フォーマットもカスタムで捕捉

LangSmithは最初「LangChain使っていないと意味ないのでは」と思っていたんだけど、実は単体でも使えて、カスタムのOpenTelemetryスパンを送れる。これを知るのが半年遅かった。

from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree


@traceable(name="classify_document", run_type="chain")
def classify_document_with_tracing(doc_id: str, content: str) -> dict:
    run_tree = get_current_run_tree()

    # メタデータを追加しておくと、後でフィルタリング・集計が楽になる
    if run_tree:
        run_tree.add_metadata({
            "doc_id": doc_id,
            "content_length": len(content),
            "model": "claude-sonnet-4-6",
        })

    result = call_llm(content)

    # 精度計測用の信頼スコアも一緒に記録しておく
    if run_tree and result.get("confidence"):
        run_tree.add_metadata({"confidence_score": result["confidence"]})

    return result

One thing I noticed: LangSmithのダッシュボードで「失敗したトレース」を並べて眺めると、失敗パターンが目で見えてくる。「この種類のドキュメントで毎回同じところで落ちてるな」というのが一目瞭然で、これがなかったらデバッグに何倍も時間がかかっていたと思う。

コストの話をすると、1万リクエストあたりのLLMコストを追い続けることで、プロンプト変更の影響が数字でわかるようになった。感覚でやっていたときより、意思決定のスピードが全然違う。

フォールバックとサーキットブレーカー: 一番後回しにしてしまった部分

LLMのAPIは落ちる。レートリミットも来る。たまに信じられないくらいレスポンスが遅くなる。これ全部、本番で経験した。

サーキットブレーカーを入れる前は、APIが重くなったときにワーカーが全部タイムアウト待ちになって、その間ずっとリソースを食い続けていた。pybreakerライブラリを入れてからは、エラー率が閾値を超えたら即座にフォールバックに切り替えるようにした——設定チューニングが難しくて、最初は誤検知が多かったけど。開放状態に入る閾値と回復判定のタイムアウトのバランスは、自分のトラフィックパターンに合わせた調整が必要で、実際の負荷テストをしないと適切な値が出てこない。

うちのフォールバックは「ルールベースの簡易分類」で、LLMほど精度は高くないけど、システム全体が止まるよりは圧倒的にいい。B案が存在することの安心感が、想像以上に大きかった。

あと、モデルのバージョン固定はやっておいたほうがいい。claude-3-5-sonnet-latestのような可変エイリアスを使っていたら、モデルが更新された際に挙動が変わって気づかないことがある。claude-sonnet-4-6のように明示的なバージョンを指定して、意識的にアップデートするほうが精神的に楽だし、変更の影響をちゃんと計測できる。

結局、何をおすすめするか

迷っている人向けに、自分が今やり直すとしたら何をするか書いておく。

最初からやること:

リトライにはfull jitterを入れる（これだけで安定度が全然違う）。LLMの入出力トレーシングは初日から入れる。LangSmithかHoneyhiveかは好みでいい。そしてコスト/リクエストを追う仕組みを作る——スプレッドシートでもいいから数字にする。

ある程度動いてから考えること:

サーキットブレーカーとフォールバック設計、モデルの精度計測パイプライン（ゴールデンデータセットで定期評価）、あとバッチ処理できるタスクはバッチに寄せること。Anthropic Batches APIはコスト削減に効く。うちでは非同期タスクの約40%をバッチに移行して、LLMコストが月あたり約23%落ちた。

正直まだ自信がないこと:

数十万リクエスト/日のスケールでこの設計が通用するか、僕には検証できていない。マルチモーダル入力が混じったときのコスト予測精度も、まだ感覚値の域を出ていない。

Anyway — AIパイプラインの「本番」は、プロトタイプとは全く別の問題が出てくる。LLMを呼ぶこと自体は簡単になった。難しいのは、それを継続的に安定させながら、コストと品質のバランスを保ち続けること。

1万回動かして気づいたのは、結局ソフトウェアエンジニアリングの基本がそのまま通用するということ。観測可能性、フォールバック、テスト、段階的なデプロイ——新しい技術に見えても、問題の本質は変わらない。金曜のデプロイは今でもちょっと怖い。でも怖さが適切なチェックを生む、という意味では悪くないかもしれない。