How to Build AI Applications That Switch Models Automatically

pip install openai anthropic python-dotenv pydantic

ai-model-router/

│

├── .env

├── README.md

└── app.py

Ini, TOML

OPENAI_API_KEY=your_openai_api_key_here ANTHROPIC_API_KEY=your_anthropic_api_key_here ENVIRONMENT=development

import re
from enum import Enum
from pydantic import BaseModel

class TaskComplexity(Enum):
    SIMPLE = "simple"      # 常见问题解答、简短摘要、基础翻译
    MEDIUM = "medium"      # 标准文本生成、内容改写
    COMPLEX = "complex"    # 代码编写、数学逻辑、结构分析

class PromptAnalyzer:
    def __init__(self):
        # 表示复杂任务的正则表达式模式
        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:
        """
        确定性评估输入文本，输出任务复杂度评级。
        """
        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 = (
        "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}")

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"]

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:
        """
        将提示执行分发到对应的提供方 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"不支持的提供方: {config.provider}"
            )

    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} (Fallback)"
        )

    except Exception as fallback_error:
        raise RuntimeError(
            "主系统和备用系统均失败。 "
            f"错误: {fallback_error}"
        )

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("==========================================")

        # 步骤 1: 分析提示复杂度
        complexity = self.analyzer.analyze_complexity(
            user_prompt
        )
        print(
            f"[步骤 1] 提示被分类为: "
            f"{complexity.value.upper()}"
        )

        # 步骤 2: 确定路由目标
        primary_model, fallback_model = (
            self.router.get_models_for_tier(
                complexity
            )
        )

        print(
            f"[步骤 2] 选定主模型: "
            f"{primary_model.model_name}"
        )

        # 步骤 3: 使用弹性回退机制执行请求
        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"响应片段: "
    f"{result_2['response'][:100]}..."
)