LangGraphマルチモデルルーティング：エージェントワークフローのゲートウェイチェック

LangGraphマルチモデルルーティングは、第一にグラフ設計の問題であり、第二にゲートウェイの問題です。グラフはどのような作業が行われるかを決定し、モデルクライアントはどのOpenAI互換ルートを呼び出すかを決定し、ゲートウェイは各呼び出しが予算、ツール、リトライ、ロギング、および停止条件のルール内に収まっているかどうかを証明する必要があります。

Flatkeyはそのゲートウェイの境界に適合します。その公開サイトでは、https://router.flatkey.ai/v1/chat/completionsにOpenAI互換のチャットルートが示されており、1つのキーで、使用状況の可視化、請求管理、モデルルーティングが1か所で行えます。このガイドでは、すべてのアカウントでモデルエイリアス、ツールの動作、またはフォールバックルートがすでに検証されていることを前提とせずに、エージェントワークフローのための実用的なLangGraphマルチモデルルーティングのセットアップを示します。

LangGraphマルチモデルルーティングのクイックアンサー

安全なLangGraphマルチモデルルーティングのセットアップでは、ルーティングを3つのレイヤーに分離します。

LangGraphマルチモデルルーティングを1つの長いプロンプトの中に隠さないでください。ポリシーは、チームが検査できるグラフノード、条件付きエッジ、およびゲートウェイチェックに配置してください。

モデルルーティングがグラフの境界に属する理由

ほとんどのLangGraphの例は、グラフの制御フロー（状態、ノード、エッジ、ツール、永続性）に焦点を当てています。それは正しい出発点ですが、本番環境のエージェントには、プラットフォームチームや財務チームが判断できるルートの境界も必要です。

ワークフローに明らかに異なるコスト、レイテンシー、またはツールリスクのプロファイルがある場合は、グラフレベルのルートを使用します。

その境界こそ、Flatkeyが役立つ場所です。各ノードに個別のプロバイダーキーを散在させる代わりに、各ルートはOpenAI互換のAPIベースURLを呼び出すことができ、その間Flatkeyはゲートウェイレベルの可視性を維持します。隣接するポリシー設計については、FlatkeyのガイドAIエージェントゲートウェイコントロールを参照してください。

ルートを接続する前のソースチェック

テンプレートを本番コードとして扱う前に、最新のドキュメントと実際のアカウントで確認を行ってください。

以下のコードはテンプレートです。LangGraphマルチモデルルーティングポリシーをリリースする前に、現在のlangchain、langchain-openai、およびlanggraphのバージョンで実行してください。

Flatkeyをバックエンドとするチャットモデルの設定

グラフを実行する環境に最新のパッケージをインストールします。

pip install -U langgraph langchain langchain-openai

APIルートはエンドポイントパスとは別にしてください。LangChainクライアントはベースURLを受け取る必要があります。クライアントはOpenAI互換のチャット呼び出しのためにChat Completionsパスを追加します。

FLATKEY_API_KEY=sk-fk-your-key
FLATKEY_BASE_URL=https://router.flatkey.ai/v1
FLATKEY_FAST_MODEL=your-fast-model-alias
FLATKEY_REASONING_MODEL=your-reasoning-model-alias
FLATKEY_TOOL_MODEL=your-tool-capable-model-alias

次に、各ルートに対して明示的なモデルクライアントを作成します。

import os
from langchain.chat_models import init_chat_model

BASE_URL = os.getenv("FLATKEY_BASE_URL", "https://router.flatkey.ai/v1")
API_KEY = os.environ["FLATKEY_API_KEY"]

fast_model = init_chat_model(
    model=os.environ["FLATKEY_FAST_MODEL"],
    model_provider="openai",
    base_url=BASE_URL,
    api_key=API_KEY,
)

reasoning_model = init_chat_model(
    model=os.environ["FLATKEY_REASONING_MODEL"],
    model_provider="openai",
    base_url=BASE_URL,
    api_key=API_KEY,
)

tool_model = init_chat_model(
    model=os.environ["FLATKEY_TOOL_MODEL"],
    model_provider="openai",
    base_url=BASE_URL,
    api_key=API_KEY,
)

これは、LangGraphマルチモデルルーティングにとって最小限の有用な形です。各ルートには名前付きのモデルクライアントがありますが、すべての呼び出しは同じゲートウェイの境界を使用します。別のエイリアスがストリーミングやツールをサポートしているからといって、あるモデルエイリアスもサポートしていると想定しないでください。各ルートを個別にテストしてください。

LangGraphルーティングテンプレートの構築

分類ノードを使用してルートを状態に書き込みます。次に、条件付きエッジによって次に実行するノードを決定させます。

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages


class RouteState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    route: Literal["simple", "complex", "tools", "stop"]
    budget_cents: float
    errors: int
    tool_calls: int


@tool
def lookup_account_status(account_id: str) -> str:
    """テンプレートツール。実際の読み取り専用ルックアップに置き換えてください。"""
    return "active"


tool_ready_model = tool_model.bind_tools([lookup_account_status])


def choose_route(state: RouteState) -> dict:
    last_message = state["messages"][-1].content.lower()

    if state.get("budget_cents", 0) <= 0:
        return {"route": "stop"}
    if state.get("errors", 0) >= 2:
        return {"route": "stop"}
    if "lookup" in last_message or "account" in last_message:
        return {"route": "tools"}
    if len(last_message) > 600:
        return {"route": "complex"}
    return {"route": "simple"}


def simple_answer(state: RouteState) -> dict:
    return {"messages": [fast_model.invoke(state["messages"])]}


def complex_answer(state: RouteState) -> dict:
    return {"messages": [reasoning_model.invoke(state["messages"])]}


def tool_answer(state: RouteState) -> dict:
    if state.get("tool_calls", 0) >= 3:
        return {"messages": [AIMessage(content="Tool-call limit reached.")] }
    return {"messages": [tool_ready_model.invoke(state["messages"])]}


workflow = StateGraph(RouteState)
workflow.add_node("choose_route", choose_route)
workflow.add_node("simple_answer", simple_answer)
workflow.add_node("complex_answer", complex_answer)
workflow.add_node("tool_answer", tool_answer)

workflow.set_entry_point("choose_route")
workflow.add_conditional_edges(
    "choose_route",
    lambda state: state["route"],
    {
        "simple": "simple_answer",
        "complex": "complex_answer",
        "tools": "tool_answer",
        "stop": END,
    },
)
workflow.add_edge("simple_answer", END)
workflow.add_edge("complex_answer", END)
workflow.add_edge("tool_answer", END)

graph = workflow.compile()

このテンプレートにより、LangGraphマルチモデルルーティングが可視化されます。レビュー担当者は、プロンプトをリバースエンジニアリングすることなく、リクエストがsimple_answer、complex_answer、tool_answer、またはstopに送られた理由を確認できます。

上記のtool_answerノードは、ツールコールを提案できるモデルルートを検証します。グラフでツールを実行する必要がある場合は、そのモデルノードの後にLangGraphのToolNodeまたは独自のエグゼキュータを追加し、同じ予算と停止条件チェックを使用してツール結果をグラフにルーティングして戻します。

すべてのモデルコールの前後にゲートウェイチェックを追加する

グラフのルートだけをガードレールにすべきではありません。モデルコールの前に予算をチェックし、モデルコールの後に証拠を記録する小さなラッパーを追加します。

from dataclasses import dataclass
from time import time


@dataclass
class GatewayDecision:
    route: str
    model_alias: str
    allowed: bool
    reason: str
    started_at: float


def preflight(route: str, model_alias: str, state: RouteState) -> GatewayDecision:
    if state.get("budget_cents", 0) <= 0:
        return GatewayDecision(route, model_alias, False, "budget_exhausted", time())
    if route == "tools" and state.get("tool_calls", 0) >= 3:
        return GatewayDecision(route, model_alias, False, "tool_limit_reached", time())
    if state.get("errors", 0) >= 2:
        return GatewayDecision(route, model_alias, False, "retry_limit_reached", time())
    return GatewayDecision(route, model_alias, True, "allowed", time())


def record_gateway_result(decision: GatewayDecision, response: BaseMessage) -> None:
    # これを構造化ロギングに置き換えます。編集済みのキーラベル、
    # ルート名、モデルエイリアス、リクエストのタイムスタンプ、レスポンスのメタデータを含めます。
    print(
        {
            "route": decision.route,
            "model_alias": decision.model_alias,
            "allowed": decision.allowed,
            "reason": decision.reason,
            "started_at": decision.started_at,
            "response_type": response.type,
        }
    )

本番環境では、ローカルログだけで止めないでください。タイムスタンプとモデルエイリアスをFlatkeyの使用状況ログまたはリクエストログと照合します。それが、LangGraphマルチモデルルーティングが意図したゲートウェイルートを使用したという証拠になります。

ツールを多用するルートを信頼する前にツールをチェックする

ツールを多用するノードには、独自の検証パケットが必要です。モデルは通常のチャットには優れているかもしれませんが、ツールスキーマに失敗したり、ツールの差分を異なる方法でストリーミングしたり、エグゼキュータが拒否するツール引数を生成したりする可能性があります。より詳細な互換性チェックについては、Flatkeyのプロバイダー間のツール呼び出しの互換性に関するガイドを使用してください。

ここはまた、ルートがChat Completionsに属するのか、それともResponsesに属するのかを決定する場所でもあります。エンドポイントファミリーは分けておきます。Responsesスタイルのエージェントルートを評価している場合は、このガイドをFlatkeyのOpenAI互換Responses APIルーターと組み合わせてください。

リトライ、フォールバック、停止条件のポリシー

リトライはルーティングのミスを隠してしまう可能性があります。厳格なLangGraphマルチモデルルーティングポリシーでは、いつリトライし、いつフォールバックし、いつ停止するかを規定すべきです。

LangChainミドルウェアは、エージェントレベルのリトライ、フォールバック、呼び出し制限に役立ちます。手動で構築されたLangGraphワークフローの場合、モデルを呼び出すノードの近くに同じポリシーを保持し、その結果をFlatkeyログで証明します。

公開前のテストチェックリスト

Flatkeyをエージェントのデフォルトルートにする前に、これらのチェックを実行してください。

1. 意図したキーとモデルエイリアスを使用して、https://router.flatkey.ai/v1/chat/completionsに直接curlリクエストを1回送信します。
2. LangGraphの外部で各LangChainモデルクライアントを1回呼び出します。
3. シンプルなルート用に1つ、複雑なルート用に1つ、ツールを多用するルート用に1つのプロンプトでグラフを実行します。
4. budget_centsを0に設定して、停止パスを強制します。
5. errorsを許容される最大値に設定して、リトライ制限を強制します。
6. 各リクエストが、期待されるタイムスタンプ、キーラベル、ルート名、モデルエイリアスとともにFlatkeyログに表示されることを確認します。
7. ワークフローの予算を承認する所有者とコストまたは使用状況のフィールドを比較します。そのレビューを開始するには、Flatkeyの料金ページが適切な公開場所です。

このチェックリストの出力は、LangGraphマルチモデルルーティングの受け入れパケットになります。これは、グラフ、モデルクライアント、ゲートウェイがすべて同意していることを証明するため、機能しているプロンプトのスクリーンショットよりも有用です。

LangGraphマルチモデルルーティングのトラブルシューティング

最初のルートが失敗した場合は、デバッグ対象を小さく保ちます：キー、ベースURL、モデルエイリアス、1つのモデルクライアント、1つのグラフノード、および1つのFlatkeyログ検索。

よくある質問

LangGraphマルチモデルルーティングはプロバイダーのフェイルオーバーと同じですか？

いいえ。LangGraphマルチモデルルーティングは、グラフの実行前または実行中に、どのルートがタスクに適しているかを決定します。プロバイダーのフェイルオーバーは、選択されたルートが完了できない場合に発生します。両方を使用できますが、別々にテストする必要があります。

すべてのLangGraphノードで異なるモデルを使用すべきですか？

いいえ。2つか3つの明示的なルートから始めてください。モデルエイリアスが多すぎると、ログ、予算、ロールバックの確認が難しくなります。

FlatkeyをChat CompletionsとResponsesの両方のルートで使用できますか？

FlatkeyはOpenAI互換のエンドポイントファミリーを公開しますが、各クライアントパスには独自のテストが必要です。特にツール呼び出し、ストリーミング、使用状況フィールド、フォールバック動作については、Chat CompletionsとResponsesのチェックを分けてください。

最終チェックリスト

LangGraphマルチモデルルーティングのセットアップがテンプレートの域を超える準備ができたのは、次の場合です。

1. グラフのルートが状態とログで確認できる。
2. 各ルートが名前付きモデルエイリアスにマッピングされている。
3. FlatkeyキーがOpenAI互換のベースURLを介して選択された各エイリアスを呼び出すことができる。
4. ツールを多用するルートが、スキーマ、引数、上限、エラーのテストに合格する。
5. 予算と再試行の停止条件が、次のモデル呼び出しの前に実行される。
6. Flatkeyの使用状況またはリクエストログで、選択されたルートが確認できる。
7. ロールバックがアプリケーションの書き換えではなく、1つの設定変更で済む。

LangGraphエージェントのトラフィックを1つのゲートウェイ経由でルーティングする準備ができたら、Flatkeyキーを取得し、単一の低リスクなグラフルートから始め、ログでポリシーが機能することを確認した後にのみ拡張してください。