如何构建自动切换模型的AI应用

大型语言模型（LLM）从根本上改变了我们构建现代软件的方式。

但是，对每个用户请求都依赖单一AI模型会带来严重的生产风险。API中断时有发生。对于简单任务，专有模型可能成本高昂。而较便宜的开源模型可能难以处理复杂的逻辑推理。

当我和我的团队为客户支持平台构建企业级AI引擎时，我们所有事情都依赖一个顶级模型。

一个月内，我们遇到了两个严重问题：一次大范围的API中断完全冻结了我们的应用，而且我们每月的API账单上涨，因为我们用昂贵的推理模型来回答简单的常见问题。

为了解决这个问题，我构建了一个弹性多模型编排器。在本指南中，您将学习如何使用Python构建一个智能的多层AI应用，它能够动态路由提示词并自动处理模型回退。

• 内容概览

• 前提条件和环境设置

  • 包安装

  • 本地目录结构

  • 环境配置

• 单一模型架构的问题

• 理解动态模型路由生命周期

• 步骤1：实现第1层 – 提示词复杂度与意图分析

  • 解析第1层的代码逻辑

• 步骤2：实现第2层 – 动态模型路由逻辑

  • 解析第2层的代码逻辑

• 步骤3：实现第3层 – 自动回退

  • 解析第3层的代码逻辑

  • 将架构整合为统一执行流水线

  • 解析代码逻辑

• 生产环境中动态模型切换的经验教训

• 结论

  • 感谢您的阅读！

前提条件和环境设置

要按照本教程操作，您需要具备以下环境：

• 具备Python和异步编程的基本技能。

• 系统上安装Python 3.9或更高版本。

• 代码编辑器，如Visual Studio Code。

• 至少两个模型提供商的API密钥（例如OpenAI和Anthropic），或通过Ollama运行的本地模型。

包安装

打开终端并安装所需依赖：

pip install openai anthropic python-dotenv pydantic

本地目录结构

按如下方式组织你的项目目录，以保持代码整洁：

ai-model-router/

│

├── .env

├── README.md

└── app.py

环境配置

在你的项目目录根目录下创建一个 .env 文件，并添加你的凭据：

Ini, TOML

OPENAI_API_KEY=your_openai_api_key_here ANTHROPIC_API_KEY=your_anthropic_api_key_here ENVIRONMENT=development

单一模型架构的问题

如果你将每个查询都路由到旗舰模型（如GPT-4o或Claude 3.5 Sonnet），你会在简单任务上过度支出。相反，如果你为了省钱而将所有内容路由到更小、更快的模型（如GPT-4o-mini或Claude 3.5 Haiku），当用户提交复杂的代码生成或分析任务时，你的系统将失败。

除了成本问题，单一模型系统还面临单点故障。当API提供商宕机或限制你的账户速率时，你的整个应用都会崩溃。

要解决这个问题，你需要一个编排层，在调用LLM之前评估提示的复杂性，将请求路由到最具成本效益的模型，并在主提供商失败时回退到备用提供商。

理解动态模型路由生命周期

以下是用户请求在动态多模型系统中的流转过程：

首先是复杂性分析。系统使用轻量级指标检查传入的提示，以分配任务等级（简单、中等或复杂）。

其次是模型路由。系统将等级映射到相应的模型（例如，轻量级任务发送到Haiku/Mini，而繁重推理任务发送到Sonnet/GPT-4o）。

还需要自动回退：如果主提供商超时或抛出API错误，系统会自动将查询重定向到等效的回退模型。

第1步：实现第1层——提示复杂性与意图分析

首先，你需要一种确定性的、快速的方法来对提示进行分类，而不需要为了决定使用哪个模型而进行昂贵的API调用。

在花钱调用LLM API之前，我们可以在代码中直接查看文本。把这一步想象成一个智能门卫。通过检查文本长度、代码片段或棘手的关键词等简单特征，我们可以在毫秒内免费判断任务的难度。

以下是我们如何在app.py中设置分类规则的：

import re
from enum import Enum
from pydantic import BaseModel


class TaskComplexity(Enum):
    SIMPLE = "simple"      # FAQs, short summaries, basic translation
    MEDIUM = "medium"      # Standard text generation, content rewriting
    COMPLEX = "complex"    # Code writing, math logic, structural analysis


class PromptAnalyzer:
    def __init__(self):
        # Regex patterns indicative of complex tasks
        self.complex_keywords = [
            r"\brefactor\b",
            r"\bdebug\b",
            r"\bwrite code\b",
            r"\banalyze\b",
            r"\balgorithm\b",
            r"\barchitecture\b",
        ]

    def analyze_complexity(self, prompt: str) -> TaskComplexity:
        """
        Evaluates input text deterministically to output
        a TaskComplexity rating.
        """
        normalized = prompt.lower().strip()
        word_count = len(normalized.split())

        # Check for code blocks or complex request patterns
        contains_code = "```" in prompt
        has_complex_keyword = any(
            re.search(pattern, normalized)
            for pattern in self.complex_keywords
        )

        if contains_code or has_complex_keyword or word_count > 300:
            return TaskComplexity.COMPLEX
        elif word_count > 80:
            return TaskComplexity.MEDIUM
        else:
            return TaskComplexity.SIMPLE


# Example Usage
if __name__ == "__main__":
    analyzer = PromptAnalyzer()

    test_prompt = (
        "Write a Python script that implements a trie "
        "data structure with autocomplete."
    )

    complexity = analyzer.analyze_complexity(test_prompt)
    print(f"Prompt Complexity Tier: {complexity.value}")

解析第一层的代码逻辑

• TaskComplexity 枚举： 为传入请求定义明确的类别（SIMPLE、MEDIUM、COMPLEX），为整个流水线提供类型安全。

• 关键词匹配： PromptAnalyzer 类设置正则表达式模式，查找如 refactor、debug 或 algorithm 等表示高推理需求任务的动作词。

• analyze_complexity 中的确定性规则：

• 格式和长度检查：我们清理字符串，检查 Markdown 代码块（```），并计算单词数。

• 层级分配：

  • 如果提示词包含代码块、触发词或超过300个单词，则立即升级为 COMPLEX。

  • 如果提示词在80到300个单词之间且不含代码关键词，则映射为 MEDIUM。

  • 任何更短的内容默认归类为 SIMPLE。

使用复杂查询运行此代码片段时，会检查文本，识别出“write code”，并输出：

Prompt Complexity Tier: complex

第2步：实现第二层——动态模型路由逻辑

既然我们能够成功地将提示词标记为简单、中等或复杂，我们需要一本规则手册来决定实际由哪个AI模型处理它。

这一层将每个复杂层级映射到一个主模型和一个备用回退模型。例如，简单查询路由到经济型模型（gpt-4o-mini），而复杂请求则路由到重量级模型（claude-3-5-sonnet）。

同时添加此配置：

class ModelConfig(BaseModel):
    provider: str
    model_name: str


class ModelRouter:
    def __init__(self):
        # Map task complexity tiers to primary and fallback models
        self.routing_table = {
            TaskComplexity.SIMPLE: {
                "primary": ModelConfig(
                    provider="openai",
                    model_name="gpt-4o-mini",
                ),
                "fallback": ModelConfig(
                    provider="anthropic",
                    model_name="claude-3-5-haiku-20241022",
                ),
            },
            TaskComplexity.MEDIUM: {
                "primary": ModelConfig(
                    provider="openai",
                    model_name="gpt-4o-mini",
                ),
                "fallback": ModelConfig(
                    provider="anthropic",
                    model_name="claude-3-5-haiku-20241022",
                ),
            },
            TaskComplexity.COMPLEX: {
                "primary": ModelConfig(
                    provider="anthropic",
                    model_name="claude-3-5-sonnet-20241022",
                ),
                "fallback": ModelConfig(
                    provider="openai",
                    model_name="gpt-4o",
                ),
            },
        }

    def get_models_for_tier(
        self, complexity: TaskComplexity
    ) -> tuple[ModelConfig, ModelConfig]:
        """
        Returns the primary and fallback models for a given
        task complexity tier.
        """
        config = self.routing_table[complexity]
        return config["primary"], config["fallback"]

解析第二层的代码逻辑

• ModelConfig 模式： 使用 Pydantic 确保每个模型定义都同时包含一个 provider（例如 "openai"）和一个特定的 model_name 字符串。

• self.routing_table 映射： 这个字典充当模型分配的单一事实来源：

  • SIMPLE & MEDIUM 层： 主要目标是 gpt-4o-mini，用于高吞吐量、低成本输出。如果 OpenAI 失败，则回退到 Anthropic 的 claude-3-5-haiku-20241022。

  • COMPLEX 层： 主要目标切换为 claude-3-5-sonnet-20241022，用于顶级代码生成和推理，以 gpt-4o 作为备用。

• get_models_for_tier： 一个辅助函数，接收分析后的层，并安全地返回 (PrimaryModel, FallbackModel) 元组。

第 3 步：实现第三层 – 自动回退

即使最优秀的 AI 提供商也会遭遇停机、速率限制或意外超时。生产级应用不能在出现这种情况时直接向用户抛出错误屏幕。我们需要一个执行引擎，它尝试调用主要模型提供商并自动捕获错误。如果出现任何问题，它会立即转向次要备用模型，而不会中断工作流。

将执行引擎代码添加到脚本中：

import os
import time

from anthropic import Anthropic, APIError as AnthropicAPIError
from dotenv import load_dotenv
from openai import OpenAI, APIError as OpenAIAPIError

load_dotenv()


class ResilientModelEngine:
    def __init__(self):
        self.openai_client = OpenAI(
            api_key=os.getenv("OPENAI_API_KEY", "dummy")
        )
        self.anthropic_client = Anthropic(
            api_key=os.getenv("ANTHROPIC_API_KEY", "dummy")
        )

    def _call_openai(self, model: str, prompt: str) -> str:
        response = self.openai_client.chat.completions.create(
            model=model,
            messages=[
                {
                    "role": "user",
                    "content": prompt,
                }
            ],
            timeout=10.0,
        )
        return response.choices[0].message.content

    def _call_anthropic(self, model: str, prompt: str) -> str:
        response = self.anthropic_client.messages.create(
            model=model,
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": prompt,
                }
            ],
            timeout=10.0,
        )
        return response.content[0].text

    def execute_provider_call(
        self,
        config: ModelConfig,
        prompt: str,
    ) -> str:
        """
        Dispatches prompt execution to the correct provider SDK.
        """
        if config.provider == "openai":
            return self._call_openai(config.model_name, prompt)
        elif config.provider == "anthropic":
            return self._call_anthropic(config.model_name, prompt)
        else:
            raise ValueError(
                f"Unsupported provider: {config.provider}"
            )

    def execute_with_fallback(
        self,
        primary: ModelConfig,
        fallback: ModelConfig,
        prompt: str,
    ) -> tuple[str, str]:
        """
        Attempts execution on the primary model and switches to the
        fallback model if the primary provider fails.

        Returns:
            tuple[str, str]: (Response text, Model used)
        """
        try:
            print(
                f"[Attempt] Calling Primary Provider: "
                f"{primary.provider} ({primary.model_name})"
            )

            result = self.execute_provider_call(primary, prompt)

            return result, (
                f"{primary.provider}:{primary.model_name}"
            )

        except (
            OpenAIAPIError,
            AnthropicAPIError,
            Exception,
        ) as e:
            print(f"[WARNING] Primary call failed due to: {e}")

            print(
                f"[Fallback] Switching to Secondary Provider: "
                f"{fallback.provider} ({fallback.model_name})"
            )

            try:
                result = self.execute_provider_call(
                    fallback,
                    prompt,
                )

                return result, (
                    f"{fallback.provider}:"
                    f"{fallback.model_name} (Fallback)"
                )

            except Exception as fallback_error:
                raise RuntimeError(
                    "Both primary and fallback systems failed. "
                    f"Error: {fallback_error}"
                )

分解第三层的代码逻辑

提供者客户端（_call_openai & _call_anthropic）：辅助方法封装了提供者的SDK调用，建立统一的严格10秒超时。如果API挂起，它会快速中止，以便在不使用户等待的情况下启动回退。

execute_provider_call 调度器：作为抽象桥接，将请求的提供者字符串匹配到其相应的API方法。

execute_with_fallback 弹性逻辑：首先在try块中执行主提供者。通过提供者特定的异常（OpenAIAPIError、AnthropicAPIError）捕获API错误、速率限制或网络超时。在except块中逻辑上将执行重定向到回退提供者。仅当主提供者和回退提供者都失败时，才引发不可恢复的RuntimeError。如果主提供者遇到问题，控制台会透明地跟踪恢复过程：

[尝试] 调用主提供者：anthropic (claude-3-5-sonnet-20241022)

[警告] 主调用失败，原因：连接超时

[回退] 切换到备用提供者：openai (gpt-4o)

将架构组合到统一执行流水线中

现在你可以将这三层组合成一个统一的流水线。

使用此编排类完成你的app.py脚本：

class SmartAIEngine:
    def __init__(self):
        self.analyzer = PromptAnalyzer()
        self.router = ModelRouter()
        self.executor = ResilientModelEngine()

    def process_request(self, user_prompt: str) -> dict:
        print("\n==========================================")
        print("Processing New Request")
        print("==========================================")

        # Step 1: Analyze prompt complexity
        complexity = self.analyzer.analyze_complexity(
            user_prompt
        )
        print(
            f"[Step 1] Prompt classified as: "
            f"{complexity.value.upper()}"
        )

        # Step 2: Determine routing target
        primary_model, fallback_model = (
            self.router.get_models_for_tier(
                complexity
            )
        )

        print(
            f"[Step 2] Selected Primary: "
            f"{primary_model.model_name}"
        )

        # Step 3: Execute request with resilient fallbacks
        response_text, executed_model = (
            self.executor.execute_with_fallback(
                primary=primary_model,
                fallback=fallback_model,
                prompt=user_prompt,
            )
        )

        return {
            "status": "success",
            "complexity_tier": complexity.value,
            "model_used": executed_model,
            "response": response_text,
        }


# Execution Pipeline Test
if __name__ == "__main__":
    engine = SmartAIEngine()

    # Query 1: Simple task
    simple_query = (
        "What is the capital of Japan? "
        "Answer in one word."
    )

    result_1 = engine.process_request(
        simple_query
    )

    print(f"Model Used: {result_1['model_used']}")
    print(f"Response: {result_1['response']}")

    # Query 2: Complex task
    complex_query = (
        "Write a Python function to debug a "
        "memory leak in a multithreaded "
        "application."
    )

    result_2 = engine.process_request(
        complex_query
    )

    print(f"Model Used: {result_2['model_used']}")
    print(
        f"Response Snippet: "
        f"{result_2['response'][:100]}..."
    )

解析代码逻辑

• 统一编排（SmartAIEngine）：将三个模块化组件——PromptAnalyzer、ModelRouter和ResilientModelEngine——初始化为实例属性。

• 流水线步骤：

  • 分析：离线评估提示字符串以确定复杂度等级。

  • 路由：根据该等级解析主要和次要模型对。

  • 执行：以弹性方式调用模型并捕获失败场景。

• 标准化响应负载：将执行细节包装到一致的输出字典中，跟踪模型使用情况、复杂度分类和输出文本。

生产环境中动态模型切换的经验教训

构建动态AI路由系统让我们的团队学到了关于企业级LLM架构的关键经验：

首先，保持分类轻量。永远不要使用大型LLM调用来为小任务分类提示。使用正则表达式、关键词匹配和令牌长度规则。你的分类器应在5毫秒内运行。

其次，规范化系统输出。不同的模型提供商对输出的结构不同。确保你的应用程序在将数据返回用户界面之前，将响应包装在一致的架构中。

第三，设置严格的超时。提供商API经常挂起而不是立即抛出错误。在主要模型调用上设置严格的请求超时（5到10秒），以便回退机制能快速触发，不会让最终用户感到沮丧。

最后，跟踪使用指标。记录每次路由决策、模型回退和成本差异。这些数据将揭示你的复杂度阈值是否随时间得到了正确调整。

结论

随着AI应用的扩展，依赖单一的、单体式LLM变得不可持续。智能模型路由允许你在不牺牲响应质量的情况下平衡性能、延迟和成本。

通过将你的应用程序与特定模型提供商解耦，并引入自动化路由层、输入评估、提供商抽象和弹性回退，你可以构建具有成本效益、快速且具有弹性的生产级AI系统。

在部署你自己的应用程序时，将LLM提供商视为动态工具。使用轻量级模型处理日常任务，保留旗舰模型用于复杂任务，并在代码中干净地处理提供商切换。

感谢阅读！

我希望这篇文章能让你对多模型编排器和动态路由在实际应用中的工作原理，以及如何在你的项目中开始实现它们，有一个实际的理解。

如果你想讨论AI工程、智能体AI、LLM、RAG、MLOps、企业AI架构或AI治理，欢迎关注、点赞、分享并联系我：

• LinkedIn

• 探索我的GitHub仓库

📌 相关推荐

GraphRAG 是推理问题，而非数据库问题

2026/8/30

构建市场时光机：使用 Python 和 WebSocket 重放交易会话

2026/8/30

如何自行基准测试LLM推理：值得信赖的数字设计标准

2026/8/30