企業AIサービス化の課題

企業におけるAIアプリケーションの普及に伴い、典型的なシナリオは次のとおりです。マーケティング部門はGPT-4を統合してコピー生成を行い、開発チームはDeepSeekを使用してコード補完を行い、カスタマーサービスシステムは自己学習された意図認識モデルを使用しています。これらのAIサービスはそれぞれ独立して動作し、統一された管理が欠如しています。APIキーは散在し、呼び出し量は監視できず、コストの配分は困難で、モデル切り替えにはコード変更が必要です。AIミドルウェアの目標は、これらの断片化問題を解決し、統一されたAI能力アクセス層を構築することです。

AIゲートウェイのコア機能

AIミドルウェアのコアはAIゲートウェイであり、以下の機能を備えるべきです。統一API（すべてのモデルは統一されたOpenAI互換インターフェースを介してアクセスされ、上流アプリケーションはバックエンドモデルを気にする必要はありません）、インテリジェントルーティング（リクエストタイプに基づいて最適なモデルを自動選択します。単純なQ&Aは低コストモデル、複雑な推論は高度なモデル）、レート制限とクォータ（テナント/アプリケーションごとのレート制限とトークンクォータ管理）、プロトコル変換（OpenAI形式のリクエストを各プロバイダーのネイティブ形式に変換）、キャッシュとデグラデーション（高頻度リクエストのキャッシュ、モデルが利用できない場合のバックアップモデルへのデグラデーション）。

AIゲートウェイのコア実装

import time, hashlib, json
from collections import defaultdict
from openai import OpenAI

class AIGateway:
    def __init__(self):
        self.models = {}           # モデルレジストリ
        self.rate_limits = {}      # テナントレート制限設定
        self.usage = defaultdict(list)  # 使用統計
        self.cache = {}

    def register_model(self, name, provider, model_id, api_key, base_url, cost_per_1k):
        """モデルを登録"""
        self.models[name] = {
            "provider": provider, "model_id": model_id,
            "client": OpenAI(api_key=api_key, base_url=base_url),
            "cost_per_1k": cost_per_1k
        }

    def set_rate_limit(self, tenant_id, rpm, tpm):
        """テナントレート制限を設定（毎分リクエスト数、毎分トークン数）"""
        self.rate_limits[tenant_id] = {"rpm": rpm, "tpm": tpm, "requests": [], "tokens": 0}

    def _check_limit(self, tenant_id):
        if tenant_id not in self.rate_limits:
            return True
        lim = self.rate_limits[tenant_id]
        now = time.time()
        lim["requests"] = [t for t in lim["requests"] if now - t < 60]
        if len(lim["requests"]) >= lim["rpm"]:
            return False
        return True

    def route(self, messages, tenant_id, max_cost=None):
        """インテリジェントルーティング—タスクの複雑さに基づいてモデルを選択"""
        msg_text = json.dumps(messages, ensure_ascii=False)
        if len(msg_text) < 200:
            preferred = ["deepseek-chat", "gpt-3.5-turbo"]  # 単純なタスクは低コストモデル
        else:
            preferred = ["gpt-4", "deepseek-chat"]  # 複雑なタスクは高度なモデル

        for model in preferred:
            if model in self.models and self._check_limit(tenant_id):
                return model
        return list(self.models.keys())[0]  # フォールバック

    def chat(self, tenant_id, model, messages, **kwargs):
        """レート制限とキャッシュを備えたLLM呼び出しを実行"""
        if not self._check_limit(tenant_id):
            return {"error": "rate limited", "retry_after": 60}

        # キャッシュチェック
        cache_key = hashlib.md5(
            json.dumps({"model":model,"msgs":messages},sort_keys=True).encode()
        ).hexdigest()
        if cache_key in self.cache:
            return self.cache[cache_key]

        m = self.models[model]
        resp = m["client"].chat.completions.create(
            model=m["model_id"], messages=messages, **kwargs
        )
        result = resp.choices[0].message.content

        # 統計を更新
        self.rate_limits.setdefault(tenant_id,
            {"rpm":9999,"tpm":999999,"requests":[],"tokens":0}
        )["requests"].append(time.time())

        self.cache[cache_key] = result
        return result

    def cost_report(self, tenant_id):
        """コストレポートを生成"""
        usage = self.usage.get(tenant_id, [])
        return {"total_calls": len(usage), "estimated_cost": sum(u.get("c
ost", 0) for u in usage)}

gw = AIGateway()
gw.register_model("deepseek-chat", "deepseek", "deepseek-chat",
                  "sk-xxx", "https://api.deepseek.com", 0.001)
gw.set_rate_limit("tenant-1", rpm=100, tpm=100000)
print(gw.chat("tenant-1", "deepseek-chat", [{"role":"user","content":"Hello"}]))


本番環境でのデプロイに関する考慮事項
AIゲートウェイは本番環境では追加の注意が必要です：高可用性（マルチインスタンス展開+ヘルスチェック+自動フェイルオーバー）、リクエストキュー（トラフィック急増時に直接拒否せずキューイングし、キューの長さとタイムアウトを設定）、ストリーミング応答（SSE透過をサポートしてタイプライター効果を実現し、ストリームの中断と再接続を処理）、ログと監査（すべてのリクエストの完全な情報を記録してコスト計算と問題追跡に使用し、ユーザーデータのマスキングに注意）、マルチリージョン展開（レイテンシを減らすために最寄りのリージョンにルーティングし、クロスリージョンのフェイルオーバー）。
オープンソースAIゲートウェイソリューションの比較
AIゲートウェイをゼロから構築したくない場合、以下のオープンソースソリューションが注目に値します：LiteLLM — 最も人気のあるAIゲートウェイプロキシで、100以上のモデルプロバイダーに対して統一されたOpenAI形式のインターフェースをサポートし、負荷分散、レート制限、コスト追跡が組み込まれています。Portkey — 拡張されたAIゲートウェイで、プロンプト管理、A/Bテスト、キャッシュ、カナリアリリースを追加で提供します。Helicone — 可観測性に焦点を当てたAIゲートウェイで、リクエストログ、コスト分析、使用状況ダッシュボードを提供します。3つの選択に関する推奨事項：統一APIと基本的なレート制限のみが必要な場合→LiteLLM；完全なLLMOps機能が必要な場合→Portkey；コアニーズが可観測性とコスト分析の場合→Helicone。最終的に、LiteLLMと自社開発のルーティングモジュールの組み合わせを選択しました。LiteLLMはプロトコル統一と基本機能を処理し、自社開発モジュールは複雑なマルチモデルルーティングとデグラデーション戦略を担当します。
このスキルチェーンを自分で編成してみませんか？
スキルチェーンで開く →