FastAPI開發(fā)AI應(yīng)用：多廠商模型使用指南

作者：wayn 2025-07-09 02:11:00

本文將深入講解如何在 FastAPI AI 聊天應(yīng)用中設(shè)計(jì)和實(shí)現(xiàn)多模型廠商架構(gòu)，讓你的應(yīng)用能夠靈活切換不同的 AI 提供商，提升系統(tǒng)的可靠性和成本效益。即使你是架構(gòu)設(shè)計(jì)新手，也能跟著本教程掌握多廠商集成的核心技術(shù)。

?? 項(xiàng)目地址：https://github.com/wayn111/fastapi-ai-chat-demo

項(xiàng)目概述

想象一下，你的 AI 聊天應(yīng)用不再依賴單一的 AI 提供商，而是能夠智能地在 OpenAI、DeepSeek、通義千問等多個(gè)廠商之間切換。當(dāng)某個(gè)服務(wù)出現(xiàn)問題時(shí)，可以切換到備用提供商；當(dāng)需要降低成本時(shí)，可以選擇性價(jià)比更高的模型。這就是我們要構(gòu)建的多模型廠商架構(gòu)！

圖片

在本章節(jié)中，我們重構(gòu)了前端界面，使其更加美觀，參考 lobechat 界面風(fēng)格。

核心功能

多廠商統(tǒng)一管理：一套代碼支持多個(gè) AI 提供商，無需重復(fù)開發(fā)
智能故障轉(zhuǎn)移：主提供商不可用時(shí)自動(dòng)切換到備用提供商
成本優(yōu)化策略：根據(jù)不同場(chǎng)景選擇最經(jīng)濟(jì)的模型
動(dòng)態(tài)提供商發(fā)現(xiàn)：新增提供商無需修改核心代碼
統(tǒng)一接口設(shè)計(jì)：所有提供商使用相同的調(diào)用方式

技術(shù)棧

核心框架：FastAPI（高性能異步 Web 框架）
AI 集成：OpenAI SDK（統(tǒng)一的 AI 接口標(biāo)準(zhǔn)）
設(shè)計(jì)模式：工廠模式 + 抽象工廠模式
配置管理：環(huán)境變量 + 動(dòng)態(tài)配置
數(shù)據(jù)存儲(chǔ)：Redis（會(huì)話和配置緩存）

支持的 AI 廠商

廠商	代表模型	特色	成本水平
OpenAI	GPT-4, GPT-3.5-turbo	綜合能力強(qiáng)，生態(tài)完善	較高
DeepSeek	deepseek-chat, deepseek-coder	性價(jià)比高，推理能力強(qiáng)	低
通義千問	qwen-turbo, qwen-plus	中文理解優(yōu)秀，阿里生態(tài)	中等

核心架構(gòu)設(shè)計(jì)

設(shè)計(jì)理念

我們的多模型廠商架構(gòu)基于三個(gè)核心設(shè)計(jì)原則：

1. 統(tǒng)一接口原則所有 AI 提供商都遵循相同的接口規(guī)范，就像不同品牌的手機(jī)都使用相同的充電接口一樣。這樣可以確保切換提供商時(shí)不需要修改業(yè)務(wù)代碼。

2. 開放擴(kuò)展原則新增 AI 提供商時(shí)，只需要繼承基類并配置幾個(gè)參數(shù)，系統(tǒng)會(huì)自動(dòng)發(fā)現(xiàn)并集成新的提供商。

3. 故障隔離原則每個(gè)提供商都是獨(dú)立的實(shí)例，一個(gè)提供商的故障不會(huì)影響其他提供商的正常工作。

架構(gòu)層次

圖片

我們的架構(gòu)分為四個(gè)清晰的層次，每一層都有明確的職責(zé)：

1. 抽象接口層（BaseAIProvider）

這是整個(gè)架構(gòu)的"憲法"，定義了所有 AI 提供商必須遵循的接口規(guī)范：

from abc import ABC, abstractmethod
from typing import List, Dict, Any, AsyncGenerator

class BaseAIProvider(ABC):
    """AI提供商抽象基類 - 定義統(tǒng)一接口規(guī)范"""

    def __init__(self, config: Dict[str, Any]):
        """初始化提供商配置"""
        self.config = config
        self.provider_name = self.__class__.__name__.replace('Provider', '').lower()

    @abstractmethod
    asyncdef generate_response(self, messages: List[AIMessage], **kwargs) -> AIResponse:
        """生成AI響應(yīng) - 所有提供商必須實(shí)現(xiàn)"""
        pass

    @abstractmethod
    asyncdef generate_streaming_response(self, messages: List[AIMessage], **kwargs) -> AsyncGenerator[str, None]:
        """生成流式響應(yīng) - 支持實(shí)時(shí)對(duì)話"""
        pass

    @abstractmethod
    def validate_config(self) -> bool:
        """驗(yàn)證配置有效性"""
        pass

為什么這樣設(shè)計(jì)？

強(qiáng)制規(guī)范：所有提供商都必須實(shí)現(xiàn)這些方法
統(tǒng)一流式響應(yīng)：所有廠商使用同一套流式響應(yīng)
安全性：每個(gè)廠商都需要驗(yàn)證配置有效性

2. 兼容適配層（OpenAICompatibleProvider）

這一層是我們架構(gòu)的"翻譯官"，將 OpenAI 的接口標(biāo)準(zhǔn)適配給所有提供商：

class OpenAICompatibleProvider(BaseAIProvider):
    """OpenAI兼容提供商基類 - 統(tǒng)一OpenAI SDK調(diào)用方式"""

    # 子類需要重寫的配置
    DEFAULT_BASE_URL = None
    DEFAULT_MODEL = None
    PROVIDER_NAME = None
    AVAILABLE_MODELS = []

    def __init__(self, config: Dict[str, Any]):
        """初始化OpenAI兼容客戶端"""
        super().__init__(config)
        self.client = None
        self._initialize_client()

    def _initialize_client(self):
        """初始化OpenAI SDK客戶端"""
        try:
            api_key = self.get_config_value('api_key')
            ifnot api_key:
                logger.error(f"{self.get_provider_display_name()}API密鑰為空")
                return

            # 使用OpenAI SDK，但指向不同廠商的API端點(diǎn)
            self.client = OpenAI(
                api_key=api_key,
                base_url=self.get_config_value('base_url', self.DEFAULT_BASE_URL)
            )
            logger.info(f"{self.get_provider_display_name()}客戶端初始化成功")
        except Exception as e:
            logger.error(f"客戶端初始化失敗: {e}")
            self.client = None

核心優(yōu)勢(shì)：

代碼復(fù)用：所有兼容 OpenAI 接口的廠商都可以復(fù)用這套代碼
維護(hù)簡(jiǎn)單：所有廠商共用一套 openai 的 api 接口調(diào)用邏輯
標(biāo)準(zhǔn)統(tǒng)一：所有廠商都使用同一套流式響應(yīng)，使用相同的消息格式和參數(shù)

3. 具體提供商實(shí)現(xiàn)層

這一層是各個(gè) AI 廠商的"身份證"，每個(gè)提供商只需要聲明自己的基本信息，繼承 OpenAICompatibleProvider 即可。

class DeepseekProvider(OpenAICompatibleProvider):
    """DeepSeek提供商實(shí)現(xiàn) - 只需配置基本信息"""

    # 提供商配置 - 這就是全部需要的代碼！
    DEFAULT_BASE_URL = 'https://api.deepseek.com/v1'
    DEFAULT_MODEL = 'deepseek-chat'
    PROVIDER_NAME = 'DeepSeek'
    AVAILABLE_MODELS = [
        'deepseek-chat',      # 通用對(duì)話模型
        'deepseek-coder',     # 代碼專用模型
        'deepseek-math',      # 數(shù)學(xué)推理模型
        'deepseek-reasoner'   # 深度推理模型
    ]

實(shí)現(xiàn)原理：

繼承復(fù)用：繼承 OpenAICompatibleProvider 獲得所有通用功能
配置驅(qū)動(dòng)：只需要配置幾個(gè)類變量就完成了集成
自動(dòng)發(fā)現(xiàn)：系統(tǒng)會(huì)自動(dòng)掃描并注冊(cè)這個(gè)提供商

4. 工廠管理層（AIProviderFactory）

這是整個(gè)架構(gòu)的"大腦"，負(fù)責(zé)動(dòng)態(tài)發(fā)現(xiàn)、創(chuàng)建和管理所有提供商：

class AIProviderFactory:
    """AI提供商工廠 - 統(tǒng)一管理所有提供商"""

    # 提供商實(shí)例緩存
    _instances: Dict[str, BaseAIProvider] = {}
    # 動(dòng)態(tài)發(fā)現(xiàn)的提供商類緩存
    _discovered_providers: Optional[Dict[str, Type[BaseAIProvider]]] = None

    @classmethod
    def _discover_providers(cls) -> Dict[str, Type[BaseAIProvider]]:
        """動(dòng)態(tài)發(fā)現(xiàn)所有提供商類 - 這是魔法發(fā)生的地方"""
        if cls._discovered_providers isnotNone:
            return cls._discovered_providers

        providers = {}

        # 掃描 ai_providers 包中的所有模塊
        import ai_providers
        package_path = ai_providers.__path__

        for importer, modname, ispkg in pkgutil.iter_modules(package_path):
            if modname.endswith('_provider') and modname != 'openai_compatible_provider':
                try:
                    # 動(dòng)態(tài)導(dǎo)入模塊
                    module = importlib.import_module(f'ai_providers.{modname}')

                    # 查找繼承自O(shè)penAICompatibleProvider的類
                    for name, obj in inspect.getmembers(module, inspect.isclass):
                        if (issubclass(obj, OpenAICompatibleProvider) and
                            obj != OpenAICompatibleProvider and
                            hasattr(obj, 'PROVIDER_NAME')):

                            provider_key = name.lower().replace('provider', '')
                            providers[provider_key] = obj
                            logger.debug(f"發(fā)現(xiàn)提供商: {provider_key} -> {name}")

                except Exception as e:
                    logger.warning(f"導(dǎo)入模塊 {modname} 時(shí)出錯(cuò): {e}")

        cls._discovered_providers = providers
        logger.info(f"動(dòng)態(tài)發(fā)現(xiàn) {len(providers)} 個(gè)提供商: {list(providers.keys())}")
        return providers

工廠模式的威力：

動(dòng)態(tài)發(fā)現(xiàn)：自動(dòng)掃描并注冊(cè)新的提供商，無需手動(dòng)配置
實(shí)例緩存：相同配置的提供商實(shí)例會(huì)被緩存，提高性能
統(tǒng)一創(chuàng)建：所有提供商都通過工廠創(chuàng)建，確保一致性

多提供商管理器

在工廠的基礎(chǔ)上，我們還提供了多提供商管理器，讓你可以同時(shí)管理多個(gè)提供商：

class MultiProviderManager:
    """多提供商管理器 - 統(tǒng)一管理多個(gè)AI提供商實(shí)例"""

    def __init__(self, configs: Dict[str, Dict[str, Any]]):
        """初始化多提供商管理器

        Args:
            configs: 多個(gè)提供商的配置，格式：{provider_name: config}
        """
        self.providers: Dict[str, BaseAIProvider] = {}
        self.default_provider = None

        # 初始化所有配置的提供商
        for provider_name, config in configs.items():
            try:
                provider = AIProviderFactory.create_provider(provider_name, config)
                if provider.validate_config():
                    self.providers[provider_name] = provider
                    logger.info(f"提供商 {provider_name} 初始化成功")

                    # 設(shè)置默認(rèn)提供商
                    if self.default_provider isNone:
                        self.default_provider = provider_name
                else:
                    logger.warning(f"提供商 {provider_name} 配置驗(yàn)證失敗")
            except Exception as e:
                logger.error(f"提供商 {provider_name} 初始化失敗: {e}")

    def get_available_providers(self) -> List[str]:
        """獲取所有可用的提供商列表"""
        return list(self.providers.keys())

環(huán)境變量配置

在項(xiàng)目根目錄創(chuàng)建 .env 文件：

REDIS_HOST="127.0.0.1"
REDIS_PORT=6379
REDIS_PASSWORD=""

# ===========================================
# AI 提供商配置（至少配置一個(gè)）
# ===========================================

# OpenAI 配置
OPENAI_API_KEY=sk-your-openai-api-key
OPENAI_BASE_URL=https://api.openai.com/v1  # 可選，默認(rèn)官方地址
OPENAI_MODEL=gpt-4o  # 可選，默認(rèn)模型

# DeepSeek 配置
DEEPSEEK_API_KEY=sk-your-deepseek-api-key
DEEPSEEK_MODEL=deepseek-chat  # 可選

# 通義千問配置
QIANWEN_API_KEY=sk-your-qianwen-api-key
QIANWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

# ===========================================
# 應(yīng)用配置
# ===========================================
DEFAULT_AI_PROVIDER=deepseek  # 默認(rèn)使用的提供商
MAX_TOKENS=1000               # 最大生成長(zhǎng)度
TEMPERATURE=0.7              # 創(chuàng)造性參數(shù)

配置類設(shè)計(jì)

我們使用配置類來統(tǒng)一管理所有配置項(xiàng)：

class Config:
    """應(yīng)用配置管理類"""

    # 應(yīng)用基礎(chǔ)配置
    APP_NAME = "FastAPI AI Chat Demo"
    # ...

    # 新增AI提供商基礎(chǔ)信息
    _AI_PROVIDERS_INFO = {
        'openai': {
            'base_url': 'https://api.openai.com/v1',
            'model': 'gpt-4o'
        },
        'deepseek': {
            'base_url': 'https://api.deepseek.com/v1',
            'model': 'deepseek-chat'
        },
        'qianwen': {
            'base_url': 'https://dashscope.aliyuncs.com/compatible-mode/v1',
            'model': 'qwen-turbo'
        }
    }

    # AI提供商配置 - 動(dòng)態(tài)生成
    @classmethod
    def _get_ai_providers_config(cls) -> dict:
        """獲取所有AI提供商配置"""
        return {provider: cls._build_provider_config(provider) for provider in cls._AI_PROVIDERS_INFO.keys()}

    # 延遲初始化AI提供商配置
    @property
    def AI_PROVIDERS_CONFIG(self) -> dict:
        ifnot hasattr(self, '_ai_providers_config'):
            self._ai_providers_config = self._get_ai_providers_config()
        return self._ai_providers_config

    @classmethod
    def get_all_ai_configs(cls) -> dict:
        """獲取所有已配置API Key的AI提供商配置"""
        configs = cls._get_ai_providers_config()
        return {name: config for name, config in configs.items() if config.get('api_key')}

在 Config 配置中新增 AI 提供商核心配置。

核心 API 接口

1. 獲取可用提供商列表

@app.get("/providers")
asyncdef get_providers():
    """獲取可用的AI提供商列表"""
    logger.info("獲取AI提供商列表")
    try:
        configured_providers = Config.get_configured_providers()
        all_models = ai_manager.get_all_available_models()

        providers_info = []
        for provider in configured_providers:
            provider_obj = ai_manager.get_provider(provider)
            if provider_obj:
                providers_info.append({
                    "id": provider,
                    "name": provider_obj.get_provider_name(),
                    "models": provider_obj.get_available_models(),
                    "is_default": provider == Config.DEFAULT_AI_PROVIDER
                })

        return {
            "providers": providers_info,
            "default_provider": Config.DEFAULT_AI_PROVIDER,
            "all_models": all_models
        }
    except Exception as e:
        logger.error(f"獲取AI提供商列表失敗: {e}")
        raise HTTPException(status_code=500, detail=f"獲取提供商列表失敗: {str(e)}")

/providers 接口會(huì)返回所有可用 AI 提供商，以及對(duì)應(yīng)模型。

2. 流式聊天接口（支持提供商選擇）

@app.get("/chat/stream")
asyncdef chat_stream(
    user_id: str = Query(..., descriptinotallow="用戶ID"),
    session_id: str = Query(..., descriptinotallow="會(huì)話ID"),
    message: str = Query(..., descriptinotallow="用戶消息"),
    role: str = Query("assistant", descriptinotallow="AI角色"),
    provider: Optional[str] = Query(None, descriptinotallow="AI提供商"),
    model: Optional[str] = Query(None, descriptinotallow="AI模型")
):
    """流式聊天接口"""
    logger.info(f"流式聊天請(qǐng)求 - 用戶: {user_id}, 會(huì)話: {session_id[:8]}..., 角色: {role}, 消息長(zhǎng)度: {len(message)}, 提供商: {provider}")

    if role notin AI_ROLES:
        logger.warning(f"不支持的AI角色: {role}")
        raise HTTPException(status_code=400, detail="不支持的AI角色")

    return StreamingResponse(
        generate_streaming_response(user_id, session_id, message, role, provider, model),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "Access-Control-Allow-Origin": "*"
        }
    )

流式聊天接口參數(shù)中新增 AI 提供商以及 AI 模型參數(shù)。

擴(kuò)展新提供商實(shí)現(xiàn)

快速添加新提供商

對(duì)于支持 OpenAI API 格式的提供商，只需幾行代碼即可集成：

這里用 moonshot 作為新廠商接入，在 ai_providers 目錄下新增 moonshot_provider.py 文件，

# ai_providers/moonshot_provider.py
from ai_providers.openai_compatible_provider import OpenAICompatibleProvider

class MoonshotProvider(OpenAICompatibleProvider):
    """月之暗面 Kimi 提供商"""
    DEFAULT_BASE_URL = "https://api.moonshot.cn/v1"
    DEFAULT_MODEL = "moonshot-v1-8k"
    PROVIDER_NAME = "moonshot"
    AVAILABLE_MODELS = [
        "moonshot-v1-8k",
        "moonshot-v1-32k",
        "moonshot-v1-128k"
    ]

環(huán)境變量配置

在 .env 文件中新增 moonshot 配置

# Moonshot (月之暗面)
MOONSHOT_API_KEY=sk-your-moonshot-api-key
MOONSHOT_BASE_URL=https://api.moonshot.cn/v1  # 可選
MOONSHOT_MODEL=moonshot-v1-8k  # 可選

配置類更新

config.py 中添加 moonshot 配置，

class Config:
    # ... 現(xiàn)有配置 ...

    # AI提供商基礎(chǔ)信息
    _AI_PROVIDERS_INFO = {
        'openai': {
            'base_url': 'https://api.openai.com/v1',
            'model': 'gpt-4o'
        },
        'deepseek': {
            'base_url': 'https://api.deepseek.com/v1',
            'model': 'deepseek-chat'
        },
        'qianwen': {
            'base_url': 'https://dashscope.aliyuncs.com/compatible-mode/v1',
            'model': 'qwen-turbo'
        },
        'moonshot': {
            'base_url': 'https://api.moonshot.cn/v1',
            'model': 'moonshot-v1-8k'
        }
    }

通過以上步驟，您可以輕松擴(kuò)展支持任何新的 AI 提供商，系統(tǒng)會(huì)自動(dòng)發(fā)現(xiàn)并集成新的提供商，無需修改核心代碼。

總結(jié)

本文詳細(xì)介紹了如何在 FastAPI 應(yīng)用中構(gòu)建一個(gè)靈活、可擴(kuò)展的多模型廠商架構(gòu)。通過抽象接口層、兼容適配層、具體實(shí)現(xiàn)層和工廠管理層的四層架構(gòu)設(shè)計(jì)，實(shí)現(xiàn)了所有 AI 提供商的統(tǒng)一接口調(diào)用，支持 OpenAI、DeepSeek、通義千問等多個(gè)廠商的無縫集成。核心采用工廠模式和抽象工廠模式，配合動(dòng)態(tài)提供商發(fā)現(xiàn)機(jī)制，新增廠商只需幾行代碼即可完成集成。

最后覺得本文寫的不錯(cuò)的話，可以關(guān)注我，我會(huì)繼續(xù)更新 FastAPI 框架開發(fā) AI 聊天應(yīng)用代碼。

責(zé)任編輯：武曉燕來源：程序員wayn