LangGraph 多模型路由：代理工作流的网关检查

LangGraph 多模型路由首先是一个图设计问题，其次是一个网关问题。图决定了正在发生什么样的工作，模型客户端决定调用哪个与 OpenAI 兼容的路由，而网关则应证明每次调用是否都遵守了预算、工具、重试、日志记录和停止条件规则。

Flatkey 正好适用于那个网关边界。其公共站点在 https://router.flatkey.ai/v1/chat/completions 展示了一个与 OpenAI 兼容的聊天路由，将一个密钥、使用情况可见性、计费控制和模型路由集中在一处。本指南展示了一个用于代理工作流的实用 LangGraph 多模型路由设置，而无需假设每个模型别名、工具行为或备用路由都已在您的账户中得到验证。

LangGraph 多模型路由的快速解答

一个安全的 LangGraph 多模型路由设置将路由分为三个层次。

不要将 LangGraph 多模型路由隐藏在一个长长的提示中。将策略放在您的团队可以检查的图节点、条件边和网关检查中。

为什么模型路由属于图边界

大多数 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 兼容的聊天调用添加聊天补全路径。

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="已达到工具调用上限。")] }
    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 请求。
2. 在 LangGraph 之外调用每个 LangChain 模型客户端一次。
3. 为简单路由、复杂路由和重工具路由各运行一次图，每次使用一个提示。
4. 通过将 budget_cents 设置为 0 来强制执行停止路径。
5. 通过将 errors 设置为允许的最大值来强制执行重试限制。
6. 确认每个请求都出现在 Flatkey 日志中，并带有预期的时间戳、密钥标签、路由名称和模型别名。
7. 与批准工作流预算的所有者比较成本或使用情况字段。Flatkey 定价页面是开始该审查的合适公共场所。

此清单的输出将成为 LangGraph 多模型路由的验收包。它比工作提示的屏幕截图更有用，因为它证明了图、模型客户端和网关都达成了一致。

LangGraph 多模型路由故障排除

如果第一个路由失败，请保持调试范围小：密钥、基础 URL、模型别名、一个模型客户端、一个图节点和一次 Flatkey 日志查找。

常见问题解答

LangGraph 多模型路由与提供商故障转移相同吗？

不相同。LangGraph 多模型路由在图执行之前或期间决定哪个路由适合任务。提供商故障转移是在选定路由无法完成时发生的情况。您可以同时使用两者，但应分开测试。

每个 LangGraph 节点都应该使用不同的模型吗？

不。从两到三个明确的路由开始。太多的模型别名会使日志、预算和回滚更难审查。

我可以在聊天补全和响应路由中同时使用 Flatkey 吗？

Flatkey 公开了与 OpenAI 兼容的端点系列，但每个客户端路径仍然需要自己的测试。请将聊天补全和响应的检查分开，特别是对于工具调用、流式传输、使用情况字段和回退行为。

最终核对清单

当满足以下条件时，您的 LangGraph 多模型路由 设置就可以超越模板了：

1. 图路由在状态和日志中可见。
2. 每个路由都映射到一个命名的模型别名。
3. Flatkey 密钥可以通过与 OpenAI 兼容的基础 URL 调用每个选定的别名。
4. 大量使用工具的路由通过了模式、参数、上限和错误测试。
5. 预算和重试停止条件在下一次模型调用之前运行。
6. Flatkey 使用情况或请求日志确认了所选路由。
7. 回滚是一项配置更改，而不是重写应用程序。

当您准备好通过一个网关路由 LangGraph 代理流量时，获取一个 Flatkey 密钥，从单个低风险的图路由开始，只有在日志证明策略有效后再进行扩展。