如何构建能够自动切换模型的人工智能应用程序

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

但是，如果对于每个用户请求都依赖同一个人工智能模型，就会带来严重的生产风险。API可能会出现故障；对于一些简单的任务来说，使用专有模型可能会耗费高昂的成本；而那些价格较低的开源模型则可能在处理复杂的逻辑推理时遇到困难。

当我所在的团队为我们的客户支持平台开发一个企业级的人工智能引擎时，我们选择对所有场景都使用同一个顶级模型。

然而仅仅一个月后，我们就遇到了两个严重的问题：一次大规模的API故障导致我们的应用程序完全无法正常运行；同时，由于我们使用了价格较高的模型来回答一些简单的常见问题，我们的每月API使用费用也大幅增加了。

为了解决这些问题，我设计了一个具备高度弹性的多模型调度系统。在本书中，你将学习如何使用Python构建一个智能的多层架构人工智能应用系统——该系统能够动态地选择合适的模型来处理用户的请求，并在必要时自动切换到备用模型。

• 我们将涵盖哪些内容

• 先决条件与环境配置

  • 软件包安装

  • 本地目录结构

  • 环境配置

• 单一模型架构存在的问题

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

• 步骤1：实现第一层功能——提示语复杂度分析与用户意图识别

  • 分析第一层的代码逻辑

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

  • 分析第二层的代码逻辑

• 步骤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=你的OpenAI API密钥 ANTHROPIC_API_KEY=你的Anthropic API密钥 ENVIRONMENT=development

如果将所有请求都路由到像GPT-4o或Claude 3.5 Sonnet这样的旗舰模型，那么在处理简单任务时就会造成不必要的资源浪费；相反，如果为了节省成本而将所有请求都发送给规模较小、运行速度较快的模型（如GPT-4o-mini或Claude 3.5 Haiku），那么当用户提交复杂的代码生成或分析任务时，系统就会出现故障。

除了成本问题之外，单模型系统还存在单点故障的风险。一旦某个API提供商出现故障或对您的账户设置了速率限制，整个应用程序就会崩溃。

为了解决这些问题，您需要一个调度层——该层能够在调用大语言模型之前评估请求的复杂度，将请求路由到最经济高效的模型上；如果主供应商出现问题，系统还可以自动切换到备用提供商。

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

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

首先，我们需要进行复杂性分析。系统会使用一些简单的指标来评估输入提示的难度，并据此将其划分为“简单”“中等”或“复杂”三类。

其次，系统会将这些分类结果与相应的模型关联起来——例如，简单的任务会由Haiku/Mini模型处理，而需要复杂推理的任务则会由Sonnet/GPT-4o模型来处理。

此外，系统还配备了自动回退机制：如果主处理模块出现超时或错误，系统会自动将请求转发到备用的模型进行处理。

步骤1：实现第一层级分析——提示复杂性分析与意图识别

首先，我们需要一种快速、可靠的方法来对提示进行分类，而无需通过昂贵的API调用来确定应该使用哪种模型。

在花费金钱去调用大型语言模型API来了解用户的需求之前，我们完全可以在代码中直接分析文本内容。可以把这一步看作是一种“智能过滤器”：通过检查文本长度、代码片段或某些关键词汇，我们就能在几毫秒内免费判断出任务的难度。

以下是我们如何在`app.py`文件中设置这些分类规则：

import re
from enum import Enum
from pydantic import BaseModel


class TaskComplexity(Enum):
    SIMPLE = "简单"      # 常见问题解答、简短总结、基础翻译
    MEDIUM = "中等"      # 标准文本生成、内容改写
    COMPLEX = "复杂"    # 代码编写、数学逻辑分析、结构评估


class PromptAnalyzer:
    def __init__(self):
        # 用于识别复杂任务的正则表达式模式
        self.complex_keywords = [
            r"\b重构\b",
            r "\b调试\b",
            r "\b编写代码\b",
            r "\b分析\b",
            r "\b算法\b",
            r "\b架构设计\b",
        ]

    def analyze_complexity(self, prompt: str) -> TaskComplexity:
        """
        通过确定性方法分析输入文本，并返回相应的复杂性等级。
        """
        normalized = prompt.lower().strip()
        word_count = len(normalized.split())

        # 检查文本中是否包含代码块或复杂的指令结构
        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


# 使用示例
if __name__ == "__main__":
    analyzer = PromptAnalyzer()

    test_prompt = (
        "编写一个Python脚本，实现带有自动完成功能的trie数据结构。"
    )

    complexity = analyzer.analyze_complexity(test.prompt)
    print(f"提示的复杂性等级：{complexity.value}")

第一层级的代码逻辑解析

• TaskComplexity 枚举类型： 为传入的请求定义了明确的分类类别（SIMPLE、MEDIUM、COMPLEX），从而确保我们的处理流程具有类型安全性。

• 关键词匹配： PromptAnalyzer 类会设置正则表达式模式，用于识别诸如 refactor、debug 或 algorithm 这类表明需要进行复杂推理的指令词。

• 在analyze_complexity中的确定性规则：

• 格式化与长度检查：我们会清理输入文本，检测其中的 Markdown 代码块（```），并计算单词数量。

• 层级划分规则：

  • 如果提示信息中包含代码块、触发词，或者字数超过 300 个字，那么该请求会立即被归类为 COMPLEX 类型。

  • 如果提示信息的字数在 80 到 300 个字之间，并且不包含任何与代码相关的关键词，那么它会被归类为 MEDIUM 类型。

  • 如果提示信息的字数少于 80 个字，那么它默认被归类为 SIMPLE 类型。

使用包含复杂查询内容的测试样例进行测试时，系统会检测到“write code”这个关键词，并输出如下结果：

提示信息的复杂性等级：COMPLEX

步骤 2：实现二级动态模型路由逻辑

现在我们已经能够成功地将提示信息划分为简单、中等或复杂三种类型，接下来我们需要制定一套规则来确定究竟应该使用哪种 AI 模型来处理这些请求。

这一层机制会将不同的复杂性等级与相应的主模型及备用模型关联起来。例如，简单类型的查询会路由到预算规划相关的模型（如 gpt-4o-mini），而复杂类型的请求则会路由到功能更强大的模型（如 claude-3-5-sonnet）。

还需要添加以下配置：

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


class ModelRouter:
    def __init__(self):
        # 将任务复杂性等级与主模型及备用模型关联起来
        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]:
        """
        返回与给定任务复杂性等级相对应的主模型及备用模型。
        """
        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:
        """
        将提示信息发送给相应的模型提供者进行处理。
        """
        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"不支持的提供者：{configprovider}"
            )

    def execute_with_fallback(
        self,
        primary: ModelConfig,
        fallback: ModelConfig,
        prompt: str,
    ) -> tuple[str, str]:
        """
        先尝试使用主要模型进行处理，如果主要模型出现故障，则切换到备用模型。
        返回结果为：
            tuple[str, str]: (处理结果文本, 使用的模型)
        """
        try:
            print(
                f"[尝试] 调用主要模型提供者："
                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"[警告] 主要模型调用失败，原因：{e}")

            print(
                f "[切换到备用模型] 正在使用备用提供者："
                f "{fallback-provider} ({fallback.model_name})"
            )

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

                return result, (
                    f"{fallback-provider}:"
                    f"{fallback.model_name} (备用模型)"
                )

            except Exception as fallback_error:
                raise RuntimeError(
                    "主要模型和备用模型都出现了故障。
                    错误原因：{fallback_error}"
                )

解析第三层的代码逻辑

提供者客户端（_call_openai 和 _call_anthropic）：辅助方法会封装对提供者SDK的调用，并设置统一的10秒超时限制。如果某个API出现故障，系统会立即终止该请求，从而让备用方案能够迅速生效，而不会让用户等待过久。

execute_provider_call 分配器：充当抽象层，将用户请求的提供者名称与相应的API方法进行匹配。

execute_with_fallback 容错逻辑：首先在try块中尝试使用主提供者来执行请求。如果遇到API错误、速率限制或网络超时等问题，会通过特定于该提供者的异常类（如OpenAIAPIError、AnthropicAPIError）来捕获这些异常，并在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("正在处理新请求")
        print("==========================================")

        # 第一步：分析提示的复杂度
        complexity = self.analyzer.analyze_complexity(userprompt)
        print(f"[第一步] 提示被分类为：{complexity.value.upper()}")

        # 第二步：确定路由目标
        primary_model, fallback_model = self.router.get_models_for_tier(complexity)

        print(f"[第二步] 选定的主提供者是：{primary_model.model_name}")

        # 第三步：使用备用方案执行请求
        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,
        }


# 执行流程测试
if __name__ == "__main__":
    engine = SmartAIEngine()

    # 测试用例1：简单任务
    simple_query = ("日本的首都是什么？请用一个词回答。")

    result_1 = engine.process_request(simple_query)
    print(f"使用的模型：{result_1['model_used']}")
    print(f"响应内容：{result_1['response']}")

    # 测试用例2：复杂任务
    complex_query = ("请编写一个Python函数，用于调试多线程应用程序中的内存泄漏问题。")

    result_2 = engine.process_request(complex_query)
    print(f"使用的模型：{result_2['model_used']}")
    print(f"响应部分内容：{result_2['response'][:100]}...")

解析代码逻辑

• 统一调度系统（SmartAIEngine）：将PromptAnalyzer、ModelRouter和ResilientModelEngine这三个模块化组件作为实例属性进行初始化。

• 处理流程步骤：

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

  • 路由阶段：根据该复杂程度等级来选择对应的模型组合。

  • 执行阶段：以弹性方式调用相应模型，并处理可能出现的故障情况。

• 标准化响应数据：将执行细节封装成结构统一的输出字典，其中会记录模型的使用情况、复杂程度分类以及最终输出结果。

从生产环境中的动态模型切换中获得的经验

构建动态AI路由系统让我们的团队对企业级大语言模型架构有了重要的认识：

首先，要确保分类过程尽可能简单。对于小型任务，切勿使用复杂的大语言模型来进行分类处理，而是应该利用正则表达式、关键词匹配或词元长度规则来完成任务——这样的分类流程应能在5毫秒内完成。

其次，必须对系统输出结果进行标准化处理。不同的模型提供商其输出数据的格式往往不同，因此应用程序在将数据返回给用户界面之前，必须确保这些数据都遵循统一的格式。

第三，要设置合理的超时机制。很多模型提供商的API在出现故障时并不会立即抛出错误，而是会陷入无响应状态。因此，对于主要使用的模型调用，应设置5到10秒的超时时间，这样一旦发生故障，备用方案就能迅速被触发，从而避免给最终用户带来不便。

最后，要密切关注各项使用指标。记录每一次路由决策、模型切换情况以及成本变化数据，这些信息能够帮助你判断所设定的复杂程度阈值是否经过调整后依然有效。

结论

随着AI应用的规模不断扩大，依赖单一的大型语言模型已不再是一种可行的方案。智能模型调度机制能够帮助你在性能、延迟和成本之间取得平衡，同时还能保证响应质量不受影响。

通过将应用程序与特定的模型提供商分离，并引入自动化路由机制、输入评估功能、抽象层设计以及弹性备用方案，你就可以构建出既高效又可靠的AI系统。

在部署自己的应用时，可以将大语言模型提供商视为可动态调整的资源。对于日常处理任务，可以使用轻量级模型；而对于复杂任务，则应优先使用高端模型；同时，还要确保代码中能够顺利实现不同模型提供商之间的切换。

感谢阅读！

希望这篇文章能帮助你真正理解多模型调度系统与动态路由机制在实际应用中的运作方式，并让你知道如何在自己的项目中开始运用这些技术。

如果你对AI工程、智能体AI、大语言模型、RAG、MLOps、企业级AI架构或AI治理等方面感兴趣，欢迎关注我、点赞我的内容并分享给他人哦！

• LinkedIn

• 查看我的Github仓库